feat: cluster based deployment, rollback, restart across services

api/internal/features/deploy/docker/cluster.go

@@ -0,0 +1,188 @@
+package docker
+
+import (
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/events"
+	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/api/types/network"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/volume"
+	"github.com/raghavyuva/nixopus-api/internal/config"
+)
+
+func (s *DockerService) InitCluster() error {
+	config := config.AppConfig
+
+	// Use localhost as default advertise address if SSH host is not configured
+	// Useful during development
+	advertiseAddr := "127.0.0.1:2377"
+	if config.SSH.Host != "" {
+		advertiseAddr = config.SSH.Host + ":2377"
+	}
+
+	_, err := s.Cli.SwarmInit(s.Ctx, swarm.InitRequest{
+		ListenAddr: "0.0.0.0:2377",
+		// Address that Hosts can use to reach the master node
+		AdvertiseAddr: advertiseAddr,
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *DockerService) JoinCluster() error {
+	config := config.AppConfig
+
+	// Use localhost as default advertise address if SSH host is not configured
+	advertiseAddr := "127.0.0.1:2377"
+	if config.SSH.Host != "" {
+		advertiseAddr = config.SSH.Host + ":2377"
+	}
+
+	err := s.Cli.SwarmJoin(s.Ctx, swarm.JoinRequest{
+		ListenAddr:    "0.0.0.0:2377",
+		AdvertiseAddr: advertiseAddr,
+	})
+	return err
+}
+
+func (s *DockerService) LeaveCluster(force bool) error {
+	err := s.Cli.SwarmLeave(s.Ctx, force)
+	return err
+}
+
+func (s *DockerService) GetClusterInfo() (swarm.ClusterInfo, error) {
+	clusterInfo, err := s.Cli.SwarmInspect(s.Ctx)
+	return clusterInfo.ClusterInfo, err
+}
+
+func (s *DockerService) GetClusterNodes() ([]swarm.Node, error) {
+	nodes, err := s.Cli.NodeList(s.Ctx, types.NodeListOptions{})
+	return nodes, err
+}
+
+func (s *DockerService) GetClusterServices() ([]swarm.Service, error) {
+	services, err := s.Cli.ServiceList(s.Ctx, types.ServiceListOptions{})
+	return services, err
+}
+
+func (s *DockerService) GetClusterTasks() ([]swarm.Task, error) {
+	tasks, err := s.Cli.TaskList(s.Ctx, types.TaskListOptions{})
+	return tasks, err
+}
+
+func (s *DockerService) GetClusterSecrets() ([]swarm.Secret, error) {
+	secrets, err := s.Cli.SecretList(s.Ctx, types.SecretListOptions{})
+	return secrets, err
+}
+
+func (s *DockerService) GetClusterConfigs() ([]swarm.Config, error) {
+	configs, err := s.Cli.ConfigList(s.Ctx, types.ConfigListOptions{})
+	return configs, err
+}
+
+func (s *DockerService) GetClusterVolumes() ([]*volume.Volume, error) {
+	volumes, err := s.Cli.VolumeList(s.Ctx, volume.ListOptions{})
+	return volumes.Volumes, err
+}
+
+func (s *DockerService) GetClusterNetworks() ([]network.Summary, error) {
+	networks, err := s.Cli.NetworkList(s.Ctx, network.ListOptions{})
+	return networks, err
+}
+
+func (s *DockerService) UpdateNodeAvailability(nodeID string, availability swarm.NodeAvailability) error {
+	node, _, err := s.Cli.NodeInspectWithRaw(s.Ctx, nodeID)
+	if err != nil {
+		return err
+	}
+	spec := node.Spec
+	spec.Availability = availability
+	return s.Cli.NodeUpdate(s.Ctx, nodeID, node.Version, spec)
+}
+
+func (s *DockerService) ListenEvents(opts events.ListOptions) (<-chan events.Message, <-chan error) {
+	return s.Cli.Events(s.Ctx, opts)
+}
+
+func (s *DockerService) ScaleService(serviceID string, replicas uint64, rollback string) error {
+	svc, _, err := s.Cli.ServiceInspectWithRaw(s.Ctx, serviceID, types.ServiceInspectOptions{})
+	if err != nil {
+		return err
+	}
+	spec := svc.Spec
+	spec.Mode.Replicated.Replicas = &replicas
+	_, err = s.Cli.ServiceUpdate(s.Ctx, serviceID, svc.Version, spec, types.ServiceUpdateOptions{
+		Rollback: rollback,
+	})
+	return err
+}
+
+func (s *DockerService) GetServiceHealth(service swarm.Service) (int, int, error) {
+	tasks, err := s.Cli.TaskList(s.Ctx, types.TaskListOptions{
+		Filters: filters.NewArgs(
+			filters.Arg("service", service.ID),
+		),
+	})
+	if err != nil {
+		return 0, 0, err
+	}
+
+	running := 0
+	for _, t := range tasks {
+		if t.Status.State == swarm.TaskStateRunning {
+			running++
+		}
+	}
+	desired := 0
+	if service.Spec.Mode.Replicated != nil && service.Spec.Mode.Replicated.Replicas != nil {
+		desired = int(*service.Spec.Mode.Replicated.Replicas)
+	}
+	return running, desired, nil
+}
+
+func (s *DockerService) GetTaskHealth(task swarm.Task) swarm.TaskState {
+	if task.Status.State != "" {
+		return task.Status.State
+	}
+	return swarm.TaskState("")
+}
+
+func (s *DockerService) CreateService(service swarm.Service) error {
+	_, err := s.Cli.ServiceCreate(s.Ctx, service.Spec, types.ServiceCreateOptions{})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *DockerService) UpdateService(serviceID string, serviceSpec swarm.ServiceSpec, rollback string) error {
+	svc, _, err := s.Cli.ServiceInspectWithRaw(s.Ctx, serviceID, types.ServiceInspectOptions{})
+	if err != nil {
+		return err
+	}
+
+	_, err = s.Cli.ServiceUpdate(s.Ctx, serviceID, svc.Version, serviceSpec, types.ServiceUpdateOptions{
+		Rollback: rollback,
+	})
+	return err
+}
+
+func (s *DockerService) DeleteService(serviceID string) error {
+	err := s.Cli.ServiceRemove(s.Ctx, serviceID)
+	return err
+}
+
+func (s *DockerService) RollbackService(serviceID string) error {
+	_, err := s.Cli.ServiceUpdate(s.Ctx, serviceID, swarm.Version{}, swarm.ServiceSpec{}, types.ServiceUpdateOptions{
+		Rollback: "previous",
+	})
+	return err
+}
+
+func (s *DockerService) GetServiceByID(serviceID string) (swarm.Service, error) {
+	service, _, err := s.Cli.ServiceInspectWithRaw(s.Ctx, serviceID, types.ServiceInspectOptions{})
+	return service, err
+}

api/internal/features/deploy/docker/init.go

@@ -7,9 +7,12 @@ import (
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/api/types/image"
 	"github.com/docker/docker/api/types/network"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/volume"
 	"github.com/docker/docker/client"
 	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/raghavyuva/nixopus-api/internal/features/logger"
@@ -23,9 +26,9 @@ type DockerService struct {
 }
 
 type DockerRepository interface {
-    ListAllContainers() ([]container.Summary, error)
-    ListContainers(opts container.ListOptions) ([]container.Summary, error)
-    ListAllImages(opts image.ListOptions) []image.Summary
+	ListAllContainers() ([]container.Summary, error)
+	ListContainers(opts container.ListOptions) ([]container.Summary, error)
+	ListAllImages(opts image.ListOptions) []image.Summary
 
 	StopContainer(containerID string, opts container.StopOptions) error
 	RemoveContainer(containerID string, opts container.RemoveOptions) error
@@ -46,6 +49,28 @@ type DockerRepository interface {
 	RemoveImage(imageName string, opts image.RemoveOptions) error
 	PruneBuildCache(opts types.BuildCachePruneOptions) error
 	PruneImages(opts filters.Args) (image.PruneReport, error)
+
+	InitCluster() error
+	JoinCluster() error
+	LeaveCluster(force bool) error
+	GetClusterInfo() (swarm.ClusterInfo, error)
+	GetClusterNodes() ([]swarm.Node, error)
+	GetClusterServices() ([]swarm.Service, error)
+	GetClusterTasks() ([]swarm.Task, error)
+	GetClusterSecrets() ([]swarm.Secret, error)
+	GetClusterConfigs() ([]swarm.Config, error)
+	GetClusterVolumes() ([]*volume.Volume, error)
+	GetClusterNetworks() ([]network.Summary, error)
+	UpdateNodeAvailability(nodeID string, availability swarm.NodeAvailability) error
+	ScaleService(serviceID string, replicas uint64, rollback string) error
+	ListenEvents(opts events.ListOptions) (<-chan events.Message, <-chan error)
+	GetServiceHealth(service swarm.Service) (int, int, error)
+	GetTaskHealth(task swarm.Task) swarm.TaskState
+	CreateService(service swarm.Service) error
+	UpdateService(serviceID string, serviceSpec swarm.ServiceSpec, rollback string) error
+	DeleteService(serviceID string) error
+	RollbackService(serviceID string) error
+	GetServiceByID(serviceID string) (swarm.Service, error)
 }
 
 type DockerClient struct {
@@ -54,11 +79,35 @@ type DockerClient struct {
 
 // NewDockerService creates a new instance of DockerService using the default docker client.
 func NewDockerService() *DockerService {
-	return &DockerService{
-		Cli:    NewDockerClient(),
+	client := NewDockerClient()
+	service := &DockerService{
+		Cli:    client,
 		Ctx:    context.Background(),
 		logger: logger.NewLogger(),
 	}
+
+	// Initialize cluster if not already initialized, this should be run on master node only
+	// TODO: Add a check to see if the node is the master node
+	// WARNING: This should be thought again during multi-server architecture feature
+	if !isClusterInitialized(client) {
+		if err := service.InitCluster(); err != nil {
+			service.logger.Log(logger.Warning, "Failed to initialize cluster", err.Error())
+		} else {
+			service.logger.Log(logger.Info, "Cluster initialized successfully", "")
+		}
+	} else {
+		service.logger.Log(logger.Info, "Cluster already initialized", "")
+	}
+
+	return service
+}
+
+func isClusterInitialized(cli *client.Client) bool {
+	info, err := cli.Info(context.Background())
+	if err != nil {
+		return false
+	}
+	return info.Swarm.LocalNodeState == swarm.LocalNodeStateActive
 }
 
 func NewDockerServiceWithClient(cli *client.Client, ctx context.Context, logger logger.Logger) *DockerService {
@@ -101,24 +150,24 @@ func NewDockerClient() *client.Client {
 //
 // If an error occurs while listing the containers, it returns the error (no panic).
 func (s *DockerService) ListAllContainers() ([]container.Summary, error) {
-    containers, err := s.Cli.ContainerList(s.Ctx, container.ListOptions{
-        All: true,
-    })
-    if err != nil {
-        return nil, err
-    }
-
-    return containers, nil
+	containers, err := s.Cli.ContainerList(s.Ctx, container.ListOptions{
+		All: true,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return containers, nil
 }
 
 // ListContainers returns containers using the provided docker list options
 // (including native filters like name/status/ancestor and optional limits).
 func (s *DockerService) ListContainers(opts container.ListOptions) ([]container.Summary, error) {
-    containers, err := s.Cli.ContainerList(s.Ctx, opts)
-    if err != nil {
-        return nil, err
-    }
-    return containers, nil
+	containers, err := s.Cli.ContainerList(s.Ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	return containers, nil
 }
 
 // StopContainer stops the container with the given ID. If the container does not exist,

api/internal/features/deploy/tasks/delete.go

@@ -5,15 +5,14 @@ import (
 	"os"
 	"path/filepath"
 
-	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/image"
 	"github.com/google/uuid"
 	"github.com/raghavyuva/nixopus-api/internal/features/deploy/types"
 	"github.com/raghavyuva/nixopus-api/internal/features/logger"
 )
 
 // DeleteDeployment deletes a deployment and its associated resources.
-// It stops and removes the container, image, and repository.
+// It stops and removes the service, image, and repository.
 // It returns an error if any operation fails.
 func (s *TaskService) DeleteDeployment(deployment *types.DeleteDeploymentRequest, userID uuid.UUID, organizationID uuid.UUID) error {
 	application, err := s.Storage.GetApplicationById(deployment.ID.String(), organizationID)
@@ -23,23 +22,28 @@ func (s *TaskService) DeleteDeployment(deployment *types.DeleteDeploymentRequest
 
 	domain := application.Domain
 
-	deployments, err := s.Storage.GetApplicationDeployments(application.ID)
+	services, err := s.DockerRepo.GetClusterServices()
 	if err != nil {
-		s.Logger.Log(logger.Error, "Failed to get application deployments", err.Error())
+		s.Logger.Log(logger.Error, "Failed to get services", err.Error())
 	} else {
-		for _, dep := range deployments {
-			if dep.ContainerID != "" {
-				s.Logger.Log(logger.Info, "Stopping container", dep.ContainerID)
-				if err := s.DockerRepo.StopContainer(dep.ContainerID, container.StopOptions{}); err != nil {
-					s.Logger.Log(logger.Error, "Failed to stop container", err.Error())
-				}
-
-				s.Logger.Log(logger.Info, "Removing container", dep.ContainerID)
-				if err := s.DockerRepo.RemoveContainer(dep.ContainerID, container.RemoveOptions{Force: true}); err != nil {
-					s.Logger.Log(logger.Error, "Failed to remove container", err.Error())
+		for _, service := range services {
+			if service.Spec.Annotations.Name == application.Name {
+				s.Logger.Log(logger.Info, "Deleting service", service.ID)
+				if err := s.DockerRepo.DeleteService(service.ID); err != nil {
+					s.Logger.Log(logger.Error, "Failed to delete service", err.Error())
+				} else {
+					s.Logger.Log(logger.Info, "Service deleted successfully", service.ID)
 				}
+				break
 			}
+		}
+	}
 
+	deployments, err := s.Storage.GetApplicationDeployments(application.ID)
+	if err != nil {
+		s.Logger.Log(logger.Error, "Failed to get application deployments", err.Error())
+	} else {
+		for _, dep := range deployments {
 			if dep.ContainerImage != "" {
 				s.Logger.Log(logger.Info, "Removing image", dep.ContainerImage)
 				if err := s.DockerRepo.RemoveImage(dep.ContainerImage, image.RemoveOptions{Force: true}); err != nil {