diff --git a/cmd/ecs-agent/credendpoint.go b/cmd/ecs-agent/credendpoint.go index 2fb863a28..526c92661 100644 --- a/cmd/ecs-agent/credendpoint.go +++ b/cmd/ecs-agent/credendpoint.go @@ -192,6 +192,45 @@ func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Creden return creds, roleARN, nil } +// AssumeProvider returns a CredentialsProvider that mints credentials by assuming +// roleARN over the gateway — the execution-role path for ECR image pulls. It +// caches the assumed set and re-assumes within the refresh margin. session scopes +// the STS session name. +func (c *credEndpoint) AssumeProvider(roleARN, session string) credentials.CredentialsProvider { + return &assumeProvider{ep: c, roleARN: roleARN, session: session} +} + +// assumeProvider adapts a credEndpoint's AssumeRole path to the agent's +// CredentialsProvider interface, caching the assumed credentials. +type assumeProvider struct { + ep *credEndpoint + roleARN string + session string + + mu sync.Mutex + cached stsauth.Credentials +} + +var _ credentials.CredentialsProvider = (*assumeProvider)(nil) + +func (p *assumeProvider) Retrieve(ctx context.Context) (credentials.Credentials, error) { + p.mu.Lock() + defer p.mu.Unlock() + if !credValid(p.cached, credRefreshMargin) { + creds, err := p.ep.assume(ctx, p.roleARN, p.session) + if err != nil { + return credentials.Credentials{}, err + } + p.cached = creds + } + return credentials.Credentials{ + AccessKeyID: p.cached.AccessKeyID, + SecretAccessKey: p.cached.SecretAccessKey, + SessionToken: p.cached.SessionToken, + Expiration: p.cached.Expiration, + }, nil +} + // assumeOverGateway mints credentials for roleARN by SigV4-signing AssumeRole // against the gateway with the agent's instance credentials. func (c *credEndpoint) assumeOverGateway(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) { diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go index c9e752d42..906626262 100644 --- a/cmd/ecs-agent/taskrun.go +++ b/cmd/ecs-agent/taskrun.go @@ -65,10 +65,11 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { } credID := a.registerTaskCreds(as) + resolver := a.pullResolver(as) statuses := make([]bus.ContainerStatus, 0, len(as.Containers)) for _, c := range as.Containers { - if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, a.resolver); err != nil { + if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil { slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err) a.teardownTaskNetns(as) a.reportTaskState(as, bus.TaskStatusStopped, "image pull failed: "+err.Error(), statuses) @@ -137,6 +138,18 @@ func (a *Agent) teardownTaskNetns(as *bus.Assign) { } } +// pullResolver returns the ECR resolver used for a task's image pulls. When the +// assign carries an execution role, it authorizes pulls as that role (assumed +// over the gateway); an empty role or a nil credential endpoint (unit tests) +// falls back to the instance-role resolver, so a pull is never worse off. +func (a *Agent) pullResolver(as *bus.Assign) ctrruntime.Resolver { + if as.ExecutionRoleARN == "" || a.cred == nil { + return a.resolver + } + prov := a.cred.AssumeProvider(as.ExecutionRoleARN, sessionName(as.TaskID)) + return newLazyECRResolver(prov, a.cfg.Region, a.cfg.GatewayURL, a.cfg.GatewayCA) +} + // registerTaskCreds registers the task's credID -> role mapping with the // credential endpoint and returns the credID (empty when the task has no role). func (a *Agent) registerTaskCreds(as *bus.Assign) string { diff --git a/cmd/ecs-agent/taskrun_test.go b/cmd/ecs-agent/taskrun_test.go index e88eec539..bba22854a 100644 --- a/cmd/ecs-agent/taskrun_test.go +++ b/cmd/ecs-agent/taskrun_test.go @@ -227,3 +227,32 @@ func TestContainerID(t *testing.T) { t.Errorf("containerID = %q, want t-1-web", got) } } + +// sentinelResolver is an identity-comparable Resolver for pullResolver tests. +type sentinelResolver struct{} + +func (sentinelResolver) Authorize(context.Context, string) (string, string, string, error) { + return "", "", "", nil +} + +// TestPullResolver covers siv-459: an assign with no execution role (or no +// credential endpoint) uses the instance resolver; with both, a distinct +// exec-role resolver is built. +func TestPullResolver(t *testing.T) { + fake := sentinelResolver{} + a := newAgent(config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, nil, fake) + + if got := a.pullResolver(&bus.Assign{TaskID: "t1"}); got != fake { + t.Fatalf("no execution role: want instance resolver fallback") + } + execAssign := &bus.Assign{TaskID: "t1", ExecutionRoleARN: "arn:aws:iam::1:role/e"} + if got := a.pullResolver(execAssign); got != fake { + t.Fatalf("nil cred endpoint: want instance resolver fallback") + } + + a.cred = &credEndpoint{} + got := a.pullResolver(execAssign) + if got == nil || got == fake { + t.Fatalf("execution role + cred endpoint: want distinct exec-role resolver, got %v", got) + } +} diff --git a/docs/containers/ecs/README.md b/docs/containers/ecs/README.md index faf28f3ba..d9d022df9 100644 --- a/docs/containers/ecs/README.md +++ b/docs/containers/ecs/README.md @@ -62,11 +62,32 @@ A **task definition** describes one or more containers (image, CPU/memory, ports ECS v1 is deliberately minimal. Keep these in mind: - **No service discovery.** `serviceRegistries` / Cloud Map integration is not implemented — reach a service through its load balancer, not a DNS name. -- **`awslogs` log driver is discarded.** Container stdout/stderr is not shipped to CloudWatch Logs; the log configuration is accepted but dropped. -- **No rolling deployments.** The reconciler keeps the desired count running but ignores `deploymentConfiguration` (`minimumHealthyPercent` / `maximumPercent`); an `UpdateService` to a new revision replaces tasks without a health-gated rollout or circuit-breaker. +- **Only the `json-file` log driver is collected.** Container stdout/stderr is captured host-side on the container instance (see [Logging](#logging)); it is not shipped to CloudWatch Logs. Any other driver (e.g. `awslogs`) is accepted for parity but its logs are discarded — `RegisterTaskDefinition` logs a warning naming the container so the drop is not silent. - **No capacity providers or managed scaling.** Capacity is the static total of your registered instances; there is no scale-out/in or ASG binding — you manage the instance count. -- **No distinct execution role.** ECR pulls use the instance role directly; `executionRoleArn` is not honoured separately. -- **`secrets[]` are silently dropped**, and tags set via `TagResource` are not persisted. Health-check settings beyond the target group default are not forwarded. +- **`secrets[]` are rejected.** A task definition that declares container `secrets[]` fails `RegisterTaskDefinition` with `InvalidParameterException` rather than running without the secrets it expects. Tags set via `TagResource` are not persisted, and health-check settings beyond the target group default are not forwarded. + +### Logging + +Spinifex honours the host-side **`json-file`** log driver (the containerd default). A container's stdout/stderr is written on its container instance and retrievable there — no CloudWatch Logs path exists. + +To read a task's logs, find the container instance running it (`aws ecs describe-tasks` → `containerInstanceArn` → the EC2 instance), then on that host inspect the containerd task output. Containers are named `{taskId}-{containerName}` and carry `mulga.ecs.*` labels: + +```bash +# On the container instance: +ctr -n default containers ls # find {taskId}-{containerName} +ctr -n default tasks ls +journalctl -u containerd | grep # container stdout/stderr via the host journal +``` + +A task definition may still declare `awslogs` (or any other driver) for AWS compatibility — Spinifex accepts it, warns at registration that the driver is not implemented, and falls back to the host-side `json-file` behaviour. + +### Deployments + +An `UpdateService` to a new task-definition revision performs a **health-gated rolling update** honouring `deploymentConfiguration`: `minimumHealthyPercent` keeps that fraction of the desired count running while `maximumPercent` bounds how many extra tasks launch during the roll. Enabling the **deployment circuit breaker** fails a rollout whose tasks repeatedly fail to start, and — with `rollback` set — automatically reverts to the last-good task definition. + +### Execution role + +If a task definition sets `executionRoleArn`, the agent assumes that role to authorise ECR image pulls (instead of the container-instance role). When it is unset, pulls fall back to the instance role. ## Prerequisites @@ -267,4 +288,4 @@ tofu apply -var 'gateway_url=https://:9999' **Container cannot assume its task role.** Confirm the task definition sets `taskRoleArn` and the role is trusted by `ecs-tasks.amazonaws.com`. The agent injects `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI`; a container that overrides this variable, or an SDK too old to honour it, will not pick up the credentials. -**Expected a feature that is missing.** ECS v1 omits service discovery, `awslogs` shipping, rolling deployments, capacity providers, and a distinct execution role. See the Limitations in the Overview. +**Expected a feature that is missing.** ECS v1 omits service discovery, CloudWatch Logs (`awslogs`) shipping, and capacity providers / managed scaling. See the Limitations in the Overview. diff --git a/spinifex/handlers/ecs/bus/messages.go b/spinifex/handlers/ecs/bus/messages.go index 6835bc7d6..2d564ad8b 100644 --- a/spinifex/handlers/ecs/bus/messages.go +++ b/spinifex/handlers/ecs/bus/messages.go @@ -65,6 +65,9 @@ type AssignContainer struct { Command []string `json:"command,omitempty"` Environment map[string]string `json:"environment,omitempty"` PortMappings []PortMapping `json:"portMappings,omitempty"` + // LogDriver is the requested container log driver; only json-file is honored + // (host-side). Any other value means logs are discarded (warned at register). + LogDriver string `json:"logDriver,omitempty"` } // Assign is published on AssignSubject when the scheduler places a task on this @@ -89,9 +92,13 @@ type Assign struct { // TaskRoleARN is the task's IAM role (from the taskdef); when set, the agent // serves its credentials at 169.254.170.2 under CredID, defaulting CredID to // the taskID when the scheduler leaves it empty. - CredID string `json:"credId,omitempty"` - TaskRoleARN string `json:"taskRoleArn,omitempty"` - AssignedAt time.Time `json:"assignedAt"` + CredID string `json:"credId,omitempty"` + TaskRoleARN string `json:"taskRoleArn,omitempty"` + // ExecutionRoleARN is the task's execution role (from the taskdef); when set, + // the agent assumes it to authorize ECR image pulls instead of the container- + // instance role. Empty falls back to the instance role. + ExecutionRoleARN string `json:"executionRoleArn,omitempty"` + AssignedAt time.Time `json:"assignedAt"` } // ContainerStatus is a single container's reported lifecycle state. diff --git a/spinifex/handlers/ecs/conv.go b/spinifex/handlers/ecs/conv.go index 05a018d69..24c7321b4 100644 --- a/spinifex/handlers/ecs/conv.go +++ b/spinifex/handlers/ecs/conv.go @@ -82,6 +82,15 @@ func containerDefsFromAWS(in []*ecs.ContainerDefinition) []ContainerDef { Protocol: aws.StringValue(p.Protocol), }) } + if lc := c.LogConfiguration; lc != nil { + def.LogDriver = aws.StringValue(lc.LogDriver) + for k, v := range lc.Options { + if def.LogOptions == nil { + def.LogOptions = map[string]string{} + } + def.LogOptions[k] = aws.StringValue(v) + } + } out = append(out, def) } return out @@ -115,6 +124,16 @@ func (c ContainerDef) toAWS() *ecs.ContainerDefinition { } cd.PortMappings = append(cd.PortMappings, pm) } + if c.LogDriver != "" { + lc := &ecs.LogConfiguration{LogDriver: aws.String(c.LogDriver)} + if len(c.LogOptions) > 0 { + lc.Options = map[string]*string{} + for k, v := range c.LogOptions { + lc.Options[k] = aws.String(v) + } + } + cd.LogConfiguration = lc + } return cd } @@ -129,5 +148,6 @@ func (c ContainerDef) toAssignContainer() bus.AssignContainer { Command: c.Command, Environment: c.Environment, PortMappings: c.PortMappings, + LogDriver: c.LogDriver, } } diff --git a/spinifex/handlers/ecs/deployments.go b/spinifex/handlers/ecs/deployments.go new file mode 100644 index 000000000..26e422bd8 --- /dev/null +++ b/spinifex/handlers/ecs/deployments.go @@ -0,0 +1,240 @@ +package handlers_ecs + +import ( + "log/slog" + "slices" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/google/uuid" +) + +// deploymentStartedByPrefix is the AWS StartedBy prefix a service stamps on its +// tasks; the suffix is the deployment ID, so the reconciler maps a task back to +// the deployment that launched it. +const deploymentStartedByPrefix = "ecs-svc/" + +// deploymentSupersededReason is the stop reason for an old-deployment task drained +// during a rolling update. +const deploymentSupersededReason = "Superseded by new deployment" + +func deploymentStartedBy(id string) string { return deploymentStartedByPrefix + id } + +func deploymentIDFromStartedBy(startedBy string) string { + id, _ := strings.CutPrefix(startedBy, deploymentStartedByPrefix) + return id +} + +// ceilPercent returns count*percent/100 rounded up — used for minimumHealthyPercent +// (AWS rounds the healthy floor up). +func ceilPercent(count, percent int) int { + if count <= 0 { + return 0 + } + return (count*percent + 99) / 100 +} + +// applyDeploymentConfig sets the service's deploymentConfiguration, defaulting to +// the AWS REPLICA defaults (100% healthy / 200% max) when a field is omitted. +func applyDeploymentConfig(rec *ServiceRecord, dc *ecs.DeploymentConfiguration) { + rec.MinimumHealthyPercent = defaultMinimumHealthyPercent + rec.MaximumPercent = defaultMaximumPercent + if dc == nil { + return + } + if dc.MinimumHealthyPercent != nil { + rec.MinimumHealthyPercent = int(*dc.MinimumHealthyPercent) + } + if dc.MaximumPercent != nil { + rec.MaximumPercent = int(*dc.MaximumPercent) + } + if cb := dc.DeploymentCircuitBreaker; cb != nil { + rec.CircuitBreakerEnable = aws.BoolValue(cb.Enable) + rec.CircuitBreakerRollback = aws.BoolValue(cb.Rollback) + } +} + +// normalizeDeploymentConfig backfills the deploymentConfiguration defaults for a +// legacy service record written before rolling deployments existed. +func (r *ServiceRecord) normalizeDeploymentConfig() { + if r.MinimumHealthyPercent == 0 { + r.MinimumHealthyPercent = defaultMinimumHealthyPercent + } + if r.MaximumPercent == 0 { + r.MaximumPercent = defaultMaximumPercent + } +} + +// newPrimaryDeployment builds an IN_PROGRESS PRIMARY deployment for a task def. +func newPrimaryDeployment(id string, td *TaskDefRecord, desired int) Deployment { + now := time.Now().UTC() + return Deployment{ + ID: id, + Status: DeploymentStatusPrimary, + TaskDefARN: td.ARN, + TaskDefFamily: td.Family, + TaskDefRevision: td.Revision, + DesiredCount: desired, + RolloutState: RolloutStateInProgress, + CreatedAt: now, + UpdatedAt: now, + } +} + +// ensurePrimaryDeployment synthesizes a PRIMARY deployment for a legacy service +// record that predates deployment tracking, keyed by its existing DeploymentID so +// its already-running tasks (StartedBy=ecs-svc/{DeploymentID}) map to it. +func (r *ServiceRecord) ensurePrimaryDeployment() { + if r.primaryDeployment() != nil { + return + } + r.Deployments = append(r.Deployments, Deployment{ + ID: r.DeploymentID, + Status: DeploymentStatusPrimary, + TaskDefARN: r.TaskDefARN, + TaskDefFamily: r.TaskDefFamily, + TaskDefRevision: r.TaskDefRevision, + DesiredCount: r.DesiredCount, + RolloutState: RolloutStateInProgress, + CreatedAt: r.CreatedAt, + UpdatedAt: time.Now().UTC(), + }) + if r.LastGoodTaskDefARN == "" { + r.LastGoodTaskDefARN = r.TaskDefARN + } +} + +// startDeployment demotes the current PRIMARY to ACTIVE (draining) and installs a +// new PRIMARY for td, mirroring the taskdef onto the service record. +func (r *ServiceRecord) startDeployment(td *TaskDefRecord) { + r.demotePrimary() + r.DeploymentID = uuid.NewString() + r.TaskDefFamily = td.Family + r.TaskDefRevision = td.Revision + r.TaskDefARN = td.ARN + r.NetworkMode = resolveNetworkMode(td) + r.Deployments = append(r.Deployments, newPrimaryDeployment(r.DeploymentID, td, r.DesiredCount)) +} + +func (r *ServiceRecord) demotePrimary() { + for i := range r.Deployments { + if r.Deployments[i].Status == DeploymentStatusPrimary { + r.Deployments[i].Status = DeploymentStatusActive + r.Deployments[i].UpdatedAt = time.Now().UTC() + } + } +} + +// pruneCompletedDeployments drops the drained (non-PRIMARY) deployments once the +// rollout has completed, leaving a single PRIMARY at steady state. +func (r *ServiceRecord) pruneCompletedDeployments() { + r.Deployments = slices.DeleteFunc(r.Deployments, func(d Deployment) bool { + return d.Status != DeploymentStatusPrimary + }) +} + +// tallyDeployments zeroes then recomputes each deployment's RUNNING/PENDING counts +// from the service's live tasks and returns the service-wide totals. +func tallyDeployments(svc *ServiceRecord, tasks []TaskRecord) (running, pending int) { + byDep := make(map[string]*Deployment, len(svc.Deployments)) + for i := range svc.Deployments { + svc.Deployments[i].RunningCount = 0 + svc.Deployments[i].PendingCount = 0 + byDep[svc.Deployments[i].ID] = &svc.Deployments[i] + } + for i := range tasks { + d := byDep[deploymentIDFromStartedBy(tasks[i].StartedBy)] + switch tasks[i].LastStatus { + case TaskStatusRunning: + running++ + if d != nil { + d.RunningCount++ + } + case TaskStatusPending: + pending++ + if d != nil { + d.PendingCount++ + } + } + } + return running, pending +} + +// updateRolloutState advances the PRIMARY deployment's rollout state. It completes +// (and prunes drained deployments, recording the taskdef as last-good) once the +// primary is fully up and no old tasks remain; otherwise it stays IN_PROGRESS. A +// FAILED deployment is left as-is (the circuit breaker owns that transition). +func updateRolloutState(svc *ServiceRecord, primary *Deployment, desired int) { + primary.DesiredCount = desired + if primary.RolloutState == RolloutStateFailed { + return + } + remainingOld := 0 + for i := range svc.Deployments { + if svc.Deployments[i].Status != DeploymentStatusPrimary { + remainingOld += svc.Deployments[i].RunningCount + svc.Deployments[i].PendingCount + } + } + if primary.RunningCount >= desired && remainingOld == 0 { + if primary.RolloutState != RolloutStateCompleted { + primary.RolloutState = RolloutStateCompleted + primary.RolloutReason = "ECS deployment completed." + primary.UpdatedAt = time.Now().UTC() + slog.Info("ECS deployment completed", "service", svc.Name, "deployment", primary.ID, "taskDef", primary.TaskDefARN) + } + svc.LastGoodTaskDefARN = primary.TaskDefARN + svc.pruneCompletedDeployments() + return + } + primary.RolloutState = RolloutStateInProgress +} + +// tripCircuitBreaker fails a PRIMARY deployment whose task launches have failed +// past the threshold when the breaker is enabled, optionally rolling back to the +// last-good task definition. Returns true when it acted so the caller can persist +// and defer further launches to the next tick. +func tripCircuitBreaker(svc *ServiceRecord, primary *Deployment) bool { + if !svc.CircuitBreakerEnable || primary.RolloutState == RolloutStateFailed { + return false + } + if primary.FailedTasks < circuitBreakerFailureThreshold { + return false + } + primary.RolloutState = RolloutStateFailed + primary.RolloutReason = "ECS deployment circuit breaker: tasks failed to start." + primary.UpdatedAt = time.Now().UTC() + slog.Warn("ECS deployment circuit breaker tripped", "service", svc.Name, + "deployment", primary.ID, "failedTasks", primary.FailedTasks) + + if !svc.CircuitBreakerRollback || svc.LastGoodTaskDefARN == "" || svc.LastGoodTaskDefARN == primary.TaskDefARN { + return true + } + svc.rollbackToLastGood() + return true +} + +// rollbackToLastGood demotes the failed PRIMARY and starts a fresh PRIMARY from +// the last-good task definition ARN. +func (r *ServiceRecord) rollbackToLastGood() { + family, rev := parseTaskDefRef(r.LastGoodTaskDefARN) + r.demotePrimary() + now := time.Now().UTC() + r.DeploymentID = uuid.NewString() + r.Deployments = append(r.Deployments, Deployment{ + ID: r.DeploymentID, + Status: DeploymentStatusPrimary, + TaskDefARN: r.LastGoodTaskDefARN, + TaskDefFamily: family, + TaskDefRevision: rev, + DesiredCount: r.DesiredCount, + RolloutState: RolloutStateInProgress, + RolloutReason: "ECS deployment rolled back to last good task definition.", + CreatedAt: now, + UpdatedAt: now, + }) + r.TaskDefARN = r.LastGoodTaskDefARN + r.TaskDefFamily = family + r.TaskDefRevision = rev +} diff --git a/spinifex/handlers/ecs/deployments_test.go b/spinifex/handlers/ecs/deployments_test.go new file mode 100644 index 000000000..5085ef004 --- /dev/null +++ b/spinifex/handlers/ecs/deployments_test.go @@ -0,0 +1,225 @@ +package handlers_ecs + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// reloadService re-reads a service record from KV so a test sees writes made by +// the reconciler / failure accounting. +func reloadService(t *testing.T, kv nats.KeyValue, cluster, name string) *ServiceRecord { + t.Helper() + var rec ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, name), &rec) + require.NoError(t, err) + require.True(t, found) + return &rec +} + +// driveRunning reports RUNNING for every PENDING task of a service. +func driveRunning(t *testing.T, svc *Service, kv nats.KeyValue, cluster, name string) { + t.Helper() + tasks, err := svc.listServiceTasks(kv, cluster, name) + require.NoError(t, err) + for i := range tasks { + if tasks[i].LastStatus != TaskStatusPending { + continue + } + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID, + LastStatus: TaskStatusRunning, + })) + } +} + +// failPending reports STOPPED (never-RUNNING) for every PENDING task, simulating +// tasks that fail to start — the deployment circuit-breaker signal. +func failPending(t *testing.T, svc *Service, kv nats.KeyValue, cluster, name string) { + t.Helper() + tasks, err := svc.listServiceTasks(kv, cluster, name) + require.NoError(t, err) + for i := range tasks { + if tasks[i].LastStatus != TaskStatusPending { + continue + } + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: cluster, TaskID: tasks[i].TaskID, + LastStatus: TaskStatusStopped, Reason: "image pull failed", + })) + } +} + +func TestDeployment_CreateService_SeedsPrimary(t *testing.T) { + svc, _, kv := serviceTestRig(t) + out, 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) + + require.Len(t, out.Service.Deployments, 1) + assert.Equal(t, DeploymentStatusPrimary, aws.StringValue(out.Service.Deployments[0].Status)) + assert.Equal(t, RolloutStateInProgress, aws.StringValue(out.Service.Deployments[0].RolloutState)) + + rec := reloadService(t, kv, "web", "web") + require.NotNil(t, rec.primaryDeployment()) + assert.Equal(t, defaultMinimumHealthyPercent, rec.MinimumHealthyPercent) + assert.Equal(t, defaultMaximumPercent, rec.MaximumPercent) + // Every launched task carries the primary deployment id in StartedBy. + tasks, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + require.Len(t, tasks, 2) + for i := range tasks { + assert.Equal(t, rec.DeploymentID, deploymentIDFromStartedBy(tasks[i].StartedBy)) + } +} + +func TestDeployment_RollingUpdate_ReplacesOldWithNew(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) + + // Complete the initial deployment. + driveRunning(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web"))) + rec := reloadService(t, kv, "web", "web") + require.Len(t, rec.Deployments, 1) + assert.Equal(t, RolloutStateCompleted, rec.primaryDeployment().RolloutState) + firstDeployID := rec.DeploymentID + + // New taskdef revision starts a rolling deployment. + registerTaskDef(t, svc, "app", 128, 256) // app:2 + upd, err := svc.UpdateService(&ecs.UpdateServiceInput{ + Cluster: aws.String("web"), Service: aws.String("web"), + TaskDefinition: aws.String("app:2"), + }, testAccountID) + require.NoError(t, err) + require.Len(t, upd.Service.Deployments, 2) // PRIMARY app:2 + ACTIVE app:1 + + rec = reloadService(t, kv, "web", "web") + assert.NotEqual(t, firstDeployID, rec.DeploymentID) + // maximumPercent=200 lets 2 new tasks launch alongside the 2 old running ones. + newTasks := 0 + tasks, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + for i := range tasks { + if deploymentIDFromStartedBy(tasks[i].StartedBy) == rec.DeploymentID { + newTasks++ + } + } + assert.Equal(t, 2, newTasks) + + // New tasks come up; a reconcile drains the old ones and completes the rollout. + driveRunning(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web"))) + rec = reloadService(t, kv, "web", "web") + require.Len(t, rec.Deployments, 1) + assert.Equal(t, RolloutStateCompleted, rec.primaryDeployment().RolloutState) + assert.Equal(t, rec.DeploymentID, rec.primaryDeployment().ID) + + // Only new-deployment tasks remain live. + tasks, err = svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + require.Len(t, tasks, 2) + for i := range tasks { + assert.Equal(t, rec.DeploymentID, deploymentIDFromStartedBy(tasks[i].StartedBy)) + } +} + +func TestDeployment_MinimumHealthyPercent_GatesDrain(t *testing.T) { + svc, _, kv := serviceTestRig(t) + // min=100, max=150: with desired=2 the rollout may run up to 3 tasks and must + // keep 2 healthy, so old tasks only drain as new ones become healthy. + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), + TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(2), + DeploymentConfiguration: &ecs.DeploymentConfiguration{ + MinimumHealthyPercent: aws.Int64(100), MaximumPercent: aws.Int64(150), + }, + }, testAccountID) + require.NoError(t, err) + driveRunning(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web"))) + + registerTaskDef(t, svc, "app", 128, 256) // app:2 + _, err = svc.UpdateService(&ecs.UpdateServiceInput{ + Cluster: aws.String("web"), Service: aws.String("web"), TaskDefinition: aws.String("app:2"), + }, testAccountID) + require.NoError(t, err) + + // Ceiling of 3 tasks total: 2 old running + 1 new pending. No old task drained + // yet because healthy running (2) must not drop below minCount (2). + tasks, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + assert.Len(t, tasks, 3) + rec := reloadService(t, kv, "web", "web") + assert.Equal(t, 2, rec.RunningCount) + assert.Equal(t, 1, rec.PendingCount) +} + +func TestDeployment_CircuitBreaker_RollsBackToLastGood(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(1), + DeploymentConfiguration: &ecs.DeploymentConfiguration{ + DeploymentCircuitBreaker: &ecs.DeploymentCircuitBreaker{ + Enable: aws.Bool(true), Rollback: aws.Bool(true), + }, + }, + }, testAccountID) + require.NoError(t, err) + driveRunning(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web"))) + goodARN := reloadService(t, kv, "web", "web").LastGoodTaskDefARN + require.NotEmpty(t, goodARN) + + // Roll out a revision whose tasks always fail to start. + registerTaskDef(t, svc, "app", 128, 256) // app:2 + _, err = svc.UpdateService(&ecs.UpdateServiceInput{ + Cluster: aws.String("web"), Service: aws.String("web"), TaskDefinition: aws.String("app:2"), + }, testAccountID) + require.NoError(t, err) + + // Each cycle: fail the pending new task, then reconcile relaunches. After the + // failure threshold the breaker trips on the next reconcile and rolls back. + for range circuitBreakerFailureThreshold { + failPending(t, svc, kv, "web", "web") + require.NoError(t, svc.reconcileService(kv, testAccountID, reloadService(t, kv, "web", "web"))) + } + + rec := reloadService(t, kv, "web", "web") + primary := rec.primaryDeployment() + require.NotNil(t, primary) + assert.Equal(t, goodARN, primary.TaskDefARN) + assert.Equal(t, goodARN, rec.TaskDefARN) +} + +func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) { + svc, _, kv := serviceTestRig(t) + // A record written before deployment tracking existed: no Deployments slice. + rec := &ServiceRecord{ + Name: "legacy", ARN: ServiceARN(testRegion, testAccountID, "web", "legacy"), + Cluster: "web", TaskDefFamily: "app", TaskDefRevision: 1, + TaskDefARN: TaskDefARN(testRegion, testAccountID, "app", 1), + DesiredCount: 1, Status: ServiceStatusActive, + SchedulingStrategy: SchedulingStrategyReplica, DeploymentID: "legacy-dep", + } + require.NoError(t, putJSON(kv, ServiceKey("web", "legacy"), rec)) + + require.NoError(t, svc.reconcileService(kv, testAccountID, rec)) + reloaded := reloadService(t, kv, "web", "legacy") + primary := reloaded.primaryDeployment() + require.NotNil(t, primary) + assert.Equal(t, "legacy-dep", primary.ID) + assert.Equal(t, defaultMinimumHealthyPercent, reloaded.MinimumHealthyPercent) +} diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index bf63b562e..799a64218 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -243,6 +243,11 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error { // Deregister targets, release the public IP, release capacity + reclaim the // task ENI once, on the transition into STOPPED. if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped { + // A task that stops without ever reaching RUNNING is a deployment failure; + // feed the owning deployment's circuit breaker. + if task.StartedAt.IsZero() { + s.recordDeploymentFailure(kv, msg.ClusterName, &task) + } s.deregisterServiceTargets(kv, msg.AccountID, &task) s.releaseTaskPublicIP(msg.AccountID, &task) s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID) @@ -286,6 +291,32 @@ func (s *Service) SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeInput, a return &ecs.SubmitTaskStateChangeOutput{Acknowledgment: aws.String("OK")}, nil } +// recordDeploymentFailure increments the failed-task counter on the deployment +// that launched a task which stopped before ever running, driving the service's +// deployment circuit breaker. No-op for a non-service task or an unknown deployment. +func (s *Service) recordDeploymentFailure(kv nats.KeyValue, cluster string, task *TaskRecord) { + name := serviceNameFromGroup(task.Group) + depID := deploymentIDFromStartedBy(task.StartedBy) + if name == "" || depID == "" { + return + } + var svc ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, name), &svc) + if err != nil || !found { + return + } + for i := range svc.Deployments { + if svc.Deployments[i].ID == depID { + svc.Deployments[i].FailedTasks++ + svc.Deployments[i].UpdatedAt = time.Now().UTC() + if perr := putJSON(kv, ServiceKey(cluster, name), &svc); perr != nil { + slog.Error("ECS deployment failure accounting: persist failed", "service", name, "err", perr) + } + return + } + } +} + // releaseReservation returns a stopped task's capacity to its instance under CAS. func (s *Service) releaseReservation(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) error { for range reservePlacementRetries { diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 67909d67a..69753e577 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -32,8 +32,45 @@ const ( // DAEMON is rejected at CreateService. SchedulingStrategyReplica = "REPLICA" SchedulingStrategyDaemon = "DAEMON" + + // Deployment status: PRIMARY is the deployment being rolled out (or steady); + // ACTIVE is a superseded deployment still draining its tasks. + DeploymentStatusPrimary = "PRIMARY" + DeploymentStatusActive = "ACTIVE" + + // Deployment rollout state (AWS deploymentController rollout enum subset). + RolloutStateInProgress = "IN_PROGRESS" + RolloutStateCompleted = "COMPLETED" + RolloutStateFailed = "FAILED" + + // deploymentConfiguration defaults for a REPLICA service when unset. + defaultMinimumHealthyPercent = 100 + defaultMaximumPercent = 200 + + // circuitBreakerFailureThreshold trips the breaker once this many of the + // primary deployment's task launches stop before ever reaching RUNNING. + circuitBreakerFailureThreshold = 3 ) +// Deployment is one rollout of a service's task definition. A service has exactly +// one PRIMARY deployment plus zero or more ACTIVE (superseded, draining) ones +// while a rolling update is in flight; steady state is a single PRIMARY. +type Deployment struct { + ID string `json:"id"` + Status string `json:"status"` + TaskDefARN string `json:"taskDefArn"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + DesiredCount int `json:"desiredCount"` + RunningCount int `json:"runningCount"` + PendingCount int `json:"pendingCount"` + FailedTasks int `json:"failedTasks"` + RolloutState string `json:"rolloutState"` + RolloutReason string `json:"rolloutStateReason,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + // ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID // scope every ARN; the partition is fixed to "aws" to match the rest of the // stack. @@ -94,20 +131,30 @@ type ContainerDef struct { Command []string `json:"command,omitempty"` Environment map[string]string `json:"environment,omitempty"` PortMappings []bus.PortMapping `json:"portMappings,omitempty"` + // LogDriver / LogOptions capture the container's logConfiguration. Only the + // host-side json-file default is honored; any other driver is accepted for + // parity but warned at register time (logs are discarded). + LogDriver string `json:"logDriver,omitempty"` + LogOptions map[string]string `json:"logOptions,omitempty"` } +// LogDriverJSONFile is the only log driver the agent honors: containerd's task IO +// lands in the host journal/file, retrievable per ecs-logging.md. +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"` - 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"` + RegisteredAt time.Time `json:"registeredAt"` } // reservedCPU/reservedMemory sum the task definition's per-container reservations @@ -239,6 +286,26 @@ type ServiceRecord struct { DeploymentID string `json:"deploymentId"` RunningCount int `json:"runningCount"` PendingCount int `json:"pendingCount"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + // 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. + MinimumHealthyPercent int `json:"minimumHealthyPercent,omitempty"` + MaximumPercent int `json:"maximumPercent,omitempty"` + CircuitBreakerEnable bool `json:"deploymentCircuitBreakerEnable,omitempty"` + CircuitBreakerRollback bool `json:"deploymentCircuitBreakerRollback,omitempty"` + LastGoodTaskDefARN string `json:"lastGoodTaskDefArn,omitempty"` + Deployments []Deployment `json:"deployments,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// primaryDeployment returns a pointer to the service's PRIMARY deployment, or nil +// when none exists (a legacy record before ensurePrimaryDeployment runs). +func (r *ServiceRecord) primaryDeployment() *Deployment { + for i := range r.Deployments { + if r.Deployments[i].Status == DeploymentStatusPrimary { + return &r.Deployments[i] + } + } + return nil } diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index ad8a534e1..b4b2878a7 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "maps" "slices" "strconv" @@ -334,6 +335,10 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, if family == "" { return nil, errors.New(awserrors.ErrorECSInvalidParameter) } + if err := validateContainerDefs(input.ContainerDefinitions); err != nil { + return nil, err + } + warnUnsupportedLogDrivers(family, input.ContainerDefinitions) kv, err := s.bucket(accountID) if err != nil { return nil, err @@ -345,16 +350,17 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, } rec := TaskDefRecord{ - Family: family, - Revision: rev, - ARN: TaskDefARN(s.region, accountID, family, rev), - NetworkMode: aws.StringValue(input.NetworkMode), - CPU: aws.StringValue(input.Cpu), - Memory: aws.StringValue(input.Memory), - TaskRoleArn: aws.StringValue(input.TaskRoleArn), - Status: TaskDefStatusActive, - RegisteredAt: time.Now().UTC(), - Containers: containerDefsFromAWS(input.ContainerDefinitions), + Family: family, + Revision: rev, + ARN: TaskDefARN(s.region, accountID, family, rev), + NetworkMode: aws.StringValue(input.NetworkMode), + CPU: aws.StringValue(input.Cpu), + Memory: aws.StringValue(input.Memory), + TaskRoleArn: aws.StringValue(input.TaskRoleArn), + ExecutionRoleArn: aws.StringValue(input.ExecutionRoleArn), + Status: TaskDefStatusActive, + RegisteredAt: time.Now().UTC(), + Containers: containerDefsFromAWS(input.ContainerDefinitions), } if err := putJSON(kv, TaskDefRevKey(family, rev), &rec); err != nil { return nil, err @@ -365,6 +371,35 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, return &ecs.RegisterTaskDefinitionOutput{TaskDefinition: rec.toAWS()}, nil } +// validateContainerDefs hard-rejects taskdef features the data plane cannot honor. +// secrets[] is dropped silently by the assign path, so a task would run believing +// it has secrets it never receives (ecs-v1 Q18) — reject at register instead. +func validateContainerDefs(defs []*ecs.ContainerDefinition) error { + for _, c := range defs { + if c != nil && len(c.Secrets) > 0 { + return errors.New(awserrors.ErrorECSInvalidParameter) + } + } + return nil +} + +// warnUnsupportedLogDrivers emits the honest "logs discarded" operator signal for +// any container whose log driver is not the host-side json-file default. The +// taskdef is still accepted for parity; only json-file is actually collected. +func warnUnsupportedLogDrivers(family string, defs []*ecs.ContainerDefinition) { + for _, c := range defs { + if c == nil || c.LogConfiguration == nil { + continue + } + driver := aws.StringValue(c.LogConfiguration.LogDriver) + if driver == "" || driver == LogDriverJSONFile { + continue + } + slog.Warn("ECS RegisterTaskDefinition: logDriver not implemented; container logs will be discarded", + "family", family, "container", aws.StringValue(c.Name), "logDriver", driver) + } +} + // nextRevision reads the family's latest-rev and returns latest+1 (1 if absent). func (s *Service) nextRevision(kv nats.KeyValue, family string) (int, error) { var latest int @@ -521,6 +556,9 @@ func (r *TaskDefRecord) toAWS() *ecs.TaskDefinition { if r.TaskRoleArn != "" { td.TaskRoleArn = aws.String(r.TaskRoleArn) } + if r.ExecutionRoleArn != "" { + td.ExecutionRoleArn = aws.String(r.ExecutionRoleArn) + } for _, c := range r.Containers { td.ContainerDefinitions = append(td.ContainerDefinitions, c.toAWS()) } diff --git a/spinifex/handlers/ecs/services.go b/spinifex/handlers/ecs/services.go index 1637bd15d..f8302a6d7 100644 --- a/spinifex/handlers/ecs/services.go +++ b/spinifex/handlers/ecs/services.go @@ -183,6 +183,9 @@ func (s *Service) CreateService(input *ecs.CreateServiceInput, accountID string) rec.SecurityGroups = awsStringSlice(v.SecurityGroups) rec.AssignPublicIP = aws.StringValue(v.AssignPublicIp) } + applyDeploymentConfig(&rec, input.DeploymentConfiguration) + rec.LastGoodTaskDefARN = taskDef.ARN + rec.Deployments = []Deployment{newPrimaryDeployment(rec.DeploymentID, taskDef, rec.DesiredCount)} if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil { return nil, err } @@ -210,6 +213,10 @@ func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string) return nil, errors.New(awserrors.ErrorECSServiceNotFound) } + rec.ensurePrimaryDeployment() + if input.DeploymentConfiguration != nil { + applyDeploymentConfig(&rec, input.DeploymentConfiguration) + } if input.DesiredCount != nil { rec.DesiredCount = int(aws.Int64Value(input.DesiredCount)) } @@ -218,11 +225,11 @@ func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string) if terr != nil { return nil, terr } - rec.TaskDefFamily = taskDef.Family - rec.TaskDefRevision = taskDef.Revision - rec.TaskDefARN = taskDef.ARN - rec.NetworkMode = resolveNetworkMode(taskDef) - rec.DeploymentID = uuid.NewString() + // A new task definition starts a rolling deployment only when it differs + // from the current PRIMARY; a no-op re-apply keeps the deployment steady. + if primary := rec.primaryDeployment(); primary == nil || primary.TaskDefARN != taskDef.ARN { + rec.startDeployment(taskDef) + } } rec.UpdatedAt = time.Now().UTC() if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil { @@ -314,59 +321,56 @@ func (s *Service) ListServices(input *ecs.ListServicesInput, accountID string) ( // --- Reconciliation --- -// reconcileService converges a service's running task count on its desired count: -// it launches the shortfall via RunTask (tagged with the service group so the -// task counts toward the service) and force-stops any surplus. Counts on the -// record are refreshed as a side effect. +// reconcileService drives a service's rolling deployment toward its desired +// count: it launches PRIMARY-deployment tasks within the maximumPercent ceiling +// and drains superseded (old) tasks while healthy running stays at/above +// minimumHealthyPercent, then advances the rollout / circuit-breaker state. Per- +// deployment and overall counts are refreshed as a side effect. func (s *Service) reconcileService(kv nats.KeyValue, accountID string, svc *ServiceRecord) error { if svc.Status != ServiceStatusActive { return nil } + svc.normalizeDeploymentConfig() + svc.ensurePrimaryDeployment() + tasks, err := s.listServiceTasks(kv, svc.Cluster, svc.Name) if err != nil { return err } + running, pending := tallyDeployments(svc, tasks) - running, pending := 0, 0 - for i := range tasks { - switch tasks[i].LastStatus { - case TaskStatusRunning: - running++ - case TaskStatusPending: - pending++ + primary := svc.primaryDeployment() + if primary != nil { + // A failing deployment trips the breaker (and may roll back); persist and + // let the next tick roll the replacement in. + if tripCircuitBreaker(svc, primary) { + svc.RunningCount, svc.PendingCount = running, pending + return putJSON(kv, ServiceKey(svc.Cluster, svc.Name), svc) } + desired := svc.DesiredCount + maxCount := desired * svc.MaximumPercent / 100 + minCount := ceilPercent(desired, svc.MinimumHealthyPercent) + + primaryActive := primary.RunningCount + primary.PendingCount + if primaryActive < desired && primary.RolloutState != RolloutStateFailed { + n := min(desired-primaryActive, max(maxCount-(running+pending), 0)) + if n > 0 { + s.launchDeploymentTasks(accountID, svc, primary, n) + } + } + s.stopSurplusTasks(kv, accountID, tasks, primary.ID, desired, running, minCount) + + // Re-tally after launch/stop so rollout state + persisted counts are current. + if fresh, ferr := s.listServiceTasks(kv, svc.Cluster, svc.Name); ferr == nil { + running, pending = tallyDeployments(svc, fresh) + } + updateRolloutState(svc, primary, desired) } - active := running + pending - switch { - case active < svc.DesiredCount: - s.launchServiceTasks(accountID, svc, svc.DesiredCount-active) - case active > svc.DesiredCount: - s.stopServiceSurplus(kv, accountID, tasks, active-svc.DesiredCount) - } - - // Recount after launch/stop so the persisted counts reflect current state. - svc.RunningCount, svc.PendingCount = s.countServiceTasks(kv, svc.Cluster, svc.Name) + svc.RunningCount, svc.PendingCount = running, pending return putJSON(kv, ServiceKey(svc.Cluster, svc.Name), svc) } -// countServiceTasks returns a service's current RUNNING and PENDING task counts. -func (s *Service) countServiceTasks(kv nats.KeyValue, cluster, name string) (running, pending int) { - tasks, err := s.listServiceTasks(kv, cluster, name) - if err != nil { - return 0, 0 - } - for i := range tasks { - switch tasks[i].LastStatus { - case TaskStatusRunning: - running++ - case TaskStatusPending: - pending++ - } - } - return running, pending -} - // reconcileAllServices is the scheduler-tick fan-out: every ACTIVE service in // every ECS account bucket is reconciled. Runs only on the scheduler leader. func (s *Service) reconcileAllServices() { @@ -395,15 +399,16 @@ func (s *Service) reconcileAllServices() { } } -// launchServiceTasks places n replacement tasks via the standard RunTask flow, -// tagged with the service group + deployment so they count toward the service. -func (s *Service) launchServiceTasks(accountID string, svc *ServiceRecord, n int) { +// launchDeploymentTasks places n tasks for a specific deployment via the standard +// RunTask flow, tagged with the service group + the deployment ID (StartedBy) so +// the reconciler maps them back to the deployment they belong to. +func (s *Service) launchDeploymentTasks(accountID string, svc *ServiceRecord, dep *Deployment, n int) { in := &ecs.RunTaskInput{ Cluster: aws.String(svc.Cluster), - TaskDefinition: aws.String(svc.TaskDefARN), + TaskDefinition: aws.String(dep.TaskDefARN), Count: aws.Int64(int64(n)), Group: aws.String(serviceTaskGroup(svc.Name)), - StartedBy: aws.String("ecs-svc/" + svc.DeploymentID), + StartedBy: aws.String(deploymentStartedBy(dep.ID)), } if svc.PlacementStrategy != "" { in.PlacementStrategy = []*ecs.PlacementStrategy{{Type: aws.String(svc.PlacementStrategy)}} @@ -429,26 +434,57 @@ func (s *Service) launchServiceTasks(accountID string, svc *ServiceRecord, n int } } -// stopServiceSurplus force-stops n of a service's tasks, preferring PENDING tasks -// (least disruptive) before RUNNING ones. -func (s *Service) stopServiceSurplus(kv nats.KeyValue, accountID string, tasks []TaskRecord, n int) { - order := make([]int, 0, len(tasks)) +// stopSurplusTasks drains a service's excess tasks: superseded (non-primary) +// deployment tasks toward zero, and the primary's own surplus on a scale-in. +// PENDING surplus is stopped freely; RUNNING surplus is gated so healthy running +// never drops below minimumHealthyPercent (minCount). +func (s *Service) stopSurplusTasks(kv nats.KeyValue, accountID string, tasks []TaskRecord, primaryID string, desired, runningTotal, minCount int) { + var oldPending, oldRunning, primaryPending, primaryRunning []int for i := range tasks { - if tasks[i].LastStatus == TaskStatusPending { - order = append(order, i) + isPrimary := deploymentIDFromStartedBy(tasks[i].StartedBy) == primaryID + switch tasks[i].LastStatus { + case TaskStatusPending: + if isPrimary { + primaryPending = append(primaryPending, i) + } else { + oldPending = append(oldPending, i) + } + case TaskStatusRunning: + if isPrimary { + primaryRunning = append(primaryRunning, i) + } else { + oldRunning = append(oldRunning, i) + } } } - for i := range tasks { - if tasks[i].LastStatus == TaskStatusRunning { - order = append(order, i) - } + primarySurplus := max(len(primaryPending)+len(primaryRunning)-desired, 0) + runBudget := max(runningTotal-minCount, 0) + stop := func(i int, reason string) { s.forceStopTask(kv, accountID, &tasks[i], reason) } + + for _, i := range oldPending { + stop(i, deploymentSupersededReason) } - for _, idx := range order { - if n <= 0 { + for _, i := range primaryPending { + if primarySurplus <= 0 { break } - s.forceStopTask(kv, accountID, &tasks[idx], "Service scaled in") - n-- + stop(i, "Service scaled in") + primarySurplus-- + } + for _, i := range oldRunning { + if runBudget <= 0 { + break + } + stop(i, deploymentSupersededReason) + runBudget-- + } + for _, i := range primaryRunning { + if primarySurplus <= 0 || runBudget <= 0 { + break + } + stop(i, "Service scaled in") + primarySurplus-- + runBudget-- } } @@ -580,5 +616,34 @@ func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service ContainerPort: aws.Int64(int64(lb.ContainerPort)), }) } + if r.MinimumHealthyPercent > 0 || r.MaximumPercent > 0 { + svc.DeploymentConfiguration = &ecs.DeploymentConfiguration{ + MinimumHealthyPercent: aws.Int64(int64(r.MinimumHealthyPercent)), + MaximumPercent: aws.Int64(int64(r.MaximumPercent)), + DeploymentCircuitBreaker: &ecs.DeploymentCircuitBreaker{ + Enable: aws.Bool(r.CircuitBreakerEnable), + Rollback: aws.Bool(r.CircuitBreakerRollback), + }, + } + } + for i := range r.Deployments { + d := &r.Deployments[i] + svc.Deployments = append(svc.Deployments, &ecs.Deployment{ + Id: aws.String(d.ID), + Status: aws.String(d.Status), + TaskDefinition: aws.String(d.TaskDefARN), + DesiredCount: aws.Int64(int64(d.DesiredCount)), + RunningCount: aws.Int64(int64(d.RunningCount)), + PendingCount: aws.Int64(int64(d.PendingCount)), + FailedTasks: aws.Int64(int64(d.FailedTasks)), + RolloutState: aws.String(d.RolloutState), + RolloutStateReason: func() *string { + if d.RolloutReason == "" { + return nil + } + return aws.String(d.RolloutReason) + }(), + }) + } return svc } diff --git a/spinifex/handlers/ecs/taskdef_validation_test.go b/spinifex/handlers/ecs/taskdef_validation_test.go new file mode 100644 index 000000000..8579dad19 --- /dev/null +++ b/spinifex/handlers/ecs/taskdef_validation_test.go @@ -0,0 +1,91 @@ +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" +) + +// TestRegisterTaskDefinition_RejectsSecrets covers siv-458: a container secrets[] +// is hard-rejected rather than silently dropped. +func TestRegisterTaskDefinition_RejectsSecrets(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String("app"), + ContainerDefinitions: []*ecs.ContainerDefinition{{ + Name: aws.String("app"), Image: aws.String("registry/app:1"), Essential: aws.Bool(true), + Secrets: []*ecs.Secret{{ + Name: aws.String("DB_PASSWORD"), ValueFrom: aws.String("arn:aws:ssm:::parameter/db"), + }}, + }}, + }, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "InvalidParameterException") +} + +// TestRegisterTaskDefinition_AcceptsUnsupportedLogDriver covers siv-455/458: a +// non-json-file driver is accepted for parity (warned, not rejected) and the +// driver round-trips through DescribeTaskDefinition. +func TestRegisterTaskDefinition_AcceptsUnsupportedLogDriver(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String("app"), + ContainerDefinitions: []*ecs.ContainerDefinition{{ + Name: aws.String("app"), Image: aws.String("registry/app:1"), Essential: aws.Bool(true), + LogConfiguration: &ecs.LogConfiguration{ + LogDriver: aws.String("awslogs"), + Options: map[string]*string{"awslogs-group": aws.String("/ecs/app")}, + }, + }}, + }, testAccountID) + require.NoError(t, err) + + d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{ + TaskDefinition: aws.String("app"), + }, testAccountID) + require.NoError(t, err) + require.Len(t, d.TaskDefinition.ContainerDefinitions, 1) + lc := d.TaskDefinition.ContainerDefinitions[0].LogConfiguration + require.NotNil(t, lc) + assert.Equal(t, "awslogs", aws.StringValue(lc.LogDriver)) + assert.Equal(t, "/ecs/app", aws.StringValue(lc.Options["awslogs-group"])) +} + +// TestRunTask_AssignCarriesExecutionRoleAndLogDriver covers siv-459 (execution +// role plumbed to the agent) and siv-455 (log driver reaches the assign). +func TestRunTask_AssignCarriesExecutionRoleAndLogDriver(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + execARN := "arn:aws:iam::123456789012:role/exec-app" + _, err = svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String("app"), + ExecutionRoleArn: aws.String(execARN), + ContainerDefinitions: []*ecs.ContainerDefinition{{ + Name: aws.String("app"), Image: aws.String("registry/app:1"), + Cpu: aws.Int64(128), Memory: aws.Int64(256), Essential: aws.Bool(true), + LogConfiguration: &ecs.LogConfiguration{LogDriver: aws.String(LogDriverJSONFile)}, + }}, + }, testAccountID) + require.NoError(t, err) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("app")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, execARN, aws.StringValue(d.TaskDefinition.ExecutionRoleArn)) + + _, err = svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + + poll, err := svc.PollAssignments(&PollAssignmentsInput{Cluster: "web", ContainerInstance: "i-1"}, testAccountID) + require.NoError(t, err) + require.Len(t, poll.Assignments, 1) + assert.Equal(t, execARN, poll.Assignments[0].ExecutionRoleARN) + require.Len(t, poll.Assignments[0].Containers, 1) + assert.Equal(t, LogDriverJSONFile, poll.Assignments[0].Containers[0].LogDriver) +} diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go index 5cf56877c..da4831b0b 100644 --- a/spinifex/handlers/ecs/tasks.go +++ b/spinifex/handlers/ecs/tasks.go @@ -192,19 +192,20 @@ func (s *Service) newTaskRecord(accountID, cluster, taskID string, td *TaskDefRe // an unacked assign survives an agent crash and is re-delivered on the next poll. func (s *Service) publishAssign(kv nats.KeyValue, accountID, cluster, instanceID string, rec *TaskRecord, td *TaskDefRecord) error { msg := bus.Assign{ - AccountID: accountID, - ClusterName: cluster, - InstanceID: instanceID, - TaskID: rec.TaskID, - TaskARN: rec.ARN, - TaskDefFamily: td.Family, - TaskDefRevision: td.Revision, - TaskRoleARN: td.TaskRoleArn, - ENIID: rec.ENIID, - ENIMacAddress: rec.ENIMacAddress, - ENIPrivateIP: rec.ENIPrivateIP, - ENISubnetID: rec.ENISubnetID, - AssignedAt: time.Now().UTC(), + AccountID: accountID, + ClusterName: cluster, + InstanceID: instanceID, + TaskID: rec.TaskID, + TaskARN: rec.ARN, + TaskDefFamily: td.Family, + TaskDefRevision: td.Revision, + TaskRoleARN: td.TaskRoleArn, + ExecutionRoleARN: td.ExecutionRoleArn, + ENIID: rec.ENIID, + ENIMacAddress: rec.ENIMacAddress, + ENIPrivateIP: rec.ENIPrivateIP, + ENISubnetID: rec.ENISubnetID, + AssignedAt: time.Now().UTC(), } for _, c := range td.Containers { msg.Containers = append(msg.Containers, c.toAssignContainer())