delete removed Docker containers from store, sync only uncloud labeled

internal/cli/client/service.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/distribution/reference"
-	"github.com/docker/docker/api/types/container"
+	dockercontainer "github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/network"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
@@ -13,14 +13,10 @@ import (
 	"strings"
 	"uncloud/internal/machine/api/pb"
 	"uncloud/internal/machine/docker"
+	"uncloud/internal/machine/docker/container"
 	"uncloud/internal/secret"
 )
 
-const (
-	LabelServiceID   = "uncloud.service.id"
-	LabelServiceName = "uncloud.service.name"
-)
-
 // ServiceOptions contains all the options for creating a service.
 type ServiceOptions struct {
 	Image   string
@@ -57,6 +53,8 @@ func (c *Client) RunService(ctx context.Context, opts *ServiceOptions) (RunServi
 				machine = m
 				break
 			}
+		}
+		if machine == nil {
 			return resp, fmt.Errorf("machine %q not found", opts.Machine)
 		}
 	} else {
@@ -102,11 +100,11 @@ func (c *Client) RunService(ctx context.Context, opts *ServiceOptions) (RunServi
 	}
 	containerName := fmt.Sprintf("%s-%s", serviceName, suffix)
 
-	config := &container.Config{
+	config := &dockercontainer.Config{
 		Image: opts.Image,
 		Labels: map[string]string{
-			LabelServiceID:   serviceID,
-			LabelServiceName: serviceName,
+			container.LabelServiceID:   serviceID,
+			container.LabelServiceName: serviceName,
 		},
 	}
 	netConfig := &network.NetworkingConfig{
@@ -119,7 +117,7 @@ func (c *Client) RunService(ctx context.Context, opts *ServiceOptions) (RunServi
 	if err != nil {
 		return resp, fmt.Errorf("create container: %w", err)
 	}
-	if err = c.StartContainer(ctx, createResp.ID, container.StartOptions{}); err != nil {
+	if err = c.StartContainer(ctx, createResp.ID, dockercontainer.StartOptions{}); err != nil {
 		return resp, fmt.Errorf("start container: %w", err)
 	}
 

internal/machine/docker/container/container.go

@@ -2,6 +2,11 @@ package container
 
 import "github.com/docker/docker/api/types"
 
+const (
+	LabelServiceID   = "uncloud.service.id"
+	LabelServiceName = "uncloud.service.name"
+)
+
 type Container struct {
 	types.Container
 }

internal/machine/docker/manager.go

@@ -100,14 +100,16 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
 		select {
 		case e := <-eventCh:
 			switch e.Action {
-			// Actions that may trigger a container state change.
-			case events.ActionStart,
+			// Actions that may trigger a container state change or creation/deletion of a container.
+			case events.ActionCreate,
+				events.ActionStart,
 				events.ActionStop,
 				events.ActionPause,
 				events.ActionUnPause,
 				events.ActionKill,
 				events.ActionDie,
 				events.ActionOOM,
+				events.ActionDestroy,
 				events.ActionHealthStatusHealthy,
 				events.ActionHealthStatusUnhealthy:
 
@@ -143,19 +145,50 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
 }
 
 func (m *Manager) syncContainersToStore(ctx context.Context) error {
-	containers, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{})
+	storeContainers, err := m.store.ListContainers(ctx, store.ListOptions{MachineIDs: []string{m.machineID}})
+	if err != nil {
+		return fmt.Errorf("list containers from store: %w", err)
+	}
+	// List only Uncloud service containers identified by their labels.
+	containers, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{
+		Filters: filters.NewArgs(
+			filters.Arg("label", container.LabelServiceID),
+			filters.Arg("label", container.LabelServiceName),
+		),
+	})
 	if err != nil {
 		// TODO: mark all containers as outdated in the store.
-		return fmt.Errorf("list containers: %w", err)
+		return fmt.Errorf("list Docker containers: %w", err)
 	}
-	for _, dc := range containers {
-		c := &container.Container{
-			Container: dc,
+
+	// Delete containers that are not present in the Docker daemon from the store.
+	var deleteIDs []string
+	for _, sc := range storeContainers {
+		found := false
+		for i, _ := range containers {
+			if containers[i].ID == sc.Container.ID {
+				found = true
+				break
+			}
 		}
-		if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil {
-			return fmt.Errorf("create or update container: %w", err)
+		if !found {
+			deleteIDs = append(deleteIDs, sc.Container.ID)
 		}
 	}
 
-	return nil
+	var storeErr error
+	if len(deleteIDs) > 0 {
+		if err = m.store.DeleteContainers(ctx, store.DeleteOptions{IDs: deleteIDs}); err != nil {
+			storeErr = fmt.Errorf("delete containers from store: %w", err)
+		}
+	}
+
+	// Create or update the current Docker containers in the store.
+	for _, dc := range containers {
+		c := &container.Container{Container: dc}
+		if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil {
+			storeErr = errors.Join(storeErr, fmt.Errorf("create or update container %q: %w", c.ID, err))
+		}
+	}
+	return storeErr
 }