diff --git a/cmd/ecs-agent/ecrresolver.go b/cmd/ecs-agent/ecrresolver.go index 3ff036661..3a79d2d32 100644 --- a/cmd/ecs-agent/ecrresolver.go +++ b/cmd/ecs-agent/ecrresolver.go @@ -3,6 +3,7 @@ package main import ( "context" "net/http" + "strings" "sync" "github.com/mulgadc/spinifex/cmd/ecs-agent/credentials" @@ -59,8 +60,24 @@ func (e *ecrResolver) client() (*http.Client, error) { return c, nil } +// isECRRef reports whether ref targets an ECR registry, i.e. its registry host +// is ".dkr.ecr..". Only ECR pulls need a minted token; +// public registries (docker.io and friends) pull anonymously, so refs without +// an ECR host must not drag in IMDS host credentials. +func isECRRef(ref string) bool { + host, _, _ := strings.Cut(ref, "/") + if !strings.Contains(host, ".") && !strings.Contains(host, ":") { + return false // no registry host -> docker.io default + } + return strings.Contains(host, ".dkr.ecr.") +} + // Authorize returns the ECR token's user/password and proxy endpoint for ref. +// Non-ECR refs resolve to anonymous pull (empty credentials). func (e *ecrResolver) Authorize(ctx context.Context, ref string) (user, pass, endpoint string, err error) { + if !isECRRef(ref) { + return "", "", "", nil + } c, err := e.creds.Retrieve(ctx) if err != nil { return "", "", "", err diff --git a/cmd/ecs-agent/ecrresolver_test.go b/cmd/ecs-agent/ecrresolver_test.go index bc5edb0e8..a2cc418e2 100644 --- a/cmd/ecs-agent/ecrresolver_test.go +++ b/cmd/ecs-agent/ecrresolver_test.go @@ -78,7 +78,7 @@ func TestECRResolver_Authorize(t *testing.T) { func TestECRResolver_CredsError(t *testing.T) { r := newECRResolver(stubCreds{err: errors.New("no imds")}, "us-east-1", "https://x", http.DefaultClient) - if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil { + if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil { t.Fatal("expected creds error to propagate") } } @@ -87,7 +87,7 @@ func TestECRResolver_GatewayError(t *testing.T) { srv := gatewayStub(t, "", "", http.StatusInternalServerError) creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}} r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv)) - if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil { + if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil { t.Fatal("expected gateway error") } } @@ -96,7 +96,36 @@ func TestECRResolver_LazyClient_BadCADeferredToAuthorize(t *testing.T) { creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}} // Construction with a bogus CA path must not fail — only Authorize does. r := newLazyECRResolver(creds, "us-east-1", "https://gw", filepath.Join(t.TempDir(), "absent.pem")) - if _, _, _, err := r.Authorize(context.Background(), "ref"); err == nil { + if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil { t.Fatal("expected lazy gateway-client build to fail on missing CA") } } + +// TestECRResolver_NonECRAnonymous asserts that a public-registry ref pulls +// anonymously: Authorize returns empty creds without ever touching IMDS, even +// when the credentials provider would error. +func TestECRResolver_NonECRAnonymous(t *testing.T) { + r := newECRResolver(stubCreds{err: errors.New("imds must not be called")}, "us-east-1", "https://x", http.DefaultClient) + for _, ref := range []string{"nginx:alpine", "docker.io/library/nginx:alpine", "ghcr.io/org/app:1"} { + user, pass, endpoint, err := r.Authorize(context.Background(), ref) + if err != nil || user != "" || pass != "" || endpoint != "" { + t.Errorf("ref %q: got (%q,%q,%q,%v), want anonymous", ref, user, pass, endpoint, err) + } + } +} + +func TestIsECRRef(t *testing.T) { + cases := map[string]bool{ + "nginx:alpine": false, + "docker.io/library/nginx": false, + "ghcr.io/org/app:1": false, + "registry:5000/app:1": false, + "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1": true, + "123456789012.dkr.ecr.ap-southeast-2.amazonaws.com/x": true, + } + for ref, want := range cases { + if got := isECRRef(ref); got != want { + t.Errorf("isECRRef(%q) = %v, want %v", ref, got, want) + } + } +} diff --git a/cmd/ecs-agent/runtime/containerd.go b/cmd/ecs-agent/runtime/containerd.go index 5c8abeb87..1e8760e68 100644 --- a/cmd/ecs-agent/runtime/containerd.go +++ b/cmd/ecs-agent/runtime/containerd.go @@ -9,6 +9,7 @@ import ( "github.com/containerd/containerd/v2/core/remotes" "github.com/containerd/containerd/v2/core/remotes/docker" "github.com/containerd/containerd/v2/pkg/namespaces" + "github.com/distribution/reference" ) // ecsNamespace isolates ecs-agent images/containers in containerd, mirroring the @@ -51,27 +52,44 @@ func New(socket string) (Runtime, error) { func (p *containerdPuller) Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) { ctx = namespaces.WithNamespace(ctx, ecsNamespace) - resolver, err := newDockerResolver(ctx, spec.Ref, r) + ref, err := normalizeRef(spec.Ref) if err != nil { return Image{}, err } - img, err := p.client.Pull(ctx, spec.Ref, + resolver, err := newDockerResolver(ctx, ref, r) + if err != nil { + return Image{}, err + } + + img, err := p.client.Pull(ctx, ref, containerd.WithPullUnpack, containerd.WithResolver(resolver), ) if err != nil { - return Image{}, fmt.Errorf("pull %s: %w", spec.Ref, err) + return Image{}, fmt.Errorf("pull %s: %w", ref, err) } target := img.Target() return Image{ - Ref: spec.Ref, + Ref: ref, Digest: target.Digest.String(), Size: target.Size, }, nil } +// normalizeRef expands a task-definition image string into the fully-qualified +// reference containerd's resolver requires: a bare "nginx:alpine" becomes +// "docker.io/library/nginx:alpine" and an untagged ref gets ":latest". Without +// this, containerd parses the bare ref as "dummy://nginx:alpine" and rejects it. +func normalizeRef(ref string) (string, error) { + named, err := reference.ParseNormalizedNamed(ref) + if err != nil { + return "", fmt.Errorf("parse image ref %q: %w", ref, err) + } + return reference.TagNameOnly(named).String(), nil +} + // Close releases the containerd client connection. func (p *containerdPuller) Close() error { if p.client == nil { diff --git a/cmd/ecs-agent/runtime/containerd_run.go b/cmd/ecs-agent/runtime/containerd_run.go index d99f9d6ca..fde122577 100644 --- a/cmd/ecs-agent/runtime/containerd_run.go +++ b/cmd/ecs-agent/runtime/containerd_run.go @@ -21,9 +21,13 @@ var _ Runner = (*containerdPuller)(nil) func (p *containerdPuller) Run(ctx context.Context, id string, spec RunSpec) (string, error) { ctx = namespaces.WithNamespace(ctx, ecsNamespace) - image, err := p.client.GetImage(ctx, spec.Image) + ref, err := normalizeRef(spec.Image) if err != nil { - return "", fmt.Errorf("get image %s: %w", spec.Image, err) + return "", err + } + image, err := p.client.GetImage(ctx, ref) + if err != nil { + return "", fmt.Errorf("get image %s: %w", ref, err) } specOpts := []oci.SpecOpts{ diff --git a/cmd/ecs-agent/runtime/runtime_test.go b/cmd/ecs-agent/runtime/runtime_test.go index a149e5f93..b4e34c52d 100644 --- a/cmd/ecs-agent/runtime/runtime_test.go +++ b/cmd/ecs-agent/runtime/runtime_test.go @@ -65,3 +65,25 @@ func TestFakePuller_Close(t *testing.T) { t.Fatalf("Close: err=%v closed=%v", err, f.Closed) } } + +func TestNormalizeRef(t *testing.T) { + cases := map[string]string{ + "nginx:alpine": "docker.io/library/nginx:alpine", + "nginx": "docker.io/library/nginx:latest", + "ghcr.io/org/app:1": "ghcr.io/org/app:1", + "123456789012.dkr.ecr.ap-southeast-2.spinifex.internal/app:1": "123456789012.dkr.ecr.ap-southeast-2.spinifex.internal/app:1", + } + for in, want := range cases { + got, err := normalizeRef(in) + if err != nil { + t.Errorf("normalizeRef(%q): %v", in, err) + continue + } + if got != want { + t.Errorf("normalizeRef(%q) = %q, want %q", in, got, want) + } + } + if _, err := normalizeRef("Bad::Ref"); err == nil { + t.Error("normalizeRef accepted invalid ref") + } +} diff --git a/go.mod b/go.mod index e1a1ed16a..7df3e9b84 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/charmbracelet/bubbletea v1.3.10 github.com/charmbracelet/lipgloss v1.1.0 github.com/containerd/containerd/v2 v2.3.2 + github.com/distribution/reference v0.6.0 github.com/go-chi/chi/v5 v5.3.0 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 @@ -73,7 +74,6 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dgraph-io/badger/v4 v4.9.2 // indirect github.com/dgraph-io/ristretto/v2 v2.4.0 // indirect - github.com/distribution/reference v0.6.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/fatih/color v1.19.0 // indirect diff --git a/scripts/ecs-node-bringup.sh b/scripts/ecs-node-bringup.sh index 5c7681d0e..37017b2f1 100755 --- a/scripts/ecs-node-bringup.sh +++ b/scripts/ecs-node-bringup.sh @@ -14,7 +14,7 @@ # AWS_PROFILE=spinifex scripts/ecs-node-bringup.sh # # Env overrides: -# GATEWAY_IP gateway/mgmt IP the guest dials (default 10.15.8.1) +# GATEWAY_IP gateway IP the guest dials (default: host WAN-bridge IP) # GATEWAY_CA host path to the gateway CA PEM (default /etc/spinifex/ca.pem) # AMI_ID ECS-node AMI (default: latest spinifex-ecs-node) # SUBNET_ID / SG_ID / KEY_NAME / INSTANCE_TYPE (default: first available / t3.small) @@ -23,6 +23,14 @@ set -euo pipefail CLUSTER="${1:?usage: ecs-node-bringup.sh }" PROFILE="${AWS_PROFILE:-spinifex}" REGION="$(aws configure get region --profile "$PROFILE")" + +# VPC guests NAT out through the WAN and reach host-run services on the host's +# WAN-bridge IP; the mgmt bridge (10.15.8.x) is not routable from a guest netns. +# Default to the br-wan address, falling back to the mgmt IP if it can't be read. +default_gateway_ip() { + ip -4 addr show br-wan 2>/dev/null | awk '/inet /{print $2; exit}' | cut -d/ -f1 +} +GATEWAY_IP="${GATEWAY_IP:-$(default_gateway_ip)}" GATEWAY_IP="${GATEWAY_IP:-10.15.8.1}" GATEWAY_CA="${GATEWAY_CA:-/etc/spinifex/ca.pem}" INSTANCE_TYPE="${INSTANCE_TYPE:-t3.small}" diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index db87d8d6e..2da1430c3 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -378,6 +378,7 @@ var ( // __type; do not cross-wire with the EC2/ACM not-found codes above. ErrorECSClusterNotFound = "ClusterNotFoundException" ErrorECSInvalidParameter = "InvalidParameterException" + ErrorECSServiceNotFound = "ServiceNotFoundException" ErrorRetryableError = "RetryableError" ErrorRouteAlreadyExists = "RouteAlreadyExists" ErrorRouteLimitExceeded = "RouteLimitExceeded" @@ -900,6 +901,7 @@ var ErrorLookup = map[string]ErrorMessage{ ErrorEKSResourceNotFound: {HTTPCode: 404, Message: "The specified cluster could not be found. You can view your available clusters with ListClusters. Amazon EKS clusters are Region specific."}, ErrorECSClusterNotFound: {HTTPCode: 400, Message: "The specified cluster was not found. You can view your available clusters with ListClusters. Amazon ECS clusters are Region specific."}, ErrorECSInvalidParameter: {HTTPCode: 400, Message: "The specified parameter is not valid. Review the available parameters for the API request."}, + ErrorECSServiceNotFound: {HTTPCode: 400, Message: "The specified service was not found. You can view your available services with ListServices. Amazon ECS services are cluster specific and Region specific."}, ErrorRetryableError: {HTTPCode: 400, Message: "A request submitted by an AWS service on your behalf could not be completed. The requesting service might automatically retry the request."}, ErrorRouteAlreadyExists: {HTTPCode: 409, Message: "A route for the specified CIDR block already exists in this route table."}, ErrorRouteLimitExceeded: {HTTPCode: 400, Message: "You've reached the limit on the number of routes that you can add to a route table."}, diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index bff7dc316..5a59e77c3 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -963,8 +963,15 @@ func (d *Daemon) subscribeAll() error { natsSub{"ecs.DescribeContainerInstances", d.handleECSDescribeContainerInstances, "spinifex-workers"}, natsSub{"ecs.ListContainerInstances", d.handleECSListContainerInstances, "spinifex-workers"}, natsSub{"ecs.RunTask", d.handleECSRunTask, "spinifex-workers"}, + natsSub{"ecs.StartTask", d.handleECSStartTask, "spinifex-workers"}, + natsSub{"ecs.StopTask", d.handleECSStopTask, "spinifex-workers"}, natsSub{"ecs.DescribeTasks", d.handleECSDescribeTasks, "spinifex-workers"}, natsSub{"ecs.ListTasks", d.handleECSListTasks, "spinifex-workers"}, + natsSub{"ecs.CreateService", d.handleECSCreateService, "spinifex-workers"}, + natsSub{"ecs.UpdateService", d.handleECSUpdateService, "spinifex-workers"}, + natsSub{"ecs.DeleteService", d.handleECSDeleteService, "spinifex-workers"}, + natsSub{"ecs.DescribeServices", d.handleECSDescribeServices, "spinifex-workers"}, + natsSub{"ecs.ListServices", d.handleECSListServices, "spinifex-workers"}, natsSub{"ecs.SubmitTaskStateChange", d.handleECSSubmitTaskStateChange, "spinifex-workers"}, natsSub{"ecs.PollAssignments", d.handleECSPollAssignments, "spinifex-workers"}, ) diff --git a/spinifex/daemon/daemon_handlers_ecs.go b/spinifex/daemon/daemon_handlers_ecs.go index d4eb017a0..8891a63f7 100644 --- a/spinifex/daemon/daemon_handlers_ecs.go +++ b/spinifex/daemon/daemon_handlers_ecs.go @@ -50,10 +50,40 @@ func (d *Daemon) handleECSRunTask(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.RunTask) } +func (d *Daemon) handleECSStartTask(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.StartTask) +} + +func (d *Daemon) handleECSStopTask(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.StopTask) +} + func (d *Daemon) handleECSDescribeTasks(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.DescribeTasks) } +// --- Service --- + +func (d *Daemon) handleECSCreateService(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.CreateService) +} + +func (d *Daemon) handleECSUpdateService(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.UpdateService) +} + +func (d *Daemon) handleECSDeleteService(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DeleteService) +} + +func (d *Daemon) handleECSDescribeServices(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.DescribeServices) +} + +func (d *Daemon) handleECSListServices(msg *nats.Msg) { + handleNATSRequest(msg, d.ecsService.ListServices) +} + func (d *Daemon) handleECSSubmitTaskStateChange(msg *nats.Msg) { handleNATSRequest(msg, d.ecsService.SubmitTaskStateChange) } diff --git a/spinifex/gateway/ecs/actions.go b/spinifex/gateway/ecs/actions.go index d493bd2d7..4be39f4e5 100644 --- a/spinifex/gateway/ecs/actions.go +++ b/spinifex/gateway/ecs/actions.go @@ -104,6 +104,22 @@ func RunTask(nc *nats.Conn, accountID string, body []byte) (any, error) { return handlers_ecs.NewNATSECSService(nc).RunTask(input, accountID) } +func StartTask(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.StartTaskInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).StartTask(input, accountID) +} + +func StopTask(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.StopTaskInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).StopTask(input, accountID) +} + func DescribeTasks(nc *nats.Conn, accountID string, body []byte) (any, error) { input := new(ecs.DescribeTasksInput) if err := unmarshalIfBody(body, input); err != nil { @@ -120,6 +136,48 @@ func ListTasks(nc *nats.Conn, accountID string, body []byte) (any, error) { return handlers_ecs.NewNATSECSService(nc).ListTasks(input, accountID) } +// --- Service --- + +func CreateService(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.CreateServiceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).CreateService(input, accountID) +} + +func UpdateService(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.UpdateServiceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).UpdateService(input, accountID) +} + +func DeleteService(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DeleteServiceInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DeleteService(input, accountID) +} + +func DescribeServices(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.DescribeServicesInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).DescribeServices(input, accountID) +} + +func ListServices(nc *nats.Conn, accountID string, body []byte) (any, error) { + input := new(ecs.ListServicesInput) + if err := unmarshalIfBody(body, input); err != nil { + return nil, err + } + return handlers_ecs.NewNATSECSService(nc).ListServices(input, accountID) +} + // SubmitTaskStateChange is the agent's task-state report path over the gateway // (replaces the Layer-2 bus publish). The account is the SigV4 caller, so an // instance cannot report state for another account's task. diff --git a/spinifex/gateway/ecs/handler.go b/spinifex/gateway/ecs/handler.go index a2eeae57e..4a8cdcd4d 100644 --- a/spinifex/gateway/ecs/handler.go +++ b/spinifex/gateway/ecs/handler.go @@ -56,10 +56,10 @@ var Actions = map[string]Handler{ "ListTaskDefinitions": ListTaskDefinitions, "ListTaskDefinitionFamilies": NotImplemented, - // Task. StartTask/StopTask land with the 4f service controller + reaper. + // Task. "RunTask": RunTask, - "StartTask": NotImplemented, - "StopTask": NotImplemented, + "StartTask": StartTask, + "StopTask": StopTask, "DescribeTasks": DescribeTasks, "ListTasks": ListTasks, @@ -70,11 +70,11 @@ var Actions = map[string]Handler{ "PollAssignments": PollAssignments, // Service. ListServicesByNamespace is a no-op stub in v1. - "CreateService": NotImplemented, - "UpdateService": NotImplemented, - "DeleteService": NotImplemented, - "DescribeServices": NotImplemented, - "ListServices": NotImplemented, + "CreateService": CreateService, + "UpdateService": UpdateService, + "DeleteService": DeleteService, + "DescribeServices": DescribeServices, + "ListServices": ListServices, "ListServicesByNamespace": NotImplemented, // Container instance. diff --git a/spinifex/gateway/ecs/handler_test.go b/spinifex/gateway/ecs/handler_test.go index e0204552e..03ee91e2c 100644 --- a/spinifex/gateway/ecs/handler_test.go +++ b/spinifex/gateway/ecs/handler_test.go @@ -24,7 +24,9 @@ var wiredActions = map[string]bool{ "CreateCluster": true, "DescribeClusters": true, "ListClusters": true, "RegisterTaskDefinition": true, "DescribeTaskDefinition": true, "ListTaskDefinitions": true, "RegisterContainerInstance": true, "DescribeContainerInstances": true, "ListContainerInstances": true, - "RunTask": true, "DescribeTasks": true, "ListTasks": true, + "RunTask": true, "StartTask": true, "StopTask": true, "DescribeTasks": true, "ListTasks": true, + "CreateService": true, "UpdateService": true, "DeleteService": true, + "DescribeServices": true, "ListServices": true, "SubmitTaskStateChange": true, "PollAssignments": true, } diff --git a/spinifex/handlers/ecs/instances.go b/spinifex/handlers/ecs/instances.go index 1e7aee0d3..f42cead93 100644 --- a/spinifex/handlers/ecs/instances.go +++ b/spinifex/handlers/ecs/instances.go @@ -225,8 +225,15 @@ func (s *Service) recordTaskState(msg *bus.TaskState) error { return err } - // Release capacity + reclaim the task ENI once, on the transition into STOPPED. + // Register the task's ELBv2 targets on the transition into RUNNING (Q8). + if msg.LastStatus == TaskStatusRunning && prev != TaskStatusRunning { + s.registerServiceTargets(kv, msg.AccountID, &task) + } + + // Deregister targets, release capacity + reclaim the task ENI once, on the + // transition into STOPPED. if msg.LastStatus == TaskStatusStopped && prev != TaskStatusStopped { + s.deregisterServiceTargets(kv, msg.AccountID, &task) s.reclaimAssignInbox(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID) s.reclaimTaskENI(msg.AccountID, &task) return s.releaseReservation(kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB) diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index 6836d109f..a43cf974c 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -20,6 +20,15 @@ const ( TaskStatusPending = bus.TaskStatusPending TaskStatusRunning = bus.TaskStatusRunning TaskStatusStopped = bus.TaskStatusStopped + + ServiceStatusActive = "ACTIVE" + ServiceStatusDraining = "DRAINING" + ServiceStatusInactive = "INACTIVE" + + // SchedulingStrategyReplica is the only strategy supported in v1 (Q15); + // DAEMON is rejected at CreateService. + SchedulingStrategyReplica = "REPLICA" + SchedulingStrategyDaemon = "DAEMON" ) // ARN builders for the ECS resource shapes (ecs-v1.md §1). Region + accountID @@ -41,6 +50,27 @@ func ContainerInstanceARN(region, accountID, cluster, ciID string) string { return fmt.Sprintf("arn:aws:ecs:%s:%s:container-instance/%s/%s", region, accountID, cluster, ciID) } +func ServiceARN(region, accountID, cluster, name string) string { + return fmt.Sprintf("arn:aws:ecs:%s:%s:service/%s/%s", region, accountID, cluster, 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. +func serviceTaskGroup(name string) string { + return "service:" + name +} + +// serviceNameFromGroup returns the service name encoded in a task group, or "" +// when the group is not a service group. +func serviceNameFromGroup(group string) string { + const p = "service:" + if len(group) > len(p) && group[:len(p)] == p { + return group[len(p):] + } + return "" +} + // ClusterRecord is the persisted cluster meta at ClusterMetaKey. type ClusterRecord struct { Name string `json:"name"` @@ -129,9 +159,14 @@ type ContainerState struct { // TaskRecord is the persisted task state at TaskKey; source of truth for // DescribeTasks and the placement/capacity accounting. type TaskRecord struct { - TaskID string `json:"taskId"` - ARN string `json:"arn"` - Cluster string `json:"cluster"` + TaskID string `json:"taskId"` + ARN string `json:"arn"` + Cluster string `json:"cluster"` + // 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"` @@ -157,3 +192,39 @@ type TaskRecord struct { StartedAt time.Time `json:"startedAt,omitzero"` StoppedAt time.Time `json:"stoppedAt,omitzero"` } + +// LoadBalancerTarget is one ELBv2 target-group attachment on a service. On task +// RUNNING the scheduler registers the task's ENI IP at ContainerPort; on STOPPED +// it deregisters (ecs-v1.md Q8, single-writer). +type LoadBalancerTarget struct { + TargetGroupARN string `json:"targetGroupArn"` + ContainerName string `json:"containerName,omitempty"` + ContainerPort int `json:"containerPort"` +} + +// ServiceRecord is the persisted service state at ServiceKey. The captured +// network config + placement strategy let the reconciler launch replacement +// tasks identically to the original RunTask. +type ServiceRecord struct { + Name string `json:"name"` + ARN string `json:"arn"` + Cluster string `json:"cluster"` + TaskDefFamily string `json:"taskDefFamily"` + TaskDefRevision int `json:"taskDefRevision"` + TaskDefARN string `json:"taskDefArn"` + DesiredCount int `json:"desiredCount"` + Status string `json:"status"` + SchedulingStrategy string `json:"schedulingStrategy"` + LaunchType string `json:"launchType,omitempty"` + NetworkMode string `json:"networkMode,omitempty"` + Subnets []string `json:"subnets,omitempty"` + SecurityGroups []string `json:"securityGroups,omitempty"` + AssignPublicIP string `json:"assignPublicIp,omitempty"` + PlacementStrategy string `json:"placementStrategy,omitempty"` + LoadBalancers []LoadBalancerTarget `json:"loadBalancers,omitempty"` + DeploymentID string `json:"deploymentId"` + RunningCount int `json:"runningCount"` + PendingCount int `json:"pendingCount"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} diff --git a/spinifex/handlers/ecs/scheduler.go b/spinifex/handlers/ecs/scheduler.go index b9e893607..1ecf74fa2 100644 --- a/spinifex/handlers/ecs/scheduler.go +++ b/spinifex/handlers/ecs/scheduler.go @@ -19,6 +19,7 @@ const ( schedulerLeaderKey = "scheduler" leaseRefresh = 20 * time.Second reaperInterval = 30 * time.Second + reconcileInterval = 10 * time.Second heartbeatTimeout = 90 * time.Second stoppedReasonReaped = "ContainerInstance disconnected" ) @@ -47,8 +48,10 @@ func NewScheduler(nc *nats.Conn, svc *Service, holder string) *Scheduler { func (sc *Scheduler) Run(ctx context.Context) { leaseTicker := time.NewTicker(leaseRefresh) reaperTicker := time.NewTicker(reaperInterval) + reconcileTicker := time.NewTicker(reconcileInterval) defer leaseTicker.Stop() defer reaperTicker.Stop() + defer reconcileTicker.Stop() sc.evaluateLeadership(ctx) for { @@ -62,6 +65,10 @@ func (sc *Scheduler) Run(ctx context.Context) { if sc.isLeader() { sc.reap() } + case <-reconcileTicker.C: + if sc.isLeader() { + sc.svc.reconcileAllServices() + } } } } @@ -271,13 +278,7 @@ func (sc *Scheduler) stopInstanceTasks(kv nats.KeyValue, accountID, cluster, ins if task.ContainerInstanceID != instanceID || task.LastStatus == TaskStatusStopped { continue } - task.LastStatus = TaskStatusStopped - task.StoppedReason = stoppedReasonReaped - task.StoppedAt = time.Now().UTC() - if perr := putJSON(kv, k, &task); perr != nil { - continue - } - sc.svc.reclaimTaskENI(accountID, &task) + sc.svc.forceStopTask(kv, accountID, &task, stoppedReasonReaped) } } diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index d36f59505..18db20cc1 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -32,9 +32,17 @@ type ECSService interface { ListContainerInstances(input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTaskOutput, error) + StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, error) + StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) ListTasks(input *ecs.ListTasksInput, accountID string) (*ecs.ListTasksOutput, error) + CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error) + UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error) + DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error) + DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error) + ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error) + SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeInput, accountID string) (*ecs.SubmitTaskStateChangeOutput, error) // PollAssignments drains an instance's task-assignment inbox (agent → gateway). @@ -51,6 +59,9 @@ type Service struct { // eni owns the awsvpc task-ENI control-plane (create/attach/detach/delete). // Defaults to the NATS-backed controller; tests substitute a stub. eni eniController + // targets registers/deregisters service tasks with ELBv2 target groups. + // Defaults to the NATS-backed registrar; tests substitute a stub. + targets targetRegistrar } var _ ECSService = (*Service)(nil) @@ -59,7 +70,7 @@ var _ ECSService = (*Service)(nil) // ARNs it mints; suffix is the AWS-parity internal DNS suffix (reserved for ECR // endpoint composition). func NewService(nc *nats.Conn, region, suffix string) *Service { - return &Service{nc: nc, region: region, suffix: suffix, eni: newNATSENIController(nc)} + return &Service{nc: nc, region: region, suffix: suffix, eni: newNATSENIController(nc), targets: newNATSTargetRegistrar(nc)} } // defaultCluster is the implicit cluster name when a request omits one. diff --git a/spinifex/handlers/ecs/service_nats.go b/spinifex/handlers/ecs/service_nats.go index 750729d5c..996b5b45d 100644 --- a/spinifex/handlers/ecs/service_nats.go +++ b/spinifex/handlers/ecs/service_nats.go @@ -71,6 +71,14 @@ func (s *NATSECSService) RunTask(input *ecs.RunTaskInput, accountID string) (*ec return utils.NATSRequest[ecs.RunTaskOutput](s.natsConn, "ecs.RunTask", input, defaultTimeout, accountID) } +func (s *NATSECSService) StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, error) { + return utils.NATSRequest[ecs.StartTaskOutput](s.natsConn, "ecs.StartTask", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) { + return utils.NATSRequest[ecs.StopTaskOutput](s.natsConn, "ecs.StopTask", input, defaultTimeout, accountID) +} + func (s *NATSECSService) DescribeTasks(input *ecs.DescribeTasksInput, accountID string) (*ecs.DescribeTasksOutput, error) { return utils.NATSRequest[ecs.DescribeTasksOutput](s.natsConn, "ecs.DescribeTasks", input, defaultTimeout, accountID) } @@ -83,6 +91,28 @@ func (s *NATSECSService) SubmitTaskStateChange(input *ecs.SubmitTaskStateChangeI return utils.NATSRequest[ecs.SubmitTaskStateChangeOutput](s.natsConn, "ecs.SubmitTaskStateChange", input, defaultTimeout, accountID) } +// --- Service --- + +func (s *NATSECSService) CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error) { + return utils.NATSRequest[ecs.CreateServiceOutput](s.natsConn, "ecs.CreateService", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error) { + return utils.NATSRequest[ecs.UpdateServiceOutput](s.natsConn, "ecs.UpdateService", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error) { + return utils.NATSRequest[ecs.DeleteServiceOutput](s.natsConn, "ecs.DeleteService", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error) { + return utils.NATSRequest[ecs.DescribeServicesOutput](s.natsConn, "ecs.DescribeServices", input, defaultTimeout, accountID) +} + +func (s *NATSECSService) ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error) { + return utils.NATSRequest[ecs.ListServicesOutput](s.natsConn, "ecs.ListServices", input, defaultTimeout, accountID) +} + func (s *NATSECSService) PollAssignments(input *PollAssignmentsInput, accountID string) (*PollAssignmentsOutput, error) { return utils.NATSRequest[PollAssignmentsOutput](s.natsConn, "ecs.PollAssignments", input, defaultTimeout, accountID) } diff --git a/spinifex/handlers/ecs/services.go b/spinifex/handlers/ecs/services.go new file mode 100644 index 000000000..1d386fa5d --- /dev/null +++ b/spinifex/handlers/ecs/services.go @@ -0,0 +1,520 @@ +package handlers_ecs + +import ( + "errors" + "log/slog" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/aws/aws-sdk-go/service/elbv2" + "github.com/google/uuid" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" + "github.com/nats-io/nats.go" +) + +// targetRegistrar registers/deregisters a service task's awsvpc ENI IP with an +// ELBv2 target group. The scheduler is the single writer (ecs-v1.md Q8): it +// registers on the RUNNING transition and deregisters on STOPPED. +type targetRegistrar interface { + Register(accountID, tgARN, ip string, port int) error + Deregister(accountID, tgARN, ip string, port int) error +} + +// natsTargetRegistrar drives the existing elbv2.RegisterTargets/DeregisterTargets +// NATS handlers; no new ELBv2 surface is added for ECS (Q16). +type natsTargetRegistrar struct { + nc *nats.Conn +} + +var _ targetRegistrar = (*natsTargetRegistrar)(nil) + +func newNATSTargetRegistrar(nc *nats.Conn) *natsTargetRegistrar { + return &natsTargetRegistrar{nc: nc} +} + +func (r *natsTargetRegistrar) Register(accountID, tgARN, ip string, port int) error { + _, err := handlers_elbv2.NewNATSELBv2Service(r.nc).RegisterTargets(&elbv2.RegisterTargetsInput{ + TargetGroupArn: aws.String(tgARN), + Targets: []*elbv2.TargetDescription{{Id: aws.String(ip), Port: aws.Int64(int64(port))}}, + }, accountID) + return err +} + +func (r *natsTargetRegistrar) Deregister(accountID, tgARN, ip string, port int) error { + _, err := handlers_elbv2.NewNATSELBv2Service(r.nc).DeregisterTargets(&elbv2.DeregisterTargetsInput{ + TargetGroupArn: aws.String(tgARN), + Targets: []*elbv2.TargetDescription{{Id: aws.String(ip), Port: aws.Int64(int64(port))}}, + }, accountID) + return err +} + +// --- Service API --- + +// CreateService persists a REPLICA service and reconciles up to desiredCount. +// DAEMON scheduling and serviceRegistries are rejected in v1 (Q15). Re-creating +// an existing ACTIVE service returns the existing record (idempotent). +func (s *Service) CreateService(input *ecs.CreateServiceInput, accountID string) (*ecs.CreateServiceOutput, error) { + name := aws.StringValue(input.ServiceName) + if name == "" { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + if strategy := aws.StringValue(input.SchedulingStrategy); strategy != "" && strategy != SchedulingStrategyReplica { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) // DAEMON unsupported (Q15) + } + if len(input.ServiceRegistries) > 0 { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) // Route53 v0 not landed (Q15) + } + + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var clusterRec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(cluster), &clusterRec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSClusterNotFound) + } + + var existing ServiceRecord + if ok, gerr := getJSON(kv, ServiceKey(cluster, name), &existing); gerr != nil { + return nil, gerr + } else if ok && existing.Status == ServiceStatusActive { + return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &existing)}, nil + } + + taskDef, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition)) + if err != nil { + return nil, err + } + + now := time.Now().UTC() + rec := ServiceRecord{ + Name: name, + ARN: ServiceARN(s.region, accountID, cluster, name), + Cluster: cluster, + TaskDefFamily: taskDef.Family, + TaskDefRevision: taskDef.Revision, + TaskDefARN: taskDef.ARN, + DesiredCount: int(aws.Int64Value(input.DesiredCount)), + Status: ServiceStatusActive, + SchedulingStrategy: SchedulingStrategyReplica, + LaunchType: aws.StringValue(input.LaunchType), + NetworkMode: resolveNetworkMode(taskDef), + PlacementStrategy: placementStrategyFromAWS(input.PlacementStrategy), + LoadBalancers: loadBalancersFromAWS(input.LoadBalancers), + DeploymentID: uuid.NewString(), + CreatedAt: now, + UpdatedAt: now, + } + if input.NetworkConfiguration != nil && input.NetworkConfiguration.AwsvpcConfiguration != nil { + v := input.NetworkConfiguration.AwsvpcConfiguration + rec.Subnets = awsStringSlice(v.Subnets) + rec.SecurityGroups = awsStringSlice(v.SecurityGroups) + rec.AssignPublicIP = aws.StringValue(v.AssignPublicIp) + } + if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil { + return nil, err + } + + if err := s.reconcileService(kv, accountID, &rec); err != nil { + slog.Error("ECS CreateService: initial reconcile failed", "service", name, "err", err) + } + return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil +} + +// UpdateService mutates desiredCount and/or the task definition, then reconciles. +func (s *Service) UpdateService(input *ecs.UpdateServiceInput, accountID string) (*ecs.UpdateServiceOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + name := serviceShortName(aws.StringValue(input.Service)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, name), &rec) + if err != nil { + return nil, err + } + if !found || rec.Status == ServiceStatusInactive { + return nil, errors.New(awserrors.ErrorECSServiceNotFound) + } + + if input.DesiredCount != nil { + rec.DesiredCount = int(aws.Int64Value(input.DesiredCount)) + } + if td := aws.StringValue(input.TaskDefinition); td != "" { + taskDef, terr := s.resolveTaskDef(kv, td) + 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() + } + rec.UpdatedAt = time.Now().UTC() + if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil { + return nil, err + } + if err := s.reconcileService(kv, accountID, &rec); err != nil { + slog.Error("ECS UpdateService: reconcile failed", "service", name, "err", err) + } + return &ecs.UpdateServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil +} + +// DeleteService drains the service to zero, force-stops its tasks, and marks it +// INACTIVE (kept describable, AWS parity). +func (s *Service) DeleteService(input *ecs.DeleteServiceInput, accountID string) (*ecs.DeleteServiceOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + name := serviceShortName(aws.StringValue(input.Service)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var rec ServiceRecord + found, err := getJSON(kv, ServiceKey(cluster, name), &rec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSServiceNotFound) + } + + rec.DesiredCount = 0 + tasks, err := s.listServiceTasks(kv, cluster, name) + if err != nil { + return nil, err + } + for i := range tasks { + s.forceStopTask(kv, accountID, &tasks[i], "Service deleted") + } + rec.Status = ServiceStatusInactive + rec.RunningCount = 0 + rec.PendingCount = 0 + rec.UpdatedAt = time.Now().UTC() + if err := putJSON(kv, ServiceKey(cluster, name), &rec); err != nil { + return nil, err + } + return &ecs.DeleteServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil +} + +// DescribeServices returns the named services; unknown names surface as failures. +func (s *Service) DescribeServices(input *ecs.DescribeServicesInput, accountID string) (*ecs.DescribeServicesOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + out := &ecs.DescribeServicesOutput{} + for _, ref := range awsStringSlice(input.Services) { + name := serviceShortName(ref) + var rec ServiceRecord + found, gerr := getJSON(kv, ServiceKey(cluster, name), &rec) + if gerr != nil { + return nil, gerr + } + if found { + out.Services = append(out.Services, s.serviceToAWS(accountID, &rec)) + } else { + out.Failures = append(out.Failures, &ecs.Failure{Arn: aws.String(ref), Reason: aws.String("MISSING")}) + } + } + return out, nil +} + +// ListServices returns the ARNs of every service in a cluster. +func (s *Service) ListServices(input *ecs.ListServicesInput, accountID string) (*ecs.ListServicesOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + recs, err := s.listServiceRecords(kv, cluster) + if err != nil { + return nil, err + } + out := &ecs.ListServicesOutput{} + for i := range recs { + out.ServiceArns = append(out.ServiceArns, aws.String(recs[i].ARN)) + } + return out, nil +} + +// --- 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. +func (s *Service) reconcileService(kv nats.KeyValue, accountID string, svc *ServiceRecord) error { + if svc.Status != ServiceStatusActive { + return nil + } + tasks, err := s.listServiceTasks(kv, svc.Cluster, svc.Name) + if err != nil { + return err + } + + running, pending := 0, 0 + for i := range tasks { + switch tasks[i].LastStatus { + case TaskStatusRunning: + running++ + case TaskStatusPending: + pending++ + } + } + + 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) + 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() { + js, err := s.js() + if err != nil { + return + } + for bucket := range js.KeyValueStoreNames() { + accountID, ok := accountIDFromBucket(bucket) + if !ok { + continue + } + kv, err := js.KeyValue(bucket) + if err != nil { + continue + } + recs, err := s.allServiceRecords(kv) + if err != nil { + continue + } + for i := range recs { + if err := s.reconcileService(kv, accountID, &recs[i]); err != nil { + slog.Error("ECS reconcile: service failed", "service", recs[i].Name, "err", err) + } + } + } +} + +// 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) { + in := &ecs.RunTaskInput{ + Cluster: aws.String(svc.Cluster), + TaskDefinition: aws.String(svc.TaskDefARN), + Count: aws.Int64(int64(n)), + Group: aws.String(serviceTaskGroup(svc.Name)), + StartedBy: aws.String("ecs-svc/" + svc.DeploymentID), + } + if svc.PlacementStrategy != "" { + in.PlacementStrategy = []*ecs.PlacementStrategy{{Type: aws.String(svc.PlacementStrategy)}} + } + if svc.NetworkMode == NetworkModeAwsvpc { + in.NetworkConfiguration = &ecs.NetworkConfiguration{ + AwsvpcConfiguration: &ecs.AwsVpcConfiguration{ + Subnets: aws.StringSlice(svc.Subnets), + SecurityGroups: aws.StringSlice(svc.SecurityGroups), + }, + } + if svc.AssignPublicIP != "" { + in.NetworkConfiguration.AwsvpcConfiguration.AssignPublicIp = aws.String(svc.AssignPublicIP) + } + } + out, err := s.RunTask(in, accountID) + if err != nil { + slog.Error("ECS reconcile: launch failed", "service", svc.Name, "err", err) + return + } + if len(out.Failures) > 0 { + slog.Warn("ECS reconcile: launch had placement failures", "service", svc.Name, "failures", len(out.Failures)) + } +} + +// 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)) + for i := range tasks { + if tasks[i].LastStatus == TaskStatusPending { + order = append(order, i) + } + } + for i := range tasks { + if tasks[i].LastStatus == TaskStatusRunning { + order = append(order, i) + } + } + for _, idx := range order { + if n <= 0 { + break + } + s.forceStopTask(kv, accountID, &tasks[idx], "Service scaled in") + n-- + } +} + +// --- Helpers --- + +// listServiceTasks returns a cluster's non-STOPPED tasks owned by the service. +func (s *Service) listServiceTasks(kv nats.KeyValue, cluster, name string) ([]TaskRecord, error) { + group := serviceTaskGroup(name) + all, err := s.listTaskRecords(kv, cluster) + if err != nil { + return nil, err + } + out := make([]TaskRecord, 0, len(all)) + for _, t := range all { + if t.Group == group && t.LastStatus != TaskStatusStopped { + out = append(out, t) + } + } + return out, nil +} + +// listTaskRecords returns every task record in a cluster. +func (s *Service) listTaskRecords(kv nats.KeyValue, cluster string) ([]TaskRecord, error) { + keys, err := keysWithPrefix(kv, TasksPrefix(cluster)) + if err != nil { + return nil, err + } + out := make([]TaskRecord, 0, len(keys)) + for _, k := range keys { + var rec TaskRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out = append(out, rec) + } + } + return out, nil +} + +// listServiceRecords returns every service record in a cluster. +func (s *Service) listServiceRecords(kv nats.KeyValue, cluster string) ([]ServiceRecord, error) { + keys, err := keysWithPrefix(kv, ServicesPrefix(cluster)) + if err != nil { + return nil, err + } + out := make([]ServiceRecord, 0, len(keys)) + for _, k := range keys { + var rec ServiceRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out = append(out, rec) + } + } + return out, nil +} + +// allServiceRecords returns every service record across all clusters in a bucket. +func (s *Service) allServiceRecords(kv nats.KeyValue) ([]ServiceRecord, error) { + keys, err := keysWithPrefix(kv, "clusters/") + if err != nil { + return nil, err + } + out := make([]ServiceRecord, 0) + for _, k := range keys { + if !strings.Contains(k, "/services/") { + continue + } + var rec ServiceRecord + found, err := getJSON(kv, k, &rec) + if err != nil { + return nil, err + } + if found { + out = append(out, rec) + } + } + return out, nil +} + +// serviceShortName extracts the service name from a name or a service ARN. +func serviceShortName(ref string) string { + ref = strings.TrimSpace(ref) + if i := strings.LastIndexByte(ref, '/'); i >= 0 { + return ref[i+1:] + } + return ref +} + +// loadBalancersFromAWS maps the SDK loadBalancer list to the persisted subset. +func loadBalancersFromAWS(in []*ecs.LoadBalancer) []LoadBalancerTarget { + out := make([]LoadBalancerTarget, 0, len(in)) + for _, lb := range in { + if lb == nil || aws.StringValue(lb.TargetGroupArn) == "" { + continue + } + out = append(out, LoadBalancerTarget{ + TargetGroupARN: aws.StringValue(lb.TargetGroupArn), + ContainerName: aws.StringValue(lb.ContainerName), + ContainerPort: int(aws.Int64Value(lb.ContainerPort)), + }) + } + return out +} + +func (s *Service) serviceToAWS(accountID string, r *ServiceRecord) *ecs.Service { + svc := &ecs.Service{ + ServiceName: aws.String(r.Name), + ServiceArn: aws.String(r.ARN), + ClusterArn: aws.String(ClusterARN(s.region, accountID, r.Cluster)), + Status: aws.String(r.Status), + DesiredCount: aws.Int64(int64(r.DesiredCount)), + RunningCount: aws.Int64(int64(r.RunningCount)), + PendingCount: aws.Int64(int64(r.PendingCount)), + SchedulingStrategy: aws.String(r.SchedulingStrategy), + TaskDefinition: aws.String(r.TaskDefARN), + } + if r.LaunchType != "" { + svc.LaunchType = aws.String(r.LaunchType) + } + for _, lb := range r.LoadBalancers { + svc.LoadBalancers = append(svc.LoadBalancers, &ecs.LoadBalancer{ + TargetGroupArn: aws.String(lb.TargetGroupARN), + ContainerName: aws.String(lb.ContainerName), + ContainerPort: aws.Int64(int64(lb.ContainerPort)), + }) + } + return svc +} diff --git a/spinifex/handlers/ecs/services_test.go b/spinifex/handlers/ecs/services_test.go new file mode 100644 index 000000000..aea9323c2 --- /dev/null +++ b/spinifex/handlers/ecs/services_test.go @@ -0,0 +1,323 @@ +package handlers_ecs + +import ( + "fmt" + "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" +) + +// stubRegistrar records ELBv2 target calls so the RUNNING/STOPPED hooks can be +// asserted without a live elbv2 daemon. +type stubRegistrar struct { + registered []string + deregistered []string +} + +func key(tg, ip string, port int) string { return fmt.Sprintf("%s|%s|%d", tg, ip, port) } + +func (r *stubRegistrar) Register(_, tg, ip string, port int) error { + r.registered = append(r.registered, key(tg, ip, port)) + return nil +} + +func (r *stubRegistrar) Deregister(_, tg, ip string, port int) error { + r.deregistered = append(r.deregistered, key(tg, ip, port)) + return nil +} + +// serviceTestRig wires a Service with a stub registrar and a seeded cluster + +// task definition + one fat container instance. +func serviceTestRig(t *testing.T) (*Service, *stubRegistrar, nats.KeyValue) { + t.Helper() + svc, _ := newTestService(t) + reg := &stubRegistrar{} + svc.targets = reg + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + registerTaskDef(t, svc, "app", 128, 256) + registerInstance(t, svc, "web", "i-1", 4096, 8192) + kv, err := svc.bucket(testAccountID) + require.NoError(t, err) + return svc, reg, kv +} + +func TestService_CreateService_LaunchesReplicas(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) + assert.Equal(t, ServiceStatusActive, aws.StringValue(out.Service.Status)) + assert.Equal(t, ServiceARN(testRegion, testAccountID, "web", "web"), aws.StringValue(out.Service.ServiceArn)) + assert.Equal(t, int64(2), aws.Int64Value(out.Service.PendingCount)) + + tasks, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + require.Len(t, tasks, 2) + for _, task := range tasks { + assert.Equal(t, serviceTaskGroup("web"), task.Group) + assert.Equal(t, TaskStatusPending, task.LastStatus) + } +} + +func TestService_CreateService_RejectsDaemonAndRegistries(t *testing.T) { + svc, _, _ := serviceTestRig(t) + + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("d"), TaskDefinition: aws.String("app"), + SchedulingStrategy: aws.String(SchedulingStrategyDaemon), + }, testAccountID) + require.Error(t, err) + + _, err = svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("r"), TaskDefinition: aws.String("app"), + ServiceRegistries: []*ecs.ServiceRegistry{{RegistryArn: aws.String("arn:aws:servicediscovery:::service/x")}}, + }, testAccountID) + require.Error(t, err) +} + +func TestService_CreateService_UnknownCluster(t *testing.T) { + svc, _ := newTestService(t) + registerTaskDef(t, svc, "app", 1, 1) + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("ghost"), ServiceName: aws.String("s"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(1), + }, testAccountID) + require.Error(t, err) +} + +// Reconcile relaunches the desired count after a task STOPs. +func TestService_Reconcile_ReplacesStoppedTask(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) + + tasks, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + require.Len(t, tasks, 2) + + // One task stops (agent report). + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", + TaskID: tasks[0].TaskID, LastStatus: bus.TaskStatusStopped, + })) + live, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + require.Len(t, live, 1) // STOPPED task drops out of the active set + + // Reconcile launches a replacement back to desired=2. + svc.reconcileAllServices() + live, err = svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + assert.Len(t, live, 2) +} + +// UpdateService desiredCount=1 stops the surplus task. +func TestService_UpdateService_ScalesIn(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(3), + }, testAccountID) + require.NoError(t, err) + live, _ := svc.listServiceTasks(kv, "web", "web") + require.Len(t, live, 3) + + out, err := svc.UpdateService(&ecs.UpdateServiceInput{ + Cluster: aws.String("web"), Service: aws.String("web"), DesiredCount: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, int64(1), aws.Int64Value(out.Service.DesiredCount)) + + live, err = svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + assert.Len(t, live, 1) +} + +func TestService_DeleteService_DrainsAndInactivates(t *testing.T) { + svc, _, kv := serviceTestRig(t) + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(2), + }, testAccountID) + require.NoError(t, err) + + out, err := svc.DeleteService(&ecs.DeleteServiceInput{ + Cluster: aws.String("web"), Service: aws.String("web"), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, ServiceStatusInactive, aws.StringValue(out.Service.Status)) + + live, err := svc.listServiceTasks(kv, "web", "web") + require.NoError(t, err) + assert.Empty(t, live) + + // INACTIVE service still describes; reconcile leaves it alone. + desc, err := svc.DescribeServices(&ecs.DescribeServicesInput{ + Cluster: aws.String("web"), Services: []*string{aws.String("web")}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, desc.Services, 1) + svc.reconcileAllServices() + live, _ = svc.listServiceTasks(kv, "web", "web") + assert.Empty(t, live) +} + +func TestService_DescribeAndList(t *testing.T) { + svc, _, _ := serviceTestRig(t) + for _, name := range []string{"a", "b"} { + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String(name), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(0), + }, testAccountID) + require.NoError(t, err) + } + list, err := svc.ListServices(&ecs.ListServicesInput{Cluster: aws.String("web")}, testAccountID) + require.NoError(t, err) + assert.Len(t, list.ServiceArns, 2) + + miss, err := svc.DescribeServices(&ecs.DescribeServicesInput{ + Cluster: aws.String("web"), Services: []*string{aws.String("nope")}, + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, miss.Services) + assert.Len(t, miss.Failures, 1) +} + +// StartTask places one task per named container instance and assigns it. +func TestService_StartTask_PlacesPerInstance(t *testing.T) { + svc, _, kv := serviceTestRig(t) + + out, err := svc.StartTask(&ecs.StartTaskInput{ + Cluster: aws.String("web"), + TaskDefinition: aws.String("app"), + ContainerInstances: []*string{aws.String("i-1")}, + Group: aws.String("custom"), + StartedBy: aws.String("operator"), + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Tasks, 1) + assert.Empty(t, out.Failures) + assert.Equal(t, TaskStatusPending, aws.StringValue(out.Tasks[0].LastStatus)) + + tasks, err := svc.listTaskRecords(kv, "web") + require.NoError(t, err) + require.Len(t, tasks, 1) + assert.Equal(t, "custom", tasks[0].Group) + assert.Equal(t, "operator", tasks[0].StartedBy) + assert.Equal(t, "i-1", tasks[0].ContainerInstanceID) +} + +// StartTask onto an unknown / too-small instance returns a placement Failure, +// not a hard error, and onto an unknown cluster a ClusterNotFound error. +func TestService_StartTask_Failures(t *testing.T) { + svc, _, _ := serviceTestRig(t) + + out, err := svc.StartTask(&ecs.StartTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("app"), + ContainerInstances: []*string{aws.String("i-absent")}, + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, out.Tasks) + require.Len(t, out.Failures, 1) + + _, err = svc.StartTask(&ecs.StartTaskInput{ + Cluster: aws.String("ghost"), TaskDefinition: aws.String("app"), + ContainerInstances: []*string{aws.String("i-1")}, + }, testAccountID) + require.Error(t, err) +} + +// StopTask transitions a running service task to STOPPED, releases its +// reservation, and deregisters its ELBv2 targets. +func TestService_StopTask_StopsAndDeregisters(t *testing.T) { + svc, reg, kv := serviceTestRig(t) + const tgARN = "arn:aws:elasticloadbalancing:ap-southeast-2:123456789012:targetgroup/web/abc" + + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(0), + LoadBalancers: []*ecs.LoadBalancer{{ + TargetGroupArn: aws.String(tgARN), ContainerName: aws.String("app"), ContainerPort: aws.Int64(80), + }}, + }, testAccountID) + require.NoError(t, err) + + task := &TaskRecord{ + TaskID: "t-9", Cluster: "web", Group: serviceTaskGroup("web"), + ContainerInstanceID: "i-1", LastStatus: TaskStatusRunning, + NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9", + ReservedCPU: 64, ReservedMemoryMiB: 64, + } + require.NoError(t, putJSON(kv, TaskKey("web", "t-9"), task)) + + out, err := svc.StopTask(&ecs.StopTaskInput{ + Cluster: aws.String("web"), Task: aws.String("t-9"), Reason: aws.String("bye"), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, TaskStatusStopped, aws.StringValue(out.Task.LastStatus)) + assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered) + + // Idempotent: a second stop is a no-op (no extra deregister). + _, err = svc.StopTask(&ecs.StopTaskInput{Cluster: aws.String("web"), Task: aws.String("t-9")}, testAccountID) + require.NoError(t, err) + assert.Len(t, reg.deregistered, 1) +} + +func TestService_StopTask_NotFound(t *testing.T) { + svc, _, _ := serviceTestRig(t) + _, err := svc.StopTask(&ecs.StopTaskInput{Cluster: aws.String("web"), Task: aws.String("ghost")}, testAccountID) + require.Error(t, err) +} + +// A service with a target group registers each task's ENI IP on RUNNING and +// deregisters it on STOPPED (ecs-v1.md Q8, single-writer). +func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T) { + svc, reg, kv := serviceTestRig(t) + const tgARN = "arn:aws:elasticloadbalancing:ap-southeast-2:123456789012:targetgroup/web/abc" + + _, err := svc.CreateService(&ecs.CreateServiceInput{ + Cluster: aws.String("web"), ServiceName: aws.String("web"), TaskDefinition: aws.String("app"), + DesiredCount: aws.Int64(0), + LoadBalancers: []*ecs.LoadBalancer{{ + TargetGroupArn: aws.String(tgARN), ContainerName: aws.String("app"), ContainerPort: aws.Int64(80), + }}, + }, testAccountID) + require.NoError(t, err) + + // Seed an awsvpc service task with an ENI IP (bypasses live ENI provisioning). + task := &TaskRecord{ + TaskID: "t-1", Cluster: "web", Group: serviceTaskGroup("web"), + ContainerInstanceID: "i-1", LastStatus: TaskStatusPending, + NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.5", + } + require.NoError(t, putJSON(kv, TaskKey("web", "t-1"), task)) + + // RUNNING → register. + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", + TaskID: "t-1", LastStatus: bus.TaskStatusRunning, + })) + require.Equal(t, []string{key(tgARN, "10.0.1.5", 80)}, reg.registered) + assert.Empty(t, reg.deregistered) + + // STOPPED → deregister. + require.NoError(t, svc.recordTaskState(&bus.TaskState{ + AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", + TaskID: "t-1", LastStatus: bus.TaskStatusStopped, + })) + assert.Equal(t, []string{key(tgARN, "10.0.1.5", 80)}, reg.deregistered) +} diff --git a/spinifex/handlers/ecs/task_stop.go b/spinifex/handlers/ecs/task_stop.go new file mode 100644 index 000000000..c3d280ada --- /dev/null +++ b/spinifex/handlers/ecs/task_stop.go @@ -0,0 +1,211 @@ +package handlers_ecs + +import ( + "encoding/json" + "errors" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecs" + "github.com/google/uuid" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go" +) + +// defaultStopReason is recorded when StopTask is called without a reason. +const defaultStopReason = "Task stopped by user" + +// StopTask transitions a task to STOPPED via the control-plane forced-stop path +// (capacity + ENI release, TG deregister). The live container is torn down by the +// agent reconciler diffing KV against containerd (ecs-v1.md Q13). +func (s *Service) StopTask(input *ecs.StopTaskInput, accountID string) (*ecs.StopTaskOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + taskID := taskShortID(aws.StringValue(input.Task)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var task TaskRecord + found, err := getJSON(kv, TaskKey(cluster, taskID), &task) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + reason := aws.StringValue(input.Reason) + if reason == "" { + reason = defaultStopReason + } + s.forceStopTask(kv, accountID, &task, reason) + return &ecs.StopTaskOutput{Task: s.taskToAWS(accountID, &task)}, nil +} + +// StartTask places one task per named container instance (explicit placement, no +// scheduler bin-pack). Mirrors RunTask's reserve → record → ENI → assign flow. +func (s *Service) StartTask(input *ecs.StartTaskInput, accountID string) (*ecs.StartTaskOutput, error) { + cluster := clusterShortName(aws.StringValue(input.Cluster)) + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + var clusterRec ClusterRecord + found, err := getJSON(kv, ClusterMetaKey(cluster), &clusterRec) + if err != nil { + return nil, err + } + if !found { + return nil, errors.New(awserrors.ErrorECSClusterNotFound) + } + taskDef, err := s.resolveTaskDef(kv, aws.StringValue(input.TaskDefinition)) + if err != nil { + return nil, err + } + + mode := resolveNetworkMode(taskDef) + netCfg := awsvpcConfigFromStartTask(input) + if mode == NetworkModeAwsvpc && netCfg.firstSubnet() == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cpu, mem := taskDef.reservedCPU(), taskDef.reservedMemory() + group := aws.StringValue(input.Group) + startedBy := aws.StringValue(input.StartedBy) + + out := &ecs.StartTaskOutput{} + for _, ref := range awsStringSlice(input.ContainerInstances) { + instanceID := containerInstanceShortID(ref) + taskID := uuid.NewString() + inst, rerr := s.reserveOnInstance(kv, cluster, instanceID, taskID, cpu, mem) + if rerr != nil { + out.Failures = append(out.Failures, &ecs.Failure{ + Arn: aws.String(ref), Reason: aws.String("RESOURCE:placement"), Detail: aws.String(rerr.Error()), + }) + continue + } + rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem) + rec.NetworkMode = mode + rec.Group = group + rec.StartedBy = startedBy + if mode == NetworkModeAwsvpc { + if failure := s.provisionTaskENI(kv, accountID, cluster, rec, netCfg); failure != nil { + out.Failures = append(out.Failures, failure) + continue + } + } + if err := putJSON(kv, TaskKey(cluster, taskID), rec); err != nil { + return nil, err + } + if err := s.publishAssign(kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil { + slog.Error("ECS StartTask: failed to publish assign", "task", taskID, "instance", inst.InstanceID, "err", err) + } + out.Tasks = append(out.Tasks, s.taskToAWS(accountID, rec)) + } + return out, nil +} + +// reserveOnInstance commits a task's capacity reservation onto a specific +// container instance under a KV CAS. Unlike reservePlacement it does not bin-pack; +// the instance is fixed by the caller (StartTask). +func (s *Service) reserveOnInstance(kv nats.KeyValue, cluster, instanceID, taskID string, cpu, mem int) (*InstanceRecord, error) { + for range reservePlacementRetries { + entry, err := kv.Get(InstanceKey(cluster, instanceID)) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, errors.New(awserrors.ErrorECSInvalidParameter) + } + if err != nil { + return nil, err + } + var live InstanceRecord + if uerr := json.Unmarshal(entry.Value(), &live); uerr != nil { + return nil, uerr + } + if live.Status != InstanceStatusActive || !live.fits(cpu, mem) { + return nil, errors.New("instance has insufficient capacity") + } + live.ReservedCPU += cpu + live.ReservedMemoryMiB += mem + live.PlacedTasks = append(live.PlacedTasks, taskID) + data, merr := json.Marshal(&live) + if merr != nil { + return nil, merr + } + if _, uerr := kv.Update(InstanceKey(cluster, instanceID), data, entry.Revision()); uerr == nil { + return &live, nil + } + } + return nil, errors.New("reservation contended") +} + +// forceStopTask is the control-plane forced-stop: it transitions a task to +// STOPPED, releases its capacity + ENI, drains its assign inbox, and deregisters +// its ELBv2 targets. Idempotent and reused by StopTask, service scale-in, and the +// heartbeat reaper. The mutated record is reflected back into task. +func (s *Service) forceStopTask(kv nats.KeyValue, accountID string, task *TaskRecord, reason string) { + if task.LastStatus == TaskStatusStopped { + return + } + now := time.Now().UTC() + task.LastStatus = TaskStatusStopped + task.DesiredStatus = TaskStatusStopped + task.StoppedReason = reason + task.StoppedAt = now + if perr := putJSON(kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil { + slog.Error("ECS forceStopTask: persist failed", "task", task.TaskID, "err", perr) + return + } + s.deregisterServiceTargets(kv, accountID, task) + s.reclaimAssignInbox(kv, task.Cluster, task.ContainerInstanceID, task.TaskID) + s.reclaimTaskENI(accountID, task) + if rerr := s.releaseReservation(kv, task.Cluster, task.ContainerInstanceID, task.TaskID, task.ReservedCPU, task.ReservedMemoryMiB); rerr != nil { + slog.Error("ECS forceStopTask: release reservation failed", "task", task.TaskID, "err", rerr) + } +} + +// awsvpcConfigFromStartTask extracts the awsvpc subnet/SG selection from a +// StartTask input (StartTask has no bin-pack, so no RunTaskInput to reuse). +func awsvpcConfigFromStartTask(input *ecs.StartTaskInput) awsvpcConfig { + var cfg awsvpcConfig + if input != nil && input.NetworkConfiguration != nil && input.NetworkConfiguration.AwsvpcConfiguration != nil { + v := input.NetworkConfiguration.AwsvpcConfiguration + cfg.Subnets = awsStringSlice(v.Subnets) + cfg.SecurityGroups = awsStringSlice(v.SecurityGroups) + } + return cfg +} + +// registerServiceTargets registers a task's awsvpc ENI IP with each of its +// owning service's target groups. No-op when the task is not a service task, +// the service has no load balancers, or the task has no ENI IP (bridge/host). +func (s *Service) registerServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord) { + s.applyServiceTargets(kv, accountID, task, true) +} + +// deregisterServiceTargets is the STOPPED-side inverse of registerServiceTargets. +func (s *Service) deregisterServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord) { + s.applyServiceTargets(kv, accountID, task, false) +} + +func (s *Service) applyServiceTargets(kv nats.KeyValue, accountID string, task *TaskRecord, register bool) { + name := serviceNameFromGroup(task.Group) + if name == "" || task.ENIPrivateIP == "" || s.targets == nil { + return + } + var svc ServiceRecord + found, err := getJSON(kv, ServiceKey(task.Cluster, name), &svc) + if err != nil || !found || len(svc.LoadBalancers) == 0 { + return + } + for _, lb := range svc.LoadBalancers { + var terr error + if register { + terr = s.targets.Register(accountID, lb.TargetGroupARN, task.ENIPrivateIP, lb.ContainerPort) + } else { + terr = s.targets.Deregister(accountID, lb.TargetGroupARN, task.ENIPrivateIP, lb.ContainerPort) + } + if terr != nil { + slog.Error("ECS target registration failed", "service", name, "tg", lb.TargetGroupARN, + "register", register, "err", terr) + } + } +} diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go index 5ff20cb8c..174f39750 100644 --- a/spinifex/handlers/ecs/tasks.go +++ b/spinifex/handlers/ecs/tasks.go @@ -71,6 +71,8 @@ func (s *Service) RunTask(input *ecs.RunTaskInput, accountID string) (*ecs.RunTa rec := s.newTaskRecord(accountID, cluster, taskID, taskDef, inst, cpu, mem) rec.NetworkMode = mode + rec.Group = aws.StringValue(input.Group) + rec.StartedBy = aws.StringValue(input.StartedBy) if mode == NetworkModeAwsvpc { if failure := s.provisionTaskENI(kv, accountID, cluster, rec, netCfg); failure != nil { out.Failures = append(out.Failures, failure) @@ -267,6 +269,12 @@ func (s *Service) taskToAWS(accountID string, r *TaskRecord) *ecs.Task { LastStatus: aws.String(r.LastStatus), ContainerInstanceArn: aws.String(r.ContainerInstanceARN), } + if r.Group != "" { + t.Group = aws.String(r.Group) + } + if r.StartedBy != "" { + t.StartedBy = aws.String(r.StartedBy) + } if r.StoppedReason != "" { t.StoppedReason = aws.String(r.StoppedReason) } diff --git a/spinifex/handlers/elbv2/service_impl.go b/spinifex/handlers/elbv2/service_impl.go index daaf87788..4455ad809 100644 --- a/spinifex/handlers/elbv2/service_impl.go +++ b/spinifex/handlers/elbv2/service_impl.go @@ -1996,21 +1996,41 @@ func (s *ELBv2ServiceImpl) DescribeTargetHealth(input *elbv2.DescribeTargetHealt } } + // A target group not forwarded to by any listener serves no traffic; its + // targets report "unused" (AWS Target.NotInUse), not "initial". + inUse, err := s.store.TargetGroupInUse(tg.TargetGroupArn) + if err != nil { + slog.Error("DescribeTargetHealth: failed to check TG association", "arn", tg.TargetGroupArn, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + descriptions := make([]*elbv2.TargetHealthDescription, 0) for _, t := range tg.Targets { if len(targetFilter) > 0 && !targetFilter[t.Id] { continue } + state, healthDesc := t.HealthState, t.HealthDesc + var reason string + if !inUse && t.HealthState != TargetHealthDraining { + state = TargetHealthUnused + healthDesc = "Target group is not configured to receive traffic from a load balancer" + reason = "Target.NotInUse" + } + + health := &elbv2.TargetHealth{ + State: aws.String(state), + Description: aws.String(healthDesc), + } + if reason != "" { + health.Reason = aws.String(reason) + } desc := &elbv2.TargetHealthDescription{ Target: &elbv2.TargetDescription{ Id: aws.String(t.Id), Port: aws.Int64(t.Port), }, - TargetHealth: &elbv2.TargetHealth{ - State: aws.String(t.HealthState), - Description: aws.String(t.HealthDesc), - }, + TargetHealth: health, } descriptions = append(descriptions, desc) } diff --git a/spinifex/handlers/elbv2/service_impl_health_test.go b/spinifex/handlers/elbv2/service_impl_health_test.go index 9a797edee..2dd52f35d 100644 --- a/spinifex/handlers/elbv2/service_impl_health_test.go +++ b/spinifex/handlers/elbv2/service_impl_health_test.go @@ -206,3 +206,60 @@ func TestResetTargetHealthOnStartup_NilSafe(t *testing.T) { svc2 := &ELBv2ServiceImpl{} assert.NoError(t, svc2.ResetTargetHealthOnStartup(t.Context())) } + +// A target group not forwarded to by any listener serves no traffic, so its +// targets report "unused" (Target.NotInUse) instead of stalling in "initial". +func TestDescribeTargetHealth_UnusedWhenNoListener(t *testing.T) { + svc, _, _ := setupTestServiceWithInstance(t, "i-web001", "10.0.1.50") + + tgOut, err := svc.CreateTargetGroup(&elbv2.CreateTargetGroupInput{ + Name: aws.String("detached-tg"), Port: aws.Int64(80), + }, testAccountID) + require.NoError(t, err) + tgArn := tgOut.TargetGroups[0].TargetGroupArn + + _, err = svc.RegisterTargets(&elbv2.RegisterTargetsInput{ + TargetGroupArn: tgArn, + Targets: []*elbv2.TargetDescription{{Id: aws.String("i-web001")}}, + }, testAccountID) + require.NoError(t, err) + + health, err := svc.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID) + require.NoError(t, err) + require.Len(t, health.TargetHealthDescriptions, 1) + th := health.TargetHealthDescriptions[0].TargetHealth + assert.Equal(t, TargetHealthUnused, aws.StringValue(th.State)) + assert.Equal(t, "Target.NotInUse", aws.StringValue(th.Reason)) +} + +// Once a listener forwards to the target group it is in use, so DescribeTargetHealth +// reports the live health state ("initial" until a health report lands), not "unused". +func TestDescribeTargetHealth_InUseWhenListenerForwards(t *testing.T) { + svc, _, _ := setupTestServiceWithInstance(t, "i-web001", "10.0.1.50") + + tgOut, err := svc.CreateTargetGroup(&elbv2.CreateTargetGroupInput{ + Name: aws.String("attached-tg"), Port: aws.Int64(80), + }, testAccountID) + require.NoError(t, err) + tgArn := tgOut.TargetGroups[0].TargetGroupArn + + lbOut, err := svc.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{Name: aws.String("attached-lb")}, testAccountID) + require.NoError(t, err) + _, err = svc.CreateListener(&elbv2.CreateListenerInput{ + LoadBalancerArn: lbOut.LoadBalancers[0].LoadBalancerArn, + Protocol: aws.String("HTTP"), Port: aws.Int64(80), + DefaultActions: []*elbv2.Action{{Type: aws.String("forward"), TargetGroupArn: tgArn}}, + }, testAccountID) + require.NoError(t, err) + + _, err = svc.RegisterTargets(&elbv2.RegisterTargetsInput{ + TargetGroupArn: tgArn, + Targets: []*elbv2.TargetDescription{{Id: aws.String("i-web001")}}, + }, testAccountID) + require.NoError(t, err) + + health, err := svc.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID) + require.NoError(t, err) + require.Len(t, health.TargetHealthDescriptions, 1) + assert.Equal(t, TargetHealthInitial, aws.StringValue(health.TargetHealthDescriptions[0].TargetHealth.State)) +} diff --git a/spinifex/handlers/elbv2/service_impl_test.go b/spinifex/handlers/elbv2/service_impl_test.go index a5efa3b32..fb183c28a 100644 --- a/spinifex/handlers/elbv2/service_impl_test.go +++ b/spinifex/handlers/elbv2/service_impl_test.go @@ -764,10 +764,11 @@ func TestRegisterTargets(t *testing.T) { require.NoError(t, err) require.Len(t, health.TargetHealthDescriptions, 2) - // First target should use TG default port + // First target should use TG default port. With no listener forwarding to the + // TG it serves no traffic, so health is "unused" (Target.NotInUse). assert.Equal(t, "i-aaa111", *health.TargetHealthDescriptions[0].Target.Id) assert.Equal(t, int64(80), *health.TargetHealthDescriptions[0].Target.Port) - assert.Equal(t, "initial", *health.TargetHealthDescriptions[0].TargetHealth.State) + assert.Equal(t, "unused", *health.TargetHealthDescriptions[0].TargetHealth.State) // Second target should use override port assert.Equal(t, "i-bbb222", *health.TargetHealthDescriptions[1].Target.Id) diff --git a/spinifex/handlers/elbv2/service_nats_e2e_test.go b/spinifex/handlers/elbv2/service_nats_e2e_test.go index 604f1ae3c..fb9bdc0d1 100644 --- a/spinifex/handlers/elbv2/service_nats_e2e_test.go +++ b/spinifex/handlers/elbv2/service_nats_e2e_test.go @@ -134,7 +134,8 @@ func TestNATSE2E_FullWorkflow(t *testing.T) { }, testAccountID) require.NoError(t, err) require.Len(t, healthOut.TargetHealthDescriptions, 2) - assert.Equal(t, "initial", *healthOut.TargetHealthDescriptions[0].TargetHealth.State) + // No listener forwards here yet, so the targets are "unused" (Target.NotInUse). + assert.Equal(t, "unused", *healthOut.TargetHealthDescriptions[0].TargetHealth.State) // 4. Create load balancer lbOut, err := client.CreateLoadBalancer(&elbv2.CreateLoadBalancerInput{ @@ -158,6 +159,11 @@ func TestNATSE2E_FullWorkflow(t *testing.T) { require.Len(t, lstOut.Listeners, 1) assert.Equal(t, int64(80), *lstOut.Listeners[0].Port) + // Now a listener forwards to the TG, so its targets are in use again ("initial"). + healthOut, err = client.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: tgArn}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "initial", *healthOut.TargetHealthDescriptions[0].TargetHealth.State) + // 6. Describe listeners lstDesc, err := client.DescribeListeners(&elbv2.DescribeListenersInput{ LoadBalancerArn: lbArn, diff --git a/spinifex/handlers/elbv2/store.go b/spinifex/handlers/elbv2/store.go index 7edec5006..7277fd39e 100644 --- a/spinifex/handlers/elbv2/store.go +++ b/spinifex/handlers/elbv2/store.go @@ -315,6 +315,39 @@ func (s *Store) TargetGroupsForLB(lbID string) ([]*TargetGroupRecord, error) { return tgs, nil } +// TargetGroupInUse reports whether any listener default action or rule action +// forwards to tgArn. A target group not referenced by a load balancer serves no +// traffic, so its targets are reported "unused" (AWS Target.NotInUse) instead of +// staying stuck in "initial" with no health checker to advance them. +func (s *Store) TargetGroupInUse(tgArn string) (bool, error) { + if tgArn == "" { + return false, nil + } + listeners, err := s.ListListeners() + if err != nil { + return false, err + } + for _, l := range listeners { + for _, a := range l.DefaultActions { + if a.TargetGroupArn == tgArn { + return true, nil + } + } + } + rules, err := s.ListRules() + if err != nil { + return false, err + } + for _, r := range rules { + for _, a := range r.Actions { + if a.TargetGroupArn == tgArn { + return true, nil + } + } + } + return false, nil +} + // GetTargetGroupByName finds a target group by name within a VPC. func (s *Store) GetTargetGroupByName(name, vpcID string) (*TargetGroupRecord, error) { tgs, err := s.ListTargetGroups()