diff --git a/cmd/uncloud/machine/rm.go b/cmd/uncloud/machine/rm.go index 3e749a85..181929ed 100644 --- a/cmd/uncloud/machine/rm.go +++ b/cmd/uncloud/machine/rm.go @@ -151,20 +151,22 @@ func remove(ctx context.Context, uncli *cli.CLI, nameOrID string, opts removeOpt fmt.Println() } - if _, err = client.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: m.Id}); err != nil { - return fmt.Errorf("remove machine from cluster: %w", err) - } - fmt.Printf("Machine '%s' removed from the cluster.\n", m.Name) - + // Initiate reset before removing the machine from the cluster to stop it from updating the cluster store. + // This is still optimistic as Reset only triggers the reset process that runs asynchronously. if reset && reachable { _, err = client.MachineClient.Reset(rmCtx, &pb.ResetRequest{}) if err != nil { - fmt.Printf("WARNING: Failed to reset machine: %v\n", err) + tui.PrintWarning(fmt.Sprintf("Failed to reset machine: %v\n", err)) } else { fmt.Println("Machine reset initiated and will complete in the background.") } } + if _, err = client.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: m.Id}); err != nil { + return fmt.Errorf("remove machine from cluster: %w", err) + } + fmt.Printf("Machine '%s' removed from the cluster.\n", m.Name) + // Remove the connection to the machine from the uncloud config if it exists. if uncli.Config != nil { contextName := uncli.ContextOverrideOrCurrent() diff --git a/internal/machine/cluster/cluster.go b/internal/machine/cluster/cluster.go index d3c2b87c..f7858121 100644 --- a/internal/machine/cluster/cluster.go +++ b/internal/machine/cluster/cluster.go @@ -334,6 +334,13 @@ func (c *Cluster) RemoveMachine(ctx context.Context, req *pb.RemoveMachineReques return nil, status.Error(codes.InvalidArgument, "machine ID not set") } + // Cleanup machine containers from the store that could be left if the machine is unavailable + // removed with --no-reset, or didn't have time to finish propagating changes before resetting. + if err := c.store.DeleteContainers(ctx, store.DeleteOptions{MachineIDs: []string{req.Id}}); err != nil { + slog.Error("Failed to delete container records from the cluster store for the machine being removed.", + "id", req.Id, "err", err) + } + if err := c.store.DeleteMachine(ctx, req.Id); err != nil { if errors.Is(err, store.ErrMachineNotFound) { return nil, status.Errorf(codes.NotFound, "machine not found: %s", req.Id) diff --git a/internal/machine/store/container.go b/internal/machine/store/container.go index 024998da..e32e2433 100644 --- a/internal/machine/store/container.go +++ b/internal/machine/store/container.go @@ -46,7 +46,10 @@ type ServiceIDOrNameOptions struct { } type DeleteOptions struct { + // IDs filters containers by their container IDs. IDs []string + // MachineIDs filters containers by the machine IDs they are running on. + MachineIDs []string } // CreateOrUpdateContainer creates a new container record or updates an existing one in the store database. @@ -100,21 +103,24 @@ func normaliseContainerForStore(ctr *api.ServiceContainer) { } // ListContainers returns a list of container records from the store database that match the given options. +// The result excludes orphan containers whose machine is no longer in the cluster. func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]ContainerRecord, error) { - q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers"). - Where(sq.Eq{"sync_status": SyncStatusSynced}) + q := sq.Select("c.id", "c.container", "c.machine_id", "c.sync_status", "c.updated_at"). + From("containers c"). + Join("machines m ON m.id = c.machine_id"). + Where(sq.Eq{"c.sync_status": SyncStatusSynced}) if len(opts.MachineIDs) > 0 { - q = q.Where(sq.Eq{"machine_id": opts.MachineIDs}) + q = q.Where(sq.Eq{"c.machine_id": opts.MachineIDs}) } if opts.ServiceIDOrName.ID != "" || opts.ServiceIDOrName.Name != "" { var conditions []sq.Sqlizer if opts.ServiceIDOrName.ID != "" { - conditions = append(conditions, sq.Eq{"service_id": opts.ServiceIDOrName.ID}) + conditions = append(conditions, sq.Eq{"c.service_id": opts.ServiceIDOrName.ID}) } if opts.ServiceIDOrName.Name != "" { - conditions = append(conditions, sq.Eq{"service_name": opts.ServiceIDOrName.Name}) + conditions = append(conditions, sq.Eq{"c.service_name": opts.ServiceIDOrName.Name}) } q = q.Where(sq.Or(conditions)) } @@ -172,16 +178,19 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain } // DeleteContainers deletes container records from the store database that match the given options. +// If no filter is set, all container records are deleted. Filters are combined with AND. func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error { - query := "DELETE FROM containers" - var args []any - + q := sq.Delete("containers") if len(opts.IDs) > 0 { - query += " WHERE id IN (?" + strings.Repeat(", ?", len(opts.IDs)-1) + ")" - args = make([]any, len(opts.IDs)) - for i, id := range opts.IDs { - args[i] = id - } + q = q.Where(sq.Eq{"id": opts.IDs}) + } + if len(opts.MachineIDs) > 0 { + q = q.Where(sq.Eq{"machine_id": opts.MachineIDs}) + } + + query, args, err := q.ToSql() + if err != nil { + return fmt.Errorf("build query: %w", err) } res, err := s.corro.ExecContext(ctx, query, args...) @@ -189,7 +198,8 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error return fmt.Errorf("delete query: %w", err) } if res.RowsAffected > 0 { - slog.Debug("Container records deleted from store DB.", "ids", opts.IDs, "count", res.RowsAffected) + slog.Debug("Container records deleted from store DB.", + "ids", opts.IDs, "machine_ids", opts.MachineIDs, "count", res.RowsAffected) } return nil @@ -197,10 +207,13 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error // SubscribeContainers returns a list of containers and a channel that signals changes to the list. The channel doesn't // receive any values, it just signals when a container(s) has been added, updated, or deleted in the database. +// The result excludes orphan containers whose machine is no longer in the cluster. func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-chan struct{}, error) { // TODO: figure out whether we need sync_status at all (not used at the moment). - q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers"). - Where(sq.Eq{"sync_status": SyncStatusSynced}) + q := sq.Select("c.id", "c.container", "c.machine_id", "c.sync_status", "c.updated_at"). + From("containers c"). + Join("machines m ON m.id = c.machine_id"). + Where(sq.Eq{"c.sync_status": SyncStatusSynced}) query, args, err := q.ToSql() if err != nil { return nil, nil, fmt.Errorf("build query: %w", err) diff --git a/test/e2e/machine_test.go b/test/e2e/machine_test.go index e5f945d5..1dabc65c 100644 --- a/test/e2e/machine_test.go +++ b/test/e2e/machine_test.go @@ -3,7 +3,9 @@ package e2e import ( "context" "errors" + "strings" "testing" + "time" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/ucind" @@ -434,4 +436,68 @@ func TestMachineOperations(t *testing.T) { } assert.Equal(t, len(targetMachine.Machine.Network.Endpoints), len(updatedMachine.Network.Endpoints)) }) + + t.Run("remove machine clears container records from cluster store", func(t *testing.T) { + // Deploy a global service with an HTTP ingress port so the auto-generated Caddyfile lists + // each container's IP as an upstream. + serviceName := "test-machine-rm-cleanup" + spec := api.ServiceSpec{ + Name: serviceName, + Mode: api.ServiceModeGlobal, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + }, + Ports: []api.PortSpec{ + { + Hostname: "test-machine-rm-cleanup.example.com", + ContainerPort: 8000, + Protocol: api.ProtocolHTTP, + Mode: api.PortModeIngress, + }, + }, + } + + t.Cleanup(func() { + err := cli.RemoveService(ctx, serviceName) + if err != nil && !errors.Is(err, api.ErrNotFound) { + assert.NoError(t, err) + } + }) + _, err = cli.NewDeployment(spec, nil).Run(ctx) + require.NoError(t, err) + + svc, err := cli.InspectService(ctx, serviceName) + require.NoError(t, err) + + containers := serviceContainersByMachine(svc) + + removedMachine := c.Machines[2] + removedIP := containers[removedMachine.ID][0].Container.UncloudNetworkIP().String() + keptIP := containers[c.Machines[0].ID][0].Container.UncloudNetworkIP().String() + + // The Caddyfile contains both upstream IPs before the machine removal. + require.Eventually(t, func() bool { + cfg, err := cli.Caddy.GetConfig(ctx, nil) + if err != nil { + return false + } + return strings.Contains(cfg.Caddyfile, removedIP) && strings.Contains(cfg.Caddyfile, keptIP) + }, 15*time.Second, 100*time.Millisecond, + "expected Caddyfile to include both the to-be-removed and kept container IPs") + + _, err = cli.RemoveMachine(ctx, &pb.RemoveMachineRequest{Id: removedMachine.ID}) + require.NoError(t, err) + + // After removal, the removed machine's container record should be gone from the cluster store, + // so the Caddy controller regenerates a Caddyfile without that upstream while keeping the + // upstreams for the still-running containers. + require.Eventually(t, func() bool { + cfg, err := cli.Caddy.GetConfig(ctx, nil) + if err != nil { + return false + } + return !strings.Contains(cfg.Caddyfile, removedIP) && strings.Contains(cfg.Caddyfile, keptIP) + }, 15*time.Second, 100*time.Millisecond, + "expected Caddyfile to drop the removed machine's container IP and keep the rest") + }) }