From 8da043ce767201007d0e9d36982b705f9231f559 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 10 Jul 2026 15:01:59 +1000 Subject: [PATCH] feat(ecs): tags API + capacity-provider CRUD scaffolding (#499) * feat(ecs): tags API (TagResource/UntagResource/ListTagsForResource) * feat(ecs): capacity-provider CRUD scaffolding --- spinifex/daemon/daemon.go | 7 + spinifex/daemon/daemon_handlers_ecs.go | 32 +++ spinifex/gateway/ecs/actions.go | 60 ++++ spinifex/gateway/ecs/handler.go | 16 +- spinifex/gateway/ecs/handler_test.go | 3 + spinifex/handlers/ecs/capacity_providers.go | 233 +++++++++++++++ .../handlers/ecs/capacity_providers_test.go | 146 ++++++++++ spinifex/handlers/ecs/conv.go | 10 + spinifex/handlers/ecs/instances.go | 4 + spinifex/handlers/ecs/records.go | 121 +++++--- spinifex/handlers/ecs/service.go | 47 ++- spinifex/handlers/ecs/service_nats.go | 32 +++ spinifex/handlers/ecs/services.go | 2 + spinifex/handlers/ecs/store.go | 12 + spinifex/handlers/ecs/tags.go | 268 ++++++++++++++++++ spinifex/handlers/ecs/tags_test.go | 229 +++++++++++++++ spinifex/handlers/ecs/task_stop.go | 1 + spinifex/handlers/ecs/tasks.go | 2 + 18 files changed, 1185 insertions(+), 40 deletions(-) create mode 100644 spinifex/handlers/ecs/capacity_providers.go create mode 100644 spinifex/handlers/ecs/capacity_providers_test.go create mode 100644 spinifex/handlers/ecs/tags.go create mode 100644 spinifex/handlers/ecs/tags_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index d794530ba..d5e1f458e 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1014,6 +1014,13 @@ func (d *Daemon) subscribeAll() error { natsSub{"ecs.SubmitTaskStateChange", d.handleECSSubmitTaskStateChange, "spinifex-workers"}, natsSub{"ecs.PollAssignments", d.handleECSPollAssignments, "spinifex-workers"}, natsSub{"ecs.ProvisionCapacity", d.handleECSProvisionCapacity, "spinifex-workers"}, + natsSub{"ecs.TagResource", d.handleECSTagResource, "spinifex-workers"}, + natsSub{"ecs.UntagResource", d.handleECSUntagResource, "spinifex-workers"}, + natsSub{"ecs.ListTagsForResource", d.handleECSListTagsForResource, "spinifex-workers"}, + natsSub{"ecs.PutClusterCapacityProviders", d.handleECSPutClusterCapacityProviders, "spinifex-workers"}, + natsSub{"ecs.CreateCapacityProvider", d.handleECSCreateCapacityProvider, "spinifex-workers"}, + natsSub{"ecs.DescribeCapacityProviders", d.handleECSDescribeCapacityProviders, "spinifex-workers"}, + natsSub{"ecs.DeleteCapacityProvider", d.handleECSDeleteCapacityProvider, "spinifex-workers"}, ) } diff --git a/spinifex/daemon/daemon_handlers_ecs.go b/spinifex/daemon/daemon_handlers_ecs.go index 9f59d0ebc..859452edf 100644 --- a/spinifex/daemon/daemon_handlers_ecs.go +++ b/spinifex/daemon/daemon_handlers_ecs.go @@ -117,3 +117,35 @@ func (d *Daemon) handleECSPollAssignments(msg *nats.Msg) { func (d *Daemon) handleECSProvisionCapacity(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.ProvisionCapacity) } + +// --- Tags --- + +func (d *Daemon) handleECSTagResource(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.TagResource) +} + +func (d *Daemon) handleECSUntagResource(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.UntagResource) +} + +func (d *Daemon) handleECSListTagsForResource(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListTagsForResource) +} + +// --- Capacity providers --- + +func (d *Daemon) handleECSPutClusterCapacityProviders(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.PutClusterCapacityProviders) +} + +func (d *Daemon) handleECSCreateCapacityProvider(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.CreateCapacityProvider) +} + +func (d *Daemon) handleECSDescribeCapacityProviders(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeCapacityProviders) +} + +func (d *Daemon) handleECSDeleteCapacityProvider(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DeleteCapacityProvider) +} diff --git a/spinifex/gateway/ecs/actions.go b/spinifex/gateway/ecs/actions.go index a20607c43..5fd005f7b 100644 --- a/spinifex/gateway/ecs/actions.go +++ b/spinifex/gateway/ecs/actions.go @@ -243,3 +243,63 @@ func PollAssignments(ctx context.Context, nc *nats.Conn, accountID string, body } return handlers_ecs.NewNATSECSService(nc).PollAssignments(ctx, input, accountID) } + +// --- Tags --- + +func TagResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.TagResourceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).TagResource(ctx, input, accountID) +} + +func UntagResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.UntagResourceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).UntagResource(ctx, input, accountID) +} + +func ListTagsForResource(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListTagsForResourceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListTagsForResource(ctx, input, accountID) +} + +// --- Capacity providers --- + +func PutClusterCapacityProviders(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.PutClusterCapacityProvidersInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).PutClusterCapacityProviders(ctx, input, accountID) +} + +func CreateCapacityProvider(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.CreateCapacityProviderInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).CreateCapacityProvider(ctx, input, accountID) +} + +func DescribeCapacityProviders(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeCapacityProvidersInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeCapacityProviders(ctx, input, accountID) +} + +func DeleteCapacityProvider(ctx context.Context, nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DeleteCapacityProviderInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DeleteCapacityProvider(ctx, input, accountID) +} diff --git a/spinifex/gateway/ecs/handler.go b/spinifex/gateway/ecs/handler.go index 218eda4ac..0c1790efa 100644 --- a/spinifex/gateway/ecs/handler.go +++ b/spinifex/gateway/ecs/handler.go @@ -42,13 +42,19 @@ func NotImplemented(_ context.Context, _ *nats.Conn, _ string, _ []byte) (any, e // Wired actions point at their handler; unimplemented actions keep the // NotImplemented value. The key set is the v1 API contract. var Actions = map[string]Handler{ - // Cluster. PutClusterCapacityProviders is a no-op stub in v1 (Q15). + // Cluster. "CreateCluster": CreateCluster, "DescribeClusters": DescribeClusters, "ListClusters": ListClusters, "DeleteCluster": DeleteCluster, "UpdateCluster": NotImplemented, - "PutClusterCapacityProviders": NotImplemented, + "PutClusterCapacityProviders": PutClusterCapacityProviders, + + // Capacity providers. Strategy is accepted + persisted but inert in v1 + // (no scheduler coupling, no scale loop). + "CreateCapacityProvider": CreateCapacityProvider, + "DescribeCapacityProviders": DescribeCapacityProviders, + "DeleteCapacityProvider": DeleteCapacityProvider, // Task definition. "RegisterTaskDefinition": RegisterTaskDefinition, @@ -91,9 +97,9 @@ var Actions = map[string]Handler{ "ListAccountSettings": NotImplemented, // Tags. - "TagResource": NotImplemented, - "UntagResource": NotImplemented, - "ListTagsForResource": NotImplemented, + "TagResource": TagResource, + "UntagResource": UntagResource, + "ListTagsForResource": ListTagsForResource, } // RawJSONActions encode their response with encoding/json instead of jsonutil. diff --git a/spinifex/gateway/ecs/handler_test.go b/spinifex/gateway/ecs/handler_test.go index 368bae433..b3123ed32 100644 --- a/spinifex/gateway/ecs/handler_test.go +++ b/spinifex/gateway/ecs/handler_test.go @@ -33,6 +33,9 @@ var wiredActions = map[string]bool{ "DescribeServices": true, "ListServices": true, "SubmitTaskStateChange": true, "PollAssignments": true, + "TagResource": true, "UntagResource": true, "ListTagsForResource": true, + "PutClusterCapacityProviders": true, + "CreateCapacityProvider": true, "DescribeCapacityProviders": true, "DeleteCapacityProvider": true, } func TestActions_StubsReturnNotImplemented(t *testing.T) { diff --git a/spinifex/handlers/ecs/capacity_providers.go b/spinifex/handlers/ecs/capacity_providers.go new file mode 100644 index 000000000..6ec72d9eb --- /dev/null +++ b/spinifex/handlers/ecs/capacity_providers.go @@ -0,0 +1,233 @@ +package handlers_ecs + +import ( + "context" + "errors" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/awserrors" +) + +// PutClusterCapacityProviders persists the cluster's capacity-provider names +// and default strategy. Accepted and stored only: no scheduler coupling, no +// scale loop (a separate follow-on binds this to an ASG primitive). +func (s *Service) PutClusterCapacityProviders(_ context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + 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) + } + + rec.CapacityProviders = awsStringSlice(input.CapacityProviders) + rec.DefaultCapacityProviderStrategy = strategyFromAWS(input.DefaultCapacityProviderStrategy) + if err := putJSON(kv, ClusterMetaKey(cluster), &rec); err != nil { + return nil, err + } + return &ecs.PutClusterCapacityProvidersOutput{Cluster: rec.toAWS()}, nil +} + +// CreateCapacityProvider persists a new capacity provider. Idempotent: +// re-creating an existing name returns the existing record. +func (s *Service) CreateCapacityProvider(_ context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error) { + if input == nil || aws.StringValue(input.Name) == "" || input.AutoScalingGroupProvider == nil || + aws.StringValue(input.AutoScalingGroupProvider.AutoScalingGroupArn) == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + name := aws.StringValue(input.Name) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + + var existing CapacityProviderRecord + found, err := getJSON(kv, CapacityProviderKey(name), &existing) + if err != nil { + return nil, err + } + if found { + return &ecs.CreateCapacityProviderOutput{CapacityProvider: existing.toAWS()}, nil + } + + rec := CapacityProviderRecord{ + Name: name, + ARN: CapacityProviderARN(s.region, accountID, name), + Status: CapacityProviderStatusActive, + AutoScalingGroupProvider: autoScalingGroupProviderFromAWS(input.AutoScalingGroupProvider), + Tags: tagsToMap(input.Tags), + CreatedAt: time.Now().UTC(), + } + if err := putJSON(kv, CapacityProviderKey(name), &rec); err != nil { + return nil, err + } + return &ecs.CreateCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil +} + +// DescribeCapacityProviders returns the named providers, or every provider in +// the account when none are named. Unknown names surface as failures. +func (s *Service) DescribeCapacityProviders(_ context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + out := &ecs.DescribeCapacityProvidersOutput{} + + names := awsStringSlice(input.CapacityProviders) + if len(names) == 0 { + keys, kerr := keysWithPrefix(kv, CapacityProvidersPrefix()) + if kerr != nil { + return nil, kerr + } + for _, k := range keys { + var rec CapacityProviderRecord + ok, gerr := getJSON(kv, k, &rec) + if gerr != nil { + return nil, gerr + } + if ok { + out.CapacityProviders = append(out.CapacityProviders, rec.toAWS()) + } + } + return out, nil + } + + for _, ref := range names { + name := capacityProviderShortName(ref) + var rec CapacityProviderRecord + ok, gerr := getJSON(kv, CapacityProviderKey(name), &rec) + if gerr != nil { + return nil, gerr + } + if ok { + out.CapacityProviders = append(out.CapacityProviders, rec.toAWS()) + } else { + out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")}) + } + } + return out, nil +} + +// DeleteCapacityProvider removes a capacity provider record. Clusters that +// still reference the name by CapacityProviders are left untouched (v1 does +// not enforce referential integrity here). +func (s *Service) DeleteCapacityProvider(_ context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error) { + if input == nil || aws.StringValue(input.CapacityProvider) == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + name := capacityProviderShortName(aws.StringValue(input.CapacityProvider)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec CapacityProviderRecord + found, err := getJSON(kv, CapacityProviderKey(name), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + if err := kv.Delete(CapacityProviderKey(name)); err != nil { + return nil, err + } + return &ecs.DeleteCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil +} + +// --- Conversions --- + +func (r *CapacityProviderRecord) toAWS() *ecs.CapacityProvider { + return &ecs.CapacityProvider{ + Name: aws.String(r.Name), + CapacityProviderArn: aws.String(r.ARN), + Status: aws.String(r.Status), + AutoScalingGroupProvider: r.AutoScalingGroupProvider.toAWS(), + Tags: tagsToAWS(r.Tags), + } +} + +func autoScalingGroupProviderFromAWS(in *ecs.AutoScalingGroupProvider) AutoScalingGroupProviderRecord { + if in == nil { + return AutoScalingGroupProviderRecord{} + } + rec := AutoScalingGroupProviderRecord{ + AutoScalingGroupARN: aws.StringValue(in.AutoScalingGroupArn), + ManagedTerminationProtection: aws.StringValue(in.ManagedTerminationProtection), + } + if in.ManagedScaling != nil { + rec.ManagedScaling = ManagedScalingRecord{ + Status: aws.StringValue(in.ManagedScaling.Status), + TargetCapacity: int(aws.Int64Value(in.ManagedScaling.TargetCapacity)), + MinimumScalingStepSize: int(aws.Int64Value(in.ManagedScaling.MinimumScalingStepSize)), + MaximumScalingStepSize: int(aws.Int64Value(in.ManagedScaling.MaximumScalingStepSize)), + InstanceWarmupPeriod: int(aws.Int64Value(in.ManagedScaling.InstanceWarmupPeriod)), + } + } + return rec +} + +func (r AutoScalingGroupProviderRecord) toAWS() *ecs.AutoScalingGroupProvider { + out := &ecs.AutoScalingGroupProvider{AutoScalingGroupArn: aws.String(r.AutoScalingGroupARN)} + if r.ManagedTerminationProtection != "" { + out.ManagedTerminationProtection = aws.String(r.ManagedTerminationProtection) + } + if (r.ManagedScaling != ManagedScalingRecord{}) { + ms := &ecs.ManagedScaling{} + if r.ManagedScaling.Status != "" { + ms.Status = aws.String(r.ManagedScaling.Status) + } + if r.ManagedScaling.TargetCapacity > 0 { + ms.TargetCapacity = aws.Int64(int64(r.ManagedScaling.TargetCapacity)) + } + if r.ManagedScaling.MinimumScalingStepSize > 0 { + ms.MinimumScalingStepSize = aws.Int64(int64(r.ManagedScaling.MinimumScalingStepSize)) + } + if r.ManagedScaling.MaximumScalingStepSize > 0 { + ms.MaximumScalingStepSize = aws.Int64(int64(r.ManagedScaling.MaximumScalingStepSize)) + } + if r.ManagedScaling.InstanceWarmupPeriod > 0 { + ms.InstanceWarmupPeriod = aws.Int64(int64(r.ManagedScaling.InstanceWarmupPeriod)) + } + out.ManagedScaling = ms + } + return out +} + +// strategyFromAWS maps the SDK capacity-provider strategy list to the +// persisted subset, dropping nil entries. +func strategyFromAWS(in []*ecs.CapacityProviderStrategyItem) []CapacityProviderStrategyItem { + out := make([]CapacityProviderStrategyItem, 0, len(in)) + for _, item := range in { + if item == nil { + continue + } + out = append(out, CapacityProviderStrategyItem{ + Provider: aws.StringValue(item.CapacityProvider), + Weight: int(aws.Int64Value(item.Weight)), + Base: int(aws.Int64Value(item.Base)), + }) + } + return out +} + +func (item CapacityProviderStrategyItem) toAWS() *ecs.CapacityProviderStrategyItem { + out := &ecs.CapacityProviderStrategyItem{CapacityProvider: aws.String(item.Provider)} + if item.Weight > 0 { + out.Weight = aws.Int64(int64(item.Weight)) + } + if item.Base > 0 { + out.Base = aws.Int64(int64(item.Base)) + } + return out +} diff --git a/spinifex/handlers/ecs/capacity_providers_test.go b/spinifex/handlers/ecs/capacity_providers_test.go new file mode 100644 index 000000000..753a2b636 --- /dev/null +++ b/spinifex/handlers/ecs/capacity_providers_test.go @@ -0,0 +1,146 @@ +package handlers_ecs + +import ( + "context" + "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_CreateCapacityProvider_Idempotent(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + in := &ecs.CreateCapacityProviderInput{ + Name: aws.String("cp-1"), + AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{ + AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"), + ManagedScaling: &ecs.ManagedScaling{ + Status: aws.String(ecs.ManagedScalingStatusEnabled), + TargetCapacity: aws.Int64(90), + MinimumScalingStepSize: aws.Int64(1), + MaximumScalingStepSize: aws.Int64(10), + }, + }, + Tags: []*ecs.Tag{{Key: aws.String("team"), Value: aws.String("infra")}}, + } + out, err := svc.CreateCapacityProvider(ctx, in, testAccountID) + require.NoError(t, err) + require.NotNil(t, out.CapacityProvider) + assert.Equal(t, "cp-1", aws.StringValue(out.CapacityProvider.Name)) + assert.Equal(t, CapacityProviderStatusActive, aws.StringValue(out.CapacityProvider.Status)) + assert.Equal(t, CapacityProviderARN(testRegion, testAccountID, "cp-1"), aws.StringValue(out.CapacityProvider.CapacityProviderArn)) + require.NotNil(t, out.CapacityProvider.AutoScalingGroupProvider) + require.NotNil(t, out.CapacityProvider.AutoScalingGroupProvider.ManagedScaling) + assert.Equal(t, int64(90), aws.Int64Value(out.CapacityProvider.AutoScalingGroupProvider.ManagedScaling.TargetCapacity)) + require.Len(t, out.CapacityProvider.Tags, 1) + + // Re-creating the same name returns the existing record, not a duplicate. + again, err := svc.CreateCapacityProvider(ctx, in, testAccountID) + require.NoError(t, err) + assert.Equal(t, aws.StringValue(out.CapacityProvider.CapacityProviderArn), aws.StringValue(again.CapacityProvider.CapacityProviderArn)) + + list, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{}, testAccountID) + require.NoError(t, err) + assert.Len(t, list.CapacityProviders, 1) +} + +func TestService_CreateCapacityProvider_InvalidParameter(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{}, testAccountID) + require.Error(t, err) + + _, err = svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{Name: aws.String("cp")}, testAccountID) + require.Error(t, err) +} + +func TestService_DescribeCapacityProviders_NamedAndMissing(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{ + Name: aws.String("cp-1"), + AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{ + AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"), + }, + }, testAccountID) + require.NoError(t, err) + + out, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{ + CapacityProviders: []*string{aws.String("cp-1"), aws.String("cp-missing")}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.CapacityProviders, 1) + require.Len(t, out.Failures, 1) + assert.Equal(t, "cp-missing", aws.StringValue(out.Failures[0].Arn)) +} + +func TestService_DeleteCapacityProvider(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCapacityProvider(ctx, &ecs.CreateCapacityProviderInput{ + Name: aws.String("cp-1"), + AutoScalingGroupProvider: &ecs.AutoScalingGroupProvider{ + AutoScalingGroupArn: aws.String("arn:aws:autoscaling:r:a:autoScalingGroup:x:autoScalingGroupName/asg-1"), + }, + }, testAccountID) + require.NoError(t, err) + + out, err := svc.DeleteCapacityProvider(ctx, &ecs.DeleteCapacityProviderInput{CapacityProvider: aws.String("cp-1")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "cp-1", aws.StringValue(out.CapacityProvider.Name)) + + list, err := svc.DescribeCapacityProviders(ctx, &ecs.DescribeCapacityProvidersInput{}, testAccountID) + require.NoError(t, err) + assert.Empty(t, list.CapacityProviders) + + _, err = svc.DeleteCapacityProvider(ctx, &ecs.DeleteCapacityProviderInput{CapacityProvider: aws.String("cp-1")}, testAccountID) + require.Error(t, err) +} + +// TestService_PutClusterCapacityProviders_PersistsStrategy asserts the +// strategy round-trips onto the cluster record and is returned by +// DescribeClusters, without any scheduler/placement coupling (inert in v1). +func TestService_PutClusterCapacityProviders_PersistsStrategy(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + + out, err := svc.PutClusterCapacityProviders(ctx, &ecs.PutClusterCapacityProvidersInput{ + Cluster: aws.String("web"), + CapacityProviders: []*string{aws.String("cp-1"), aws.String("cp-2")}, + DefaultCapacityProviderStrategy: []*ecs.CapacityProviderStrategyItem{ + {CapacityProvider: aws.String("cp-1"), Weight: aws.Int64(1), Base: aws.Int64(2)}, + {CapacityProvider: aws.String("cp-2"), Weight: aws.Int64(3)}, + }, + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Cluster.CapacityProviders, 2) + require.Len(t, out.Cluster.DefaultCapacityProviderStrategy, 2) + assert.Equal(t, "cp-1", aws.StringValue(out.Cluster.DefaultCapacityProviderStrategy[0].CapacityProvider)) + assert.Equal(t, int64(2), aws.Int64Value(out.Cluster.DefaultCapacityProviderStrategy[0].Base)) + + // Persisted onto the cluster record itself, visible via DescribeClusters. + described, err := svc.DescribeClusters(ctx, &ecs.DescribeClustersInput{Clusters: []*string{aws.String("web")}}, testAccountID) + require.NoError(t, err) + require.Len(t, described.Clusters, 1) + require.Len(t, described.Clusters[0].CapacityProviders, 2) + require.Len(t, described.Clusters[0].DefaultCapacityProviderStrategy, 2) +} + +func TestService_PutClusterCapacityProviders_UnknownCluster(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.PutClusterCapacityProviders(context.Background(), &ecs.PutClusterCapacityProvidersInput{ + Cluster: aws.String("ghost"), + CapacityProviders: []*string{aws.String("cp-1")}, + }, testAccountID) + require.Error(t, err) +} diff --git a/spinifex/handlers/ecs/conv.go b/spinifex/handlers/ecs/conv.go index 24c7321b4..3ff2b2a69 100644 --- a/spinifex/handlers/ecs/conv.go +++ b/spinifex/handlers/ecs/conv.go @@ -48,6 +48,16 @@ func taskShortID(ref string) string { return ref } +// capacityProviderShortName extracts the capacity-provider name from a name +// or a capacity-provider ARN. +func capacityProviderShortName(ref string) string { + ref = strings.TrimSpace(ref) + if i := strings.LastIndex(ref, "capacity-provider/"); i >= 0 { + return ref[i+len("capacity-provider/"):] + } + return ref +} + // containerDefsFromAWS maps SDK container definitions to the persisted subset. func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef { out := make([]ContainerDef, 0, len(in)) diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index 7bfbe3275..93fa4c8aa 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -55,6 +55,9 @@ func (s *Service) RegisterContainerInstance(_ context.Context, input *ecs.Regist r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue)) } } + if len(input.Tags) > 0 { + r.Tags = tagsToMap(input.Tags) + } // The agent heartbeats by re-registering. A re-register from a reaped // (involuntarily drained) instance proves the agent is back, so restore // it to ACTIVE. An operator drain (Reaped=false) is left untouched. @@ -157,6 +160,7 @@ func (s *Service) instanceToAWS(r *InstanceRecord) *ecs.ContainerInstance { RegisteredResources: registered, RemainingResources: remaining, VersionInfo: &ecs.VersionInfo{AgentVersion: aws.String(r.AgentVersion)}, + Tags: tagsToAWS(r.Tags), } } diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 69753e577..e77d4374b 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -50,6 +50,9 @@ const ( // circuitBreakerFailureThreshold trips the breaker once this many of the // primary deployment's task launches stop before ever reaching RUNNING. circuitBreakerFailureThreshold = 3 + + CapacityProviderStatusActive = "ACTIVE" + CapacityProviderStatusInactive = "INACTIVE" ) // Deployment is one rollout of a service's task definition. A service has exactly @@ -94,6 +97,10 @@ func ServiceARN(region, accountID, cluster, name string) string { return fmt.Sprintf("arn:aws:ecs:%s:%s:service/%s/%s", region, accountID, cluster, name) } +func CapacityProviderARN(region, accountID, name string) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:capacity-provider/%s", region, accountID, name) +} + // serviceTaskGroup is the AWS task-group label a service stamps on its tasks // ("service:{name}"). The reconciler counts a service's tasks by this group and // the task-state hook resolves a task back to its owning service through it. @@ -118,6 +125,52 @@ type ClusterRecord struct { Status string `json:"status"` Tags map[string]string `json:"tags,omitempty"` CreatedAt time.Time `json:"createdAt"` + // CapacityProviders / DefaultCapacityProviderStrategy are accepted and + // persisted by PutClusterCapacityProviders but are otherwise inert in v1: + // no scheduler coupling, no scale loop (a separate follow-on binds them to + // an ASG primitive). + CapacityProviders []string `json:"capacityProviders,omitempty"` + DefaultCapacityProviderStrategy []CapacityProviderStrategyItem `json:"defaultCapacityProviderStrategy,omitempty"` +} + +// CapacityProviderStrategyItem is one entry of a capacity-provider strategy: +// how many tasks (Base) and what share of the remainder (Weight) a named +// capacity provider takes. Persisted verbatim; not consulted by placement. +type CapacityProviderStrategyItem struct { + Provider string `json:"provider"` + Weight int `json:"weight,omitempty"` + Base int `json:"base,omitempty"` +} + +// ManagedScalingRecord is the persisted subset of an AutoScalingGroupProvider's +// managed-scaling configuration. Accepted and stored; no scale loop reads it. +type ManagedScalingRecord struct { + Status string `json:"status,omitempty"` + TargetCapacity int `json:"targetCapacity,omitempty"` + MinimumScalingStepSize int `json:"minimumScalingStepSize,omitempty"` + MaximumScalingStepSize int `json:"maximumScalingStepSize,omitempty"` + InstanceWarmupPeriod int `json:"instanceWarmupPeriod,omitempty"` +} + +// AutoScalingGroupProviderRecord is the persisted ASG binding for a capacity +// provider. Spinifex has no ASG primitive in v1, so this is stored for API +// parity only; nothing reads AutoScalingGroupARN to launch/terminate capacity. +type AutoScalingGroupProviderRecord struct { + AutoScalingGroupARN string `json:"autoScalingGroupArn"` + ManagedScaling ManagedScalingRecord `json:"managedScaling,omitzero"` + ManagedTerminationProtection string `json:"managedTerminationProtection,omitempty"` +} + +// CapacityProviderRecord is the persisted capacity provider at +// CapacityProviderKey. Account-scoped (not cluster-scoped), matching the AWS +// ARN shape; a cluster references providers by name via CapacityProviders. +type CapacityProviderRecord struct { + Name string `json:"name"` + ARN string `json:"arn"` + Status string `json:"status"` + AutoScalingGroupProvider AutoScalingGroupProviderRecord `json:"autoScalingGroupProvider"` + Tags map[string]string `json:"tags,omitempty"` + CreatedAt time.Time `json:"createdAt"` } // ContainerDef is the persisted subset of an ecs.ContainerDefinition needed to @@ -144,17 +197,18 @@ const LogDriverJSONFile = "json-file" // TaskDefRecord is the persisted task definition revision at TaskDefRevKey. type TaskDefRecord struct { - Family string `json:"family"` - Revision int `json:"revision"` - ARN string `json:"arn"` - NetworkMode string `json:"networkMode,omitempty"` - CPU string `json:"cpu,omitempty"` - Memory string `json:"memory,omitempty"` - TaskRoleArn string `json:"taskRoleArn,omitempty"` - ExecutionRoleArn string `json:"executionRoleArn,omitempty"` - Containers []ContainerDef `json:"containers"` - Status string `json:"status"` - RegisteredAt time.Time `json:"registeredAt"` + Family string `json:"family"` + Revision int `json:"revision"` + ARN string `json:"arn"` + NetworkMode string `json:"networkMode,omitempty"` + CPU string `json:"cpu,omitempty"` + Memory string `json:"memory,omitempty"` + TaskRoleArn string `json:"taskRoleArn,omitempty"` + ExecutionRoleArn string `json:"executionRoleArn,omitempty"` + Containers []ContainerDef `json:"containers"` + Status string `json:"status"` + Tags map[string]string `json:"tags,omitempty"` + RegisteredAt time.Time `json:"registeredAt"` } // reservedCPU/reservedMemory sum the task definition's per-container reservations @@ -180,14 +234,15 @@ func (t *TaskDefRecord) reservedMemory() int { // scheduler writes it from the Layer-2 bus (register/heartbeat) and reserves // capacity by appending placed task IDs. type InstanceRecord struct { - InstanceID string `json:"instanceId"` - ARN string `json:"arn"` - Cluster string `json:"cluster"` - AZ string `json:"availabilityZone,omitempty"` - Hostname string `json:"hostname,omitempty"` - Status string `json:"status"` - TotalCPU int `json:"totalCpu"` - TotalMemoryMiB int `json:"totalMemoryMiB"` + InstanceID string `json:"instanceId"` + ARN string `json:"arn"` + Cluster string `json:"cluster"` + AZ string `json:"availabilityZone,omitempty"` + Hostname string `json:"hostname,omitempty"` + Status string `json:"status"` + Tags map[string]string `json:"tags,omitempty"` + TotalCPU int `json:"totalCpu"` + TotalMemoryMiB int `json:"totalMemoryMiB"` // ReservedCPU/ReservedMemoryMiB track capacity committed to placed tasks; // placement increments them under a KV CAS and the task-state STOPPED path // releases them. Remaining = Total - Reserved. @@ -221,19 +276,20 @@ type TaskRecord struct { // Group / StartedBy mirror the AWS task fields. A service's tasks carry // Group="service:{name}" so the reconciler counts them and the task-state // hook resolves a RUNNING/STOPPED task back to its owning service. - Group string `json:"group,omitempty"` - StartedBy string `json:"startedBy,omitempty"` - TaskDefFamily string `json:"taskDefFamily"` - TaskDefRevision int `json:"taskDefRevision"` - TaskDefARN string `json:"taskDefArn"` - ContainerInstanceID string `json:"containerInstanceId,omitempty"` - ContainerInstanceARN string `json:"containerInstanceArn,omitempty"` - DesiredStatus string `json:"desiredStatus"` - LastStatus string `json:"lastStatus"` - StoppedReason string `json:"stoppedReason,omitempty"` - ReservedCPU int `json:"reservedCpu"` - ReservedMemoryMiB int `json:"reservedMemoryMiB"` - Containers []ContainerState `json:"containers,omitempty"` + Group string `json:"group,omitempty"` + StartedBy string `json:"startedBy,omitempty"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + TaskDefARN string `json:"taskDefArn"` + ContainerInstanceID string `json:"containerInstanceId,omitempty"` + ContainerInstanceARN string `json:"containerInstanceArn,omitempty"` + DesiredStatus string `json:"desiredStatus"` + LastStatus string `json:"lastStatus"` + StoppedReason string `json:"stoppedReason,omitempty"` + ReservedCPU int `json:"reservedCpu"` + ReservedMemoryMiB int `json:"reservedMemoryMiB"` + Tags map[string]string `json:"tags,omitempty"` + Containers []ContainerState `json:"containers,omitempty"` // NetworkMode is the resolved task network mode (awsvpc|bridge|host). The // STOPPED path consults it to decide whether an ENI must be reclaimed. NetworkMode string `json:"networkMode,omitempty"` @@ -286,6 +342,7 @@ type ServiceRecord struct { DeploymentID string `json:"deploymentId"` RunningCount int `json:"runningCount"` PendingCount int `json:"pendingCount"` + Tags map[string]string `json:"tags,omitempty"` // Rolling-update configuration (deploymentConfiguration) and its live state. // MinimumHealthyPercent / MaximumPercent gate the rollout; the circuit breaker // trips a failing deployment and optionally rolls back to LastGoodTaskDefARN. diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index 90111bd08..67a39540a 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -59,6 +59,20 @@ type ECSService interface { // ProvisionCapacity launches container-instance EC2 capacity into a cluster. ProvisionCapacity(ctx context.Context, input *ProvisionCapacityInput, accountID string) (*ProvisionCapacityOutput, error) + + // Tags. Dispatched on the resourceArn shape (ecs-v1.md ยง1); the ACM inline-tags + // pattern (map stored on the record, mutated in place). + ListTagsForResource(ctx context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error) + TagResource(ctx context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error) + UntagResource(ctx context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error) + + // Capacity providers. Strategy is accepted and persisted but inert: no + // scheduler coupling, no scale loop (a separate follow-on binds this to an + // ASG primitive). + PutClusterCapacityProviders(ctx context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error) + CreateCapacityProvider(ctx context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error) + DescribeCapacityProviders(ctx context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error) + DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error) } // ecsImageResolver is the narrow AMI surface for resolving the spinifex-ecs-node @@ -306,12 +320,19 @@ func (s *Service) ListClusters(_ context.Context, _ *ecs.ListClustersInput, acco } func (r *ClusterRecord) toAWS() *ecs.Cluster { - return &ecs.Cluster{ + c := &ecs.Cluster{ ClusterName: aws.String(r.Name), ClusterArn: aws.String(r.ARN), Status: aws.String(r.Status), Tags: tagsToAWS(r.Tags), } + if len(r.CapacityProviders) > 0 { + c.CapacityProviders = aws.StringSlice(r.CapacityProviders) + } + for _, item := range r.DefaultCapacityProviderStrategy { + c.DefaultCapacityProviderStrategy = append(c.DefaultCapacityProviderStrategy, item.toAWS()) + } + return c } // tagsToAWS converts a stored tag map into the AWS list form with a stable @@ -328,6 +349,25 @@ func tagsToAWS(tags map[string]string) []*ecs.Tag { return out } +// tagsToMap is the inverse of tagsToAWS: it converts the AWS tag list form +// into a stored map, dropping nil entries and empty keys. +func tagsToMap(tags []*ecs.Tag) map[string]string { + if len(tags) == 0 { + return nil + } + out := make(map[string]string, len(tags)) + for _, t := range tags { + if t == nil || aws.StringValue(t.Key) == "" { + continue + } + out[aws.StringValue(t.Key)] = aws.StringValue(t.Value) + } + if len(out) == 0 { + return nil + } + return out +} + // --- Task definition --- // RegisterTaskDefinition stores a new revision of a family, bumping latest-rev. @@ -360,6 +400,7 @@ func (s *Service) RegisterTaskDefinition(ctx context.Context, input *ecs.Registe TaskRoleArn: aws.StringValue(input.TaskRoleArn), ExecutionRoleArn: aws.StringValue(input.ExecutionRoleArn), Status: TaskDefStatusActive, + Tags: tagsToMap(input.Tags), RegisteredAt: time.Now().UTC(), Containers: containerDefsFromAWS(input.ContainerDefinitions), } @@ -369,7 +410,7 @@ func (s *Service) RegisterTaskDefinition(ctx context.Context, input *ecs.Registe if err := putJSON(kv, TaskDefLatestRevKey(family), rev); err != nil { return nil, err } - return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil + return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS(), Tags: tagsToAWS(rec.Tags)}, nil } // validateContainerDefs hard-rejects taskdef features the data plane cannot honor. @@ -451,7 +492,7 @@ func (s *Service) DescribeTaskDefinition(_ context.Context, input *ecs.DescribeT if err != nil { return nil, err } - return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil + return &ecs.DescribeTaskDefinitionOutput{TaskDefinition: rec.toAWS(), Tags: tagsToAWS(rec.Tags)}, nil } // ListTaskDefinitions returns revision ARNs, optionally filtered by family and diff --git a/spinifex/handlers/ecs/service_nats.go b/spinifex/handlers/ecs/service_nats.go index 0e0efecc5..6eaf7ef6b 100644 --- a/spinifex/handlers/ecs/service_nats.go +++ b/spinifex/handlers/ecs/service_nats.go @@ -139,3 +139,35 @@ func (s *NATSECSService) PollAssignments(ctx context.Context, input *PollAssignm func (s *NATSECSService) ProvisionCapacity(ctx context.Context, input *ProvisionCapacityInput, accountID string) (*ProvisionCapacityOutput, error) { return utils.NATSRequest[ProvisionCapacityOutput](ctx, s.natsConn, "ecs.ProvisionCapacity", input, defaultTimeout, accountID) } + +// --- Tags --- + +func (s *NATSECSService) ListTagsForResource(ctx context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error) { + return utils.NATSRequest[ecs.ListTagsForResourceOutput](ctx, s.natsConn, "ecs.ListTagsForResource", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) TagResource(ctx context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error) { + return utils.NATSRequest[ecs.TagResourceOutput](ctx, s.natsConn, "ecs.TagResource", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) UntagResource(ctx context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error) { + return utils.NATSRequest[ecs.UntagResourceOutput](ctx, s.natsConn, "ecs.UntagResource", input, defaultTimeout, accountID) +} + +// --- Capacity providers --- + +func (s *NATSECSService) PutClusterCapacityProviders(ctx context.Context, input *ecs.PutClusterCapacityProvidersInput, accountID string) (*ecs.PutClusterCapacityProvidersOutput, error) { + return utils.NATSRequest[ecs.PutClusterCapacityProvidersOutput](ctx, s.natsConn, "ecs.PutClusterCapacityProviders", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) CreateCapacityProvider(ctx context.Context, input *ecs.CreateCapacityProviderInput, accountID string) (*ecs.CreateCapacityProviderOutput, error) { + return utils.NATSRequest[ecs.CreateCapacityProviderOutput](ctx, s.natsConn, "ecs.CreateCapacityProvider", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeCapacityProviders(ctx context.Context, input *ecs.DescribeCapacityProvidersInput, accountID string) (*ecs.DescribeCapacityProvidersOutput, error) { + return utils.NATSRequest[ecs.DescribeCapacityProvidersOutput](ctx, s.natsConn, "ecs.DescribeCapacityProviders", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteCapacityProviderInput, accountID string) (*ecs.DeleteCapacityProviderOutput, error) { + return utils.NATSRequest[ecs.DeleteCapacityProviderOutput](ctx, s.natsConn, "ecs.DeleteCapacityProvider", input, defaultTimeout, accountID) +} diff --git a/spinifex/handlers/ecs/services.go b/spinifex/handlers/ecs/services.go index ab6261deb..090667d6d 100644 --- a/spinifex/handlers/ecs/services.go +++ b/spinifex/handlers/ecs/services.go @@ -175,6 +175,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp PlacementStrategy: placementStrategyFromAWS(input.PlacementStrategy), LoadBalancers: loadBalancersFromAWS(input.LoadBalancers), DeploymentID: uuid.NewString(), + Tags: tagsToMap(input.Tags), CreatedAt: now, UpdatedAt: now, } @@ -612,6 +613,7 @@ func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service PendingCount: aws.Int64(int64(r.PendingCount)), SchedulingStrategy: aws.String(r.SchedulingStrategy), TaskDefinition: aws.String(r.TaskDefARN), + Tags: tagsToAWS(r.Tags), } if r.LaunchType != "" { svc.LaunchType = aws.String(r.LaunchType) diff --git a/spinifex/handlers/ecs/store.go b/spinifex/handlers/ecs/store.go index 93bb5f714..c6d1d6a8b 100644 --- a/spinifex/handlers/ecs/store.go +++ b/spinifex/handlers/ecs/store.go @@ -135,6 +135,18 @@ func LeaderLeaseKey(accountID, clusterName string) string { return fmt.Sprintf("%s/%s", accountID, clusterName) } +// CapacityProvidersPrefix returns the KV key prefix under which all +// capacity-provider records live. Account-scoped (not cluster-scoped), +// matching the AWS ARN shape; a cluster references providers by name. +func CapacityProvidersPrefix() string { + return "capacity-providers/" +} + +// CapacityProviderKey returns the KV key for a named capacity-provider record. +func CapacityProviderKey(name string) string { + return CapacityProvidersPrefix() + name +} + // Store is the per-daemon ECS KV handle. Per-account and leader buckets are // accessed via the package-level factories below. type Store struct { diff --git a/spinifex/handlers/ecs/tags.go b/spinifex/handlers/ecs/tags.go new file mode 100644 index 000000000..7e8c4dda7 --- /dev/null +++ b/spinifex/handlers/ecs/tags.go @@ -0,0 +1,268 @@ +package handlers_ecs + +import ( + "context" + "errors" + "maps" + "strings" + + "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" +) + +// ecsResourceKind identifies which stored record kind an ARN refers to. +type ecsResourceKind int + +const ( + ecsResourceCluster ecsResourceKind = iota + ecsResourceTaskDefinition + ecsResourceService + ecsResourceTask + ecsResourceContainerInstance +) + +// parseResourceARN splits an ECS resource ARN into its kind, owning cluster +// (empty for cluster/task-definition, which are flat), and short id/name. +// Cluster and task-definition ARNs are flat (.../cluster/{name}, +// .../task-definition/{family}:{rev}); service/task/container-instance ARNs +// embed the owning cluster (.../service/{cluster}/{name}). +func parseResourceARN(resourceARN string) (kind ecsResourceKind, cluster, id string, err error) { + // SplitN(..., 6) keeps the resource segment intact even though a + // task-definition id ("family:1") itself contains a colon. + parts := strings.SplitN(resourceARN, ":", 6) + if len(parts) != 6 || parts[0] != "arn" { + return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter) + } + rtype, rest, ok := strings.Cut(parts[5], "/") + if !ok || rtype == "" || rest == "" { + return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter) + } + + switch rtype { + case "cluster": + return ecsResourceCluster, rest, rest, nil + case "task-definition": + return ecsResourceTaskDefinition, "", rest, nil + case "service", "task", "container-instance": + clusterName, name, ok := strings.Cut(rest, "/") + if !ok || clusterName == "" || name == "" { + return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter) + } + switch rtype { + case "service": + return ecsResourceService, clusterName, name, nil + case "task": + return ecsResourceTask, clusterName, name, nil + default: + return ecsResourceContainerInstance, clusterName, name, nil + } + default: + return 0, "", "", errors.New(awserrors.ErrorECSInvalidParameter) + } +} + +// resourceTags reads the tag map stored on the ARN-identified resource. +func (s *Service) resourceTags(kv nats.KeyValue, kind ecsResourceKind, cluster, id string) (map[string]string, error) { + switch kind { + case ecsResourceCluster: + var rec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(id), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSClusterNotFound) + } + return rec.Tags, nil + case ecsResourceTaskDefinition: + family, rev := parseTaskDefRef(id) + if family == "" || rev == 0 { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + 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) + } + return rec.Tags, nil + case ecsResourceService: + var rec ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, id), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSServiceNotFound) + } + return rec.Tags, nil + case ecsResourceTask: + var rec TaskRecord + found, err := getJSON(kv, TaskKey(cluster, id), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + return rec.Tags, nil + case ecsResourceContainerInstance: + 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) + } + return rec.Tags, nil + default: + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } +} + +// mutateResourceTags reads the ARN-identified resource, applies mutate to its +// tag map, and re-puts the record. Used by TagResource (merge) and +// UntagResource (delete). +func (s *Service) mutateResourceTags(kv nats.KeyValue, kind ecsResourceKind, cluster, id string, mutate func(map[string]string) map[string]string) error { + switch kind { + case ecsResourceCluster: + var rec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(id), &rec) + if err != nil { + return err + } + if !found { + return errors.New(awserrors.ErrorECSClusterNotFound) + } + rec.Tags = mutate(rec.Tags) + return putJSON(kv, ClusterMetaKey(id), &rec) + case ecsResourceTaskDefinition: + family, rev := parseTaskDefRef(id) + if family == "" || rev == 0 { + return errors.New(awserrors.ErrorECSInvalidParameter) + } + var rec TaskDefRecord + found, err := getJSON(kv, TaskDefRevKey(family, rev), &rec) + if err != nil { + return err + } + if !found { + return errors.New(awserrors.ErrorECSInvalidParameter) + } + rec.Tags = mutate(rec.Tags) + return putJSON(kv, TaskDefRevKey(family, rev), &rec) + case ecsResourceService: + var rec ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, id), &rec) + if err != nil { + return err + } + if !found { + return errors.New(awserrors.ErrorECSServiceNotFound) + } + rec.Tags = mutate(rec.Tags) + return putJSON(kv, ServiceKey(cluster, id), &rec) + case ecsResourceTask: + var rec TaskRecord + found, err := getJSON(kv, TaskKey(cluster, id), &rec) + if err != nil { + return err + } + if !found { + return errors.New(awserrors.ErrorECSInvalidParameter) + } + rec.Tags = mutate(rec.Tags) + return putJSON(kv, TaskKey(cluster, id), &rec) + case ecsResourceContainerInstance: + var rec InstanceRecord + found, err := getJSON(kv, InstanceKey(cluster, id), &rec) + if err != nil { + return err + } + if !found { + return errors.New(awserrors.ErrorECSInvalidParameter) + } + rec.Tags = mutate(rec.Tags) + return putJSON(kv, InstanceKey(cluster, id), &rec) + default: + return errors.New(awserrors.ErrorECSInvalidParameter) + } +} + +// ListTagsForResource returns the tags stored on the ARN-identified resource. +func (s *Service) ListTagsForResource(_ context.Context, input *ecs.ListTagsForResourceInput, accountID string) (*ecs.ListTagsForResourceOutput, error) { + if input == nil || aws.StringValue(input.ResourceArn) == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn)) + if err != nil { + return nil, err + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + tags, err := s.resourceTags(kv, kind, cluster, id) + if err != nil { + return nil, err + } + return &ecs.ListTagsForResourceOutput{Tags: tagsToAWS(tags)}, nil +} + +// TagResource merges the supplied tags onto the ARN-identified resource. +func (s *Service) TagResource(_ context.Context, input *ecs.TagResourceInput, accountID string) (*ecs.TagResourceOutput, error) { + if input == nil || aws.StringValue(input.ResourceArn) == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn)) + if err != nil { + return nil, err + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + add := tagsToMap(input.Tags) + err = s.mutateResourceTags(kv, kind, cluster, id, func(existing map[string]string) map[string]string { + if existing == nil { + existing = map[string]string{} + } + maps.Copy(existing, add) + return existing + }) + if err != nil { + return nil, err + } + return &ecs.TagResourceOutput{}, nil +} + +// UntagResource deletes the named tag keys from the ARN-identified resource. +func (s *Service) UntagResource(_ context.Context, input *ecs.UntagResourceInput, accountID string) (*ecs.UntagResourceOutput, error) { + if input == nil || aws.StringValue(input.ResourceArn) == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + kind, cluster, id, err := parseResourceARN(aws.StringValue(input.ResourceArn)) + if err != nil { + return nil, err + } + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + keys := awsStringSlice(input.TagKeys) + err = s.mutateResourceTags(kv, kind, cluster, id, func(existing map[string]string) map[string]string { + for _, k := range keys { + delete(existing, k) + } + return existing + }) + if err != nil { + return nil, err + } + return &ecs.UntagResourceOutput{}, nil +} diff --git a/spinifex/handlers/ecs/tags_test.go b/spinifex/handlers/ecs/tags_test.go new file mode 100644 index 000000000..2171582a5 --- /dev/null +++ b/spinifex/handlers/ecs/tags_test.go @@ -0,0 +1,229 @@ +package handlers_ecs + +import ( + "context" + "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 TestParseResourceARN(t *testing.T) { + cases := []struct { + name string + arn string + wantKind ecsResourceKind + wantCluster string + wantID string + }{ + {"cluster", "arn:aws:ecs:ap-southeast-2:123456789012:cluster/web", ecsResourceCluster, "web", "web"}, + {"task-definition", "arn:aws:ecs:ap-southeast-2:123456789012:task-definition/nginx:3", ecsResourceTaskDefinition, "", "nginx:3"}, + {"service", "arn:aws:ecs:ap-southeast-2:123456789012:service/web/api", ecsResourceService, "web", "api"}, + {"task", "arn:aws:ecs:ap-southeast-2:123456789012:task/web/t-1", ecsResourceTask, "web", "t-1"}, + {"container-instance", "arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/i-1", ecsResourceContainerInstance, "web", "i-1"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + kind, cluster, id, err := parseResourceARN(tc.arn) + require.NoError(t, err) + assert.Equal(t, tc.wantKind, kind) + assert.Equal(t, tc.wantCluster, cluster) + assert.Equal(t, tc.wantID, id) + }) + } +} + +func TestParseResourceARN_Invalid(t *testing.T) { + cases := []string{ + "", + "not-an-arn", + "arn:aws:ecs:r:a:cluster", // no "/" after resource type + "arn:aws:ecs:r:a:service/onlyone", // service missing embedded name + "arn:aws:ecs:r:a:unknown/x", // unrecognized resource type + } + for _, arnStr := range cases { + t.Run(arnStr, func(t *testing.T) { + _, _, _, err := parseResourceARN(arnStr) + assert.EqualError(t, err, "InvalidParameterException") + }) + } +} + +// TestService_Tags_ClusterRoundTrip covers the flat ARN shape end to end via +// the wired TagResource/ListTagsForResource/UntagResource actions. +func TestService_Tags_ClusterRoundTrip(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + created, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + arn := aws.StringValue(created.Cluster.ClusterArn) + + _, err = svc.TagResource(ctx, &ecs.TagResourceInput{ + ResourceArn: aws.String(arn), + Tags: []*ecs.Tag{ + {Key: aws.String("team"), Value: aws.String("infra")}, + {Key: aws.String("env"), Value: aws.String("prod")}, + }, + }, testAccountID) + require.NoError(t, err) + + listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, listed.Tags, 2) + + // Merge in an extra tag alongside the existing pair. + _, err = svc.TagResource(ctx, &ecs.TagResourceInput{ + ResourceArn: aws.String(arn), + Tags: []*ecs.Tag{{Key: aws.String("owner"), Value: aws.String("platform")}}, + }, testAccountID) + require.NoError(t, err) + + listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, listed.Tags, 3) + + _, err = svc.UntagResource(ctx, &ecs.UntagResourceInput{ + ResourceArn: aws.String(arn), + TagKeys: []*string{aws.String("env")}, + }, testAccountID) + require.NoError(t, err) + + listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, listed.Tags, 2) + for _, tag := range listed.Tags { + assert.NotEqual(t, "env", aws.StringValue(tag.Key)) + } +} + +// TestService_Tags_ServiceRoundTrip covers the cluster-embedded ARN shape. +func TestService_Tags_ServiceRoundTrip(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "nginx", 128, 256) + + created, err := svc.CreateService(ctx, &ecs.CreateServiceInput{ + Cluster: aws.String("web"), + ServiceName: aws.String("api"), + TaskDefinition: aws.String("nginx"), + DesiredCount: aws.Int64(0), + }, testAccountID) + require.NoError(t, err) + arn := aws.StringValue(created.Service.ServiceArn) + + _, err = svc.TagResource(ctx, &ecs.TagResourceInput{ + ResourceArn: aws.String(arn), + Tags: []*ecs.Tag{{Key: aws.String("team"), Value: aws.String("infra")}}, + }, testAccountID) + require.NoError(t, err) + + listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, listed.Tags, 1) + assert.Equal(t, "team", aws.StringValue(listed.Tags[0].Key)) + + _, err = svc.UntagResource(ctx, &ecs.UntagResourceInput{ + ResourceArn: aws.String(arn), + TagKeys: []*string{aws.String("team")}, + }, testAccountID) + require.NoError(t, err) + + listed, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + assert.Empty(t, listed.Tags) +} + +// TestService_Tags_TaskDefinitionRoundTrip covers the flat family:rev ARN shape. +func TestService_Tags_TaskDefinitionRoundTrip(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + out := registerTaskDef(t, svc, "nginx", 128, 256) + arn := aws.StringValue(out.TaskDefinition.TaskDefinitionArn) + + _, err := svc.TagResource(ctx, &ecs.TagResourceInput{ + ResourceArn: aws.String(arn), + Tags: []*ecs.Tag{{Key: aws.String("env"), Value: aws.String("prod")}}, + }, testAccountID) + require.NoError(t, err) + + listed, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ResourceArn: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, listed.Tags, 1) + + described, err := svc.DescribeTaskDefinition(ctx, &ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String(arn)}, testAccountID) + require.NoError(t, err) + require.Len(t, described.Tags, 1) + assert.Equal(t, "env", aws.StringValue(described.Tags[0].Key)) +} + +// TestService_Tags_MissingResource asserts the not-found mapping the plan +// specifies: cluster/service resolve to their named NotFound errors; task and +// container-instance resolve to the generic InvalidParameter. +func TestService_Tags_MissingResource(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ + ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:cluster/nope"), + }, testAccountID) + assert.EqualError(t, err, "ClusterNotFoundException") + + _, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ + ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:service/web/nope"), + }, testAccountID) + assert.EqualError(t, err, "ServiceNotFoundException") + + _, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ + ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:task/web/nope"), + }, testAccountID) + assert.EqualError(t, err, "InvalidParameterException") + + _, err = svc.ListTagsForResource(ctx, &ecs.ListTagsForResourceInput{ + ResourceArn: aws.String("arn:aws:ecs:ap-southeast-2:123456789012:container-instance/web/nope"), + }, testAccountID) + assert.EqualError(t, err, "InvalidParameterException") +} + +// TestService_RunTask_TagsPersisted confirms RunTask plumbs input.Tags onto +// the persisted task record and DescribeTasks reflects them back. +func TestService_RunTask_TagsPersisted(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.CreateCluster(ctx, &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "nginx", 128, 256) + _, err = svc.RegisterContainerInstance(ctx, &ecs.RegisterContainerInstanceInput{ + Cluster: aws.String("web"), + TotalResources: []*ecs.Resource{ + {Name: aws.String("CPU"), IntegerValue: aws.Int64(1024)}, + {Name: aws.String("MEMORY"), IntegerValue: aws.Int64(2048)}, + }, + }, testAccountID) + require.NoError(t, err) + + out, err := svc.RunTask(ctx, &ecs.RunTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("nginx"), + Tags: []*ecs.Tag{{Key: aws.String("env"), Value: aws.String("prod")}}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Tasks, 1) + require.Len(t, out.Tasks[0].Tags, 1) + assert.Equal(t, "env", aws.StringValue(out.Tasks[0].Tags[0].Key)) + + described, err := svc.DescribeTasks(ctx, &ecs.DescribeTasksInput{ + Cluster: aws.String("web"), + Tasks: []*string{out.Tasks[0].TaskArn}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, described.Tasks, 1) + require.Len(t, described.Tasks[0].Tags, 1) +} diff --git a/spinifex/handlers/ecs/task_stop.go b/spinifex/handlers/ecs/task_stop.go index fadde25a6..7dc212658 100644 --- a/spinifex/handlers/ecs/task_stop.go +++ b/spinifex/handlers/ecs/task_stop.go @@ -93,6 +93,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco rec.NetworkMode = mode rec.Group = group rec.StartedBy = startedBy + rec.Tags = tagsToMap(input.Tags) if mode == NetworkModeAwsvpc { if failure := s.provisionTaskENI(ctx, kv, accountID, cluster, rec, netCfg); failure != nil { out.Failures = append(out.Failures, failure) diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go index db451c054..524e8e460 100644 --- a/spinifex/handlers/ecs/tasks.go +++ b/spinifex/handlers/ecs/tasks.go @@ -74,6 +74,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI rec.NetworkMode = mode rec.Group = aws.StringValue(input.Group) rec.StartedBy = aws.StringValue(input.StartedBy) + rec.Tags = tagsToMap(input.Tags) if mode == NetworkModeAwsvpc { if failure := s.provisionTaskENI(ctx, kv, accountID, cluster, rec, netCfg); failure != nil { out.Failures = append(out.Failures, failure) @@ -271,6 +272,7 @@ func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task { DesiredStatus: aws.String(r.DesiredStatus), LastStatus: aws.String(r.LastStatus), ContainerInstanceArn: aws.String(r.ContainerInstanceARN), + Tags: tagsToAWS(r.Tags), } if r.Group != "" { t.Group = aws.String(r.Group)