From 30c58ec35886dde7fef26ebd092cef2a9a271ec2 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 25 Jun 2026 23:21:23 +1000 Subject: [PATCH] feat: ECS P0 teardown (DeleteCluster cascade, Deregister taskdef/instance, drain) (#423) Wire the four NotImplemented ECS teardown actions so an IaC stack and the console can reverse a deployment: - DeleteCluster cascades: force-stops every task (releasing ENIs, deregistering TG targets, returning capacity), then sweeps the cluster's whole KV subtree. Unlike AWS it cascades rather than rejecting a non-empty cluster, so terraform destroy round-trips regardless of teardown ordering. - DeregisterTaskDefinition marks a specific family:revision INACTIVE (kept describable); a bare family is rejected, matching AWS. - DeregisterContainerInstance removes an instance, force-stopping its tasks when Force is set and rejecting a non-empty instance otherwise. - UpdateContainerInstancesState sets ACTIVE/DRAINING; draining force-stops the instance's service-owned tasks so the reconciler relaunches them elsewhere. Reuses the existing forceStopTask primitive and the reaper drain pattern; adds a deleteKeysWithPrefix KV helper and INACTIVE status constants. --- spinifex/daemon/daemon.go | 4 + spinifex/daemon/daemon_handlers_ecs.go | 16 ++ spinifex/gateway/ecs/actions.go | 32 ++++ spinifex/gateway/ecs/handler.go | 8 +- spinifex/gateway/ecs/handler_test.go | 9 +- spinifex/gateway/ecs_test.go | 2 +- spinifex/handlers/ecs/cluster_teardown.go | 163 ++++++++++++++++++ .../handlers/ecs/cluster_teardown_test.go | 133 ++++++++++++++ spinifex/handlers/ecs/records.go | 6 +- spinifex/handlers/ecs/service.go | 49 ++++++ spinifex/handlers/ecs/service_nats.go | 16 ++ 11 files changed, 428 insertions(+), 10 deletions(-) create mode 100644 spinifex/handlers/ecs/cluster_teardown.go create mode 100644 spinifex/handlers/ecs/cluster_teardown_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 5a59e77c3..f68a4cae0 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -954,12 +954,16 @@ func (d *Daemon) subscribeAll() error { if d.ecsService != nil { subs = append(subs, natsSub{"ecs.CreateCluster", d.handleECSCreateCluster, "spinifex-workers"}, + natsSub{"ecs.DeleteCluster", d.handleECSDeleteCluster, "spinifex-workers"}, natsSub{"ecs.DescribeClusters", d.handleECSDescribeClusters, "spinifex-workers"}, natsSub{"ecs.ListClusters", d.handleECSListClusters, "spinifex-workers"}, natsSub{"ecs.RegisterTaskDefinition", d.handleECSRegisterTaskDefinition, "spinifex-workers"}, + natsSub{"ecs.DeregisterTaskDefinition", d.handleECSDeregisterTaskDefinition, "spinifex-workers"}, natsSub{"ecs.DescribeTaskDefinition", d.handleECSDescribeTaskDefinition, "spinifex-workers"}, natsSub{"ecs.ListTaskDefinitions", d.handleECSListTaskDefinitions, "spinifex-workers"}, natsSub{"ecs.RegisterContainerInstance", d.handleECSRegisterContainerInstance, "spinifex-workers"}, + natsSub{"ecs.DeregisterContainerInstance", d.handleECSDeregisterContainerInstance, "spinifex-workers"}, + natsSub{"ecs.UpdateContainerInstancesState", d.handleECSUpdateContainerInstancesState, "spinifex-workers"}, natsSub{"ecs.DescribeContainerInstances", d.handleECSDescribeContainerInstances, "spinifex-workers"}, natsSub{"ecs.ListContainerInstances", d.handleECSListContainerInstances, "spinifex-workers"}, natsSub{"ecs.RunTask", d.handleECSRunTask, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_handlers_ecs.go b/spinifex/daemon/daemon_handlers_ecs.go index 8891a63f7..557328958 100644 --- a/spinifex/daemon/daemon_handlers_ecs.go +++ b/spinifex/daemon/daemon_handlers_ecs.go @@ -8,6 +8,10 @@ func (d *Daemon) handleECSCreateCluster(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.CreateCluster) } +func (d *Daemon) handleECSDeleteCluster(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DeleteCluster) +} + func (d *Daemon) handleECSDescribeClusters(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.DescribeClusters) } @@ -22,6 +26,10 @@ func (d *Daemon) handleECSRegisterTaskDefinition(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.RegisterTaskDefinition) } +func (d *Daemon) handleECSDeregisterTaskDefinition(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DeregisterTaskDefinition) +} + func (d *Daemon) handleECSDescribeTaskDefinition(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.DescribeTaskDefinition) } @@ -36,6 +44,14 @@ func (d *Daemon) handleECSRegisterContainerInstance(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.RegisterContainerInstance) } +func (d *Daemon) handleECSDeregisterContainerInstance(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DeregisterContainerInstance) +} + +func (d *Daemon) handleECSUpdateContainerInstancesState(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.UpdateContainerInstancesState) +} + func (d *Daemon) handleECSDescribeContainerInstances(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.DescribeContainerInstances) } diff --git a/spinifex/gateway/ecs/actions.go b/spinifex/gateway/ecs/actions.go index 4be39f4e5..0743c09bf 100644 --- a/spinifex/gateway/ecs/actions.go +++ b/spinifex/gateway/ecs/actions.go @@ -26,6 +26,14 @@ func CreateCluster(nc *nats.Conn, accountID string, body []byte) (any, error) { return handlers_ecs.NewNATSECSService(nc).CreateCluster(input, accountID) } +func DeleteCluster(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DeleteClusterInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DeleteCluster(input, accountID) +} + func DescribeClusters(nc *nats.Conn, accountID string, body []byte) (any, error) { input := new(ecs.DescribeClustersInput) if err := unmarshalIfBody(body, input); err != nil { @@ -52,6 +60,14 @@ func RegisterTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, return handlers_ecs.NewNATSECSService(nc).RegisterTaskDefinition(input, accountID) } +func DeregisterTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DeregisterTaskDefinitionInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DeregisterTaskDefinition(input, accountID) +} + func DescribeTaskDefinition(nc *nats.Conn, accountID string, body []byte) (any, error) { input := new(ecs.DescribeTaskDefinitionInput) if err := unmarshalIfBody(body, input); err != nil { @@ -78,6 +94,22 @@ func RegisterContainerInstance(nc *nats.Conn, accountID string, body []byte) (an return handlers_ecs.NewNATSECSService(nc).RegisterContainerInstance(input, accountID) } +func DeregisterContainerInstance(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DeregisterContainerInstanceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DeregisterContainerInstance(input, accountID) +} + +func UpdateContainerInstancesState(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.UpdateContainerInstancesStateInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).UpdateContainerInstancesState(input, accountID) +} + func DescribeContainerInstances(nc *nats.Conn, accountID string, body []byte) (any, error) { input := new(ecs.DescribeContainerInstancesInput) if err := unmarshalIfBody(body, input); err != nil { diff --git a/spinifex/gateway/ecs/handler.go b/spinifex/gateway/ecs/handler.go index 4a8cdcd4d..fda5cad14 100644 --- a/spinifex/gateway/ecs/handler.go +++ b/spinifex/gateway/ecs/handler.go @@ -45,13 +45,13 @@ var Actions = map[string]Handler{ "CreateCluster": CreateCluster, "DescribeClusters": DescribeClusters, "ListClusters": ListClusters, - "DeleteCluster": NotImplemented, + "DeleteCluster": DeleteCluster, "UpdateCluster": NotImplemented, "PutClusterCapacityProviders": NotImplemented, // Task definition. "RegisterTaskDefinition": RegisterTaskDefinition, - "DeregisterTaskDefinition": NotImplemented, + "DeregisterTaskDefinition": DeregisterTaskDefinition, "DescribeTaskDefinition": DescribeTaskDefinition, "ListTaskDefinitions": ListTaskDefinitions, "ListTaskDefinitionFamilies": NotImplemented, @@ -79,10 +79,10 @@ var Actions = map[string]Handler{ // Container instance. "RegisterContainerInstance": RegisterContainerInstance, - "DeregisterContainerInstance": NotImplemented, + "DeregisterContainerInstance": DeregisterContainerInstance, "DescribeContainerInstances": DescribeContainerInstances, "ListContainerInstances": ListContainerInstances, - "UpdateContainerInstancesState": NotImplemented, + "UpdateContainerInstancesState": UpdateContainerInstancesState, // Account settings (passthrough; no enforcement v1). "PutAccountSetting": NotImplemented, diff --git a/spinifex/gateway/ecs/handler_test.go b/spinifex/gateway/ecs/handler_test.go index 03ee91e2c..89895409e 100644 --- a/spinifex/gateway/ecs/handler_test.go +++ b/spinifex/gateway/ecs/handler_test.go @@ -21,9 +21,12 @@ func TestNotImplemented(t *testing.T) { // NotImplemented stub. Wired actions error here because the nil NATS conn cannot // reach the daemon — that they no longer return NotImplemented is the assertion. var wiredActions = map[string]bool{ - "CreateCluster": true, "DescribeClusters": true, "ListClusters": true, - "RegisterTaskDefinition": true, "DescribeTaskDefinition": true, "ListTaskDefinitions": true, - "RegisterContainerInstance": true, "DescribeContainerInstances": true, "ListContainerInstances": true, + "CreateCluster": true, "DeleteCluster": true, "DescribeClusters": true, "ListClusters": true, + "RegisterTaskDefinition": true, "DeregisterTaskDefinition": true, + "DescribeTaskDefinition": true, "ListTaskDefinitions": true, + "RegisterContainerInstance": true, "DeregisterContainerInstance": true, + "UpdateContainerInstancesState": true, + "DescribeContainerInstances": true, "ListContainerInstances": true, "RunTask": true, "StartTask": true, "StopTask": true, "DescribeTasks": true, "ListTasks": true, "CreateService": true, "UpdateService": true, "DeleteService": true, "DescribeServices": true, "ListServices": true, diff --git a/spinifex/gateway/ecs_test.go b/spinifex/gateway/ecs_test.go index 504551ce0..e0dea996a 100644 --- a/spinifex/gateway/ecs_test.go +++ b/spinifex/gateway/ecs_test.go @@ -80,7 +80,7 @@ func TestECSRequest_UnknownAction(t *testing.T) { func TestECSRequest_KnownActionNotImplemented(t *testing.T) { gw := &GatewayConfig{DisableLogging: true} err := gw.ECS_Request(httptest.NewRecorder(), - setupECSRequest("AmazonEC2ContainerServiceV20141113.DeleteCluster", "{}")) + setupECSRequest("AmazonEC2ContainerServiceV20141113.UpdateCluster", "{}")) require.Error(t, err) assert.Equal(t, awserrors.ErrorNotImplemented, err.Error()) } diff --git a/spinifex/handlers/ecs/cluster_teardown.go b/spinifex/handlers/ecs/cluster_teardown.go new file mode 100644 index 000000000..e06f19ba8 --- /dev/null +++ b/spinifex/handlers/ecs/cluster_teardown.go @@ -0,0 +1,163 @@ +package handlers_ecs + +import ( + "errors" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go" +) + +// clusterKeyPrefix is the KV subtree holding all of a cluster's records. +func clusterKeyPrefix(cluster string) string { + return "clusters/" + cluster + "/" +} + +// DeleteCluster tears a cluster down and sweeps its KV subtree. Unlike AWS (which +// rejects a non-empty cluster), Mulga cascades so terraform destroy round-trips +// regardless of teardown ordering: every non-STOPPED task is force-stopped +// (releasing ENIs, deregistering TG targets, returning capacity), every service +// is marked INACTIVE, then the whole clusters/{name}/ prefix is deleted. Returns +// the cluster with Status INACTIVE. +func (s *Service) DeleteCluster(input *ecs.DeleteClusterInput, accountID string) (*ecs.DeleteClusterOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(cluster), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSClusterNotFound) + } + + tasks, err := s.listTaskRecords(kv, cluster) + if err != nil { + return nil, err + } + for i := range tasks { + s.forceStopTask(kv, accountID, &tasks[i], "Cluster deleted") + } + + if err := deleteKeysWithPrefix(kv, clusterKeyPrefix(cluster)); err != nil { + return nil, err + } + + rec.Status = ClusterStatusInactive + return &ecs.DeleteClusterOutput{Cluster: rec.toAWS()}, nil +} + +// DeregisterContainerInstance removes a container instance from a cluster. With +// Force set it first force-stops the instance's non-STOPPED tasks; without Force +// it rejects an instance that still has running tasks (AWS parity). The instance +// record and its assignment inbox are deleted; the response carries Status +// INACTIVE. +func (s *Service) DeregisterContainerInstance(input *ecs.DeregisterContainerInstanceInput, accountID string) (*ecs.DeregisterContainerInstanceOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + id := containerInstanceShortID(aws.StringValue(input.ContainerInstance)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec InstanceRecord + found, err := getJSON(kv, InstanceKey(cluster, id), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + + active, err := s.instanceActiveTasks(kv, cluster, id) + if err != nil { + return nil, err + } + if len(active) > 0 && !aws.BoolValue(input.Force) { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + for i := range active { + s.forceStopTask(kv, accountID, &active[i], "Container instance deregistered") + } + + if derr := deleteKeysWithPrefix(kv, AssignmentsPrefix(cluster, id)); derr != nil { + return nil, derr + } + if derr := kv.Delete(InstanceKey(cluster, id)); derr != nil { + return nil, derr + } + rec.Status = ClusterStatusInactive + return &ecs.DeregisterContainerInstanceOutput{ContainerInstance: s.instanceToAWS(&rec)}, nil +} + +// UpdateContainerInstancesState sets the requested instances ACTIVE or DRAINING. +// Draining force-stops the instance's service-owned tasks so the reconciler +// relaunches them elsewhere; standalone (non-service) tasks are left running, +// matching AWS. Unknown instances surface as Failures. +func (s *Service) UpdateContainerInstancesState(input *ecs.UpdateContainerInstancesStateInput, accountID string) (*ecs.UpdateContainerInstancesStateOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + status := aws.StringValue(input.Status) + if status != InstanceStatusActive && status != InstanceStatusDraining { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + out := &ecs.UpdateContainerInstancesStateOutput{} + for _, ref := range awsStringSlice(input.ContainerInstances) { + id := containerInstanceShortID(ref) + var rec InstanceRecord + found, gerr := getJSON(kv, InstanceKey(cluster, id), &rec) + if gerr != nil { + return nil, gerr + } + if !found { + out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")}) + continue + } + rec.Status = status + if perr := putJSON(kv, InstanceKey(cluster, id), &rec); perr != nil { + return nil, perr + } + if status == InstanceStatusDraining { + s.drainInstanceServiceTasks(kv, accountID, cluster, id) + } + out.ContainerInstances = append(out.ContainerInstances, s.instanceToAWS(&rec)) + } + return out, nil +} + +// instanceActiveTasks returns a cluster's non-STOPPED tasks placed on instanceID. +func (s *Service) instanceActiveTasks(kv nats.KeyValue, cluster, instanceID string) ([]TaskRecord, error) { + all, err := s.listTaskRecords(kv, cluster) + if err != nil { + return nil, err + } + out := make([]TaskRecord, 0, len(all)) + for _, t := range all { + if t.ContainerInstanceID == instanceID && t.LastStatus != TaskStatusStopped { + out = append(out, t) + } + } + return out, nil +} + +// drainInstanceServiceTasks force-stops the instance's service-owned tasks on an +// intentional DRAINING; the service reconciler then relaunches them on another +// instance. Standalone tasks are left running (AWS DRAINING semantics). +func (s *Service) drainInstanceServiceTasks(kv nats.KeyValue, accountID, cluster, instanceID string) { + active, err := s.instanceActiveTasks(kv, cluster, instanceID) + if err != nil { + return + } + for i := range active { + if serviceNameFromGroup(active[i].Group) == "" { + continue + } + s.forceStopTask(kv, accountID, &active[i], "Container instance draining") + } +} diff --git a/spinifex/handlers/ecs/cluster_teardown_test.go b/spinifex/handlers/ecs/cluster_teardown_test.go new file mode 100644 index 000000000..449bc420d --- /dev/null +++ b/spinifex/handlers/ecs/cluster_teardown_test.go @@ -0,0 +1,133 @@ +package handlers_ecs + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestService_DeleteCluster_CascadesAndSweeps(t *testing.T) { + svc, _, kv := serviceTestRig(t) + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(2), + }, testAccountID) + require.NoError(t, err) + + out, err := svc.DeleteCluster(&ecs.DeleteClusterInput{Cluster: aws.String("web")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, ClusterStatusInactive, aws.StringValue(out.Cluster.Status)) + + // Whole cluster subtree swept: no clusters, no leftover task/service keys. + list, err := svc.ListClusters(&ecs.ListClustersInput{}, testAccountID) + require.NoError(t, err) + assert.Empty(t, list.ClusterArns) + keys, err := keysWithPrefix(kv, clusterKeyPrefix("web")) + require.NoError(t, err) + assert.Empty(t, keys) +} + +func TestService_DeleteCluster_Unknown(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.DeleteCluster(&ecs.DeleteClusterInput{Cluster: aws.String("ghost")}, testAccountID) + require.Error(t, err) +} + +func TestService_DeregisterTaskDefinition_InactivatesRevision(t *testing.T) { + svc, _ := newTestService(t) + registerTaskDef(t, svc, "app", 128, 256) + + out, err := svc.DeregisterTaskDefinition(&ecs.DeregisterTaskDefinitionInput{ + TaskDefinition: aws.String("app:1"), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, TaskDefStatusInactive, aws.StringValue(out.TaskDefinition.Status)) + + // Still describable, now INACTIVE. + d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("app:1")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, TaskDefStatusInactive, aws.StringValue(d.TaskDefinition.Status)) +} + +func TestService_DeregisterTaskDefinition_RequiresRevision(t *testing.T) { + svc, _ := newTestService(t) + registerTaskDef(t, svc, "app", 128, 256) + // Bare family (no revision) is rejected, matching AWS. + _, err := svc.DeregisterTaskDefinition(&ecs.DeregisterTaskDefinitionInput{TaskDefinition: aws.String("app")}, testAccountID) + require.Error(t, err) +} + +func TestService_DeregisterContainerInstance_ForceStopsAndRemoves(t *testing.T) { + svc, _, kv := serviceTestRig(t) + _, err := svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + + // Without force, an instance with a live task is rejected. + _, err = svc.DeregisterContainerInstance(&ecs.DeregisterContainerInstanceInput{ + Cluster: aws.String("web"), ContainerInstance: aws.String("i-1"), + }, testAccountID) + require.Error(t, err) + + // With force, the task is stopped and the instance record removed. + out, err := svc.DeregisterContainerInstance(&ecs.DeregisterContainerInstanceInput{ + Cluster: aws.String("web"), ContainerInstance: aws.String("i-1"), Force: aws.Bool(true), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, ClusterStatusInactive, aws.StringValue(out.ContainerInstance.Status)) + + active, err := svc.instanceActiveTasks(kv, "web", "i-1") + require.NoError(t, err) + assert.Empty(t, active) + + desc, err := svc.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, desc.ContainerInstances) + require.Len(t, desc.Failures, 1) +} + +func TestService_UpdateContainerInstancesState_DrainsServiceTasks(t *testing.T) { + svc, _, kv := serviceTestRig(t) + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(2), + }, testAccountID) + require.NoError(t, err) + + out, err := svc.UpdateContainerInstancesState(&ecs.UpdateContainerInstancesStateInput{ + Cluster: aws.String("web"), + ContainerInstances: []*string{aws.String("i-1")}, + Status: aws.String(InstanceStatusDraining), + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.ContainerInstances, 1) + assert.Equal(t, InstanceStatusDraining, aws.StringValue(out.ContainerInstances[0].Status)) + + // Service-owned tasks on the drained instance are stopped. + live, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + assert.Empty(t, live) +} + +func TestService_UpdateContainerInstancesState_RejectsBadStatusAndUnknown(t *testing.T) { + svc, _, _ := serviceTestRig(t) + _, err := svc.UpdateContainerInstancesState(&ecs.UpdateContainerInstancesStateInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")}, + Status: aws.String("BOGUS"), + }, testAccountID) + require.Error(t, err) + + out, err := svc.UpdateContainerInstancesState(&ecs.UpdateContainerInstancesStateInput{ + Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-ghost")}, + Status: aws.String(InstanceStatusActive), + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.ContainerInstances) + require.Len(t, out.Failures, 1) +} diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 04fbb018c..1ebe44cda 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -10,12 +10,14 @@ import ( // Lifecycle status constants matching the AWS ECS enums. Task statuses are // re-exported from the bus package so the scheduler and the wire payloads agree. const ( - ClusterStatusActive = "ACTIVE" + ClusterStatusActive = "ACTIVE" + ClusterStatusInactive = "INACTIVE" InstanceStatusActive = "ACTIVE" InstanceStatusDraining = "DRAINING" - TaskDefStatusActive = "ACTIVE" + TaskDefStatusActive = "ACTIVE" + TaskDefStatusInactive = "INACTIVE" TaskStatusPending = bus.TaskStatusPending TaskStatusRunning = bus.TaskStatusRunning diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index 8a8a9ed14..fefa64d98 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -20,14 +20,18 @@ import ( // remaining actions stay NotImplemented at the gateway. type ECSService interface { CreateCluster(input *ecs.CreateClusterInput, accountID string) (*ecs.CreateClusterOutput, error) + DeleteCluster(input *ecs.DeleteClusterInput, accountID string) (*ecs.DeleteClusterOutput, error) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) ListClusters(input *ecs.ListClustersInput, accountID string) (*ecs.ListClustersOutput, error) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, accountID string) (*ecs.RegisterTaskDefinitionOutput, error) + DeregisterTaskDefinition(input *ecs.DeregisterTaskDefinitionInput, accountID string) (*ecs.DeregisterTaskDefinitionOutput, error) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) ListTaskDefinitions(input *ecs.ListTaskDefinitionsInput, accountID string) (*ecs.ListTaskDefinitionsOutput, error) RegisterContainerInstance(input *ecs.RegisterContainerInstanceInput, accountID string) (*ecs.RegisterContainerInstanceOutput, error) + DeregisterContainerInstance(input *ecs.DeregisterContainerInstanceInput, accountID string) (*ecs.DeregisterContainerInstanceOutput, error) + UpdateContainerInstancesState(input *ecs.UpdateContainerInstancesStateInput, accountID string) (*ecs.UpdateContainerInstancesStateOutput, error) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) @@ -119,6 +123,24 @@ func putJSON(kv nats.KeyValue, key string, v any) error { return nil } +// deleteKeysWithPrefix deletes every KV key under prefix. Used by DeleteCluster +// to sweep a cluster's whole key subtree (meta/instances/tasks/services/ +// assignments) after its tasks are stopped. Best-effort per key; the first +// delete error is returned after attempting the rest. +func deleteKeysWithPrefix(kv nats.KeyValue, prefix string) error { + keys, err := keysWithPrefix(kv, prefix) + if err != nil { + return err + } + var firstErr error + for _, k := range keys { + if derr := kv.Delete(k); derr != nil && firstErr == nil { + firstErr = derr + } + } + return firstErr +} + // keysWithPrefix returns all KV keys under prefix. func keysWithPrefix(kv nats.KeyValue, prefix string) ([]string, error) { keys, err := kv.Keys() @@ -290,6 +312,33 @@ func (s *Service) nextRevision(kv nats.KeyValue, family string) (int, error) { return latest + 1, nil } +// DeregisterTaskDefinition marks a specific task-definition revision INACTIVE. +// AWS requires an explicit family:revision (a bare family is rejected); the +// revision stays describable, matching AWS. Idempotent. +func (s *Service) DeregisterTaskDefinition(input *ecs.DeregisterTaskDefinitionInput, accountID string) (*ecs.DeregisterTaskDefinitionOutput, error) { + family, rev := parseTaskDefRef(aws.StringValue(input.TaskDefinition)) + if family == "" || rev == 0 { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec TaskDefRecord + found, err := getJSON(kv, TaskDefRevKey(family, rev), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + rec.Status = TaskDefStatusInactive + if err := putJSON(kv, TaskDefRevKey(family, rev), &rec); err != nil { + return nil, err + } + return &ecs.DeregisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil +} + // DescribeTaskDefinition resolves "family", "family:rev" or an ARN to a revision. func (s *Service) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) { kv, err := s.bucket(accountID) diff --git a/spinifex/handlers/ecs/service_nats.go b/spinifex/handlers/ecs/service_nats.go index 996b5b45d..feee9b3bb 100644 --- a/spinifex/handlers/ecs/service_nats.go +++ b/spinifex/handlers/ecs/service_nats.go @@ -29,6 +29,10 @@ func (s *NATSECSService) CreateCluster(input *ecs.CreateClusterInput, accountID return utils.NATSRequest[ecs.CreateClusterOutput](s.natsConn, "ecs.CreateCluster", input, defaultTimeout, accountID) } +func (s *NATSECSService) DeleteCluster(input *ecs.DeleteClusterInput, accountID string) (*ecs.DeleteClusterOutput, error) { + return utils.NATSRequest[ecs.DeleteClusterOutput](s.natsConn, "ecs.DeleteCluster", input, defaultTimeout, accountID) +} + func (s *NATSECSService) DescribeClusters(input *ecs.DescribeClustersInput, accountID string) (*ecs.DescribeClustersOutput, error) { return utils.NATSRequest[ecs.DescribeClustersOutput](s.natsConn, "ecs.DescribeClusters", input, defaultTimeout, accountID) } @@ -43,6 +47,10 @@ func (s *NATSECSService) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitio return utils.NATSRequest[ecs.RegisterTaskDefinitionOutput](s.natsConn, "ecs.RegisterTaskDefinition", input, defaultTimeout, accountID) } +func (s *NATSECSService) DeregisterTaskDefinition(input *ecs.DeregisterTaskDefinitionInput, accountID string) (*ecs.DeregisterTaskDefinitionOutput, error) { + return utils.NATSRequest[ecs.DeregisterTaskDefinitionOutput](s.natsConn, "ecs.DeregisterTaskDefinition", input, defaultTimeout, accountID) +} + func (s *NATSECSService) DescribeTaskDefinition(input *ecs.DescribeTaskDefinitionInput, accountID string) (*ecs.DescribeTaskDefinitionOutput, error) { return utils.NATSRequest[ecs.DescribeTaskDefinitionOutput](s.natsConn, "ecs.DescribeTaskDefinition", input, defaultTimeout, accountID) } @@ -57,6 +65,14 @@ func (s *NATSECSService) RegisterContainerInstance(input *ecs.RegisterContainerI return utils.NATSRequest[ecs.RegisterContainerInstanceOutput](s.natsConn, "ecs.RegisterContainerInstance", input, defaultTimeout, accountID) } +func (s *NATSECSService) DeregisterContainerInstance(input *ecs.DeregisterContainerInstanceInput, accountID string) (*ecs.DeregisterContainerInstanceOutput, error) { + return utils.NATSRequest[ecs.DeregisterContainerInstanceOutput](s.natsConn, "ecs.DeregisterContainerInstance", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) UpdateContainerInstancesState(input *ecs.UpdateContainerInstancesStateInput, accountID string) (*ecs.UpdateContainerInstancesStateOutput, error) { + return utils.NATSRequest[ecs.UpdateContainerInstancesStateOutput](s.natsConn, "ecs.UpdateContainerInstancesState", input, defaultTimeout, accountID) +} + func (s *NATSECSService) DescribeContainerInstances(input *ecs.DescribeContainerInstancesInput, accountID string) (*ecs.DescribeContainerInstancesOutput, error) { return utils.NATSRequest[ecs.DescribeContainerInstancesOutput](s.natsConn, "ecs.DescribeContainerInstances", input, defaultTimeout, accountID) }