fix: honour the status filter on ListContainerInstances (#1087)

The filter was never read, so ACTIVE, DRAINING and no filter at all returned every registration in the cluster. A caller could not ask which container instances can take work without describing every one of them, and a cluster carrying a drained node read as if every node were live.

The status a record holds is the one DescribeContainerInstances reports, so a filtered list and a describe of what it returns agree. REGISTERING, DEREGISTERING and REGISTRATION_FAILED are accepted and match nothing, because no record here ever holds them; a status outside the enum is refused.

spinifex/handlers/ecs/instance_list_status_test.go

@@ -0,0 +1,103 @@
+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"
+)
+
+// listInstanceARNs lists a cluster's container instances at one status, or at
+// every status when the filter is empty.
+func listInstanceARNs(t *testing.T, svc *Service, cluster, status string) []string {
+	t.Helper()
+	in := &ecs.ListContainerInstancesInput{Cluster: aws.String(cluster)}
+	if status != "" {
+		in.Status = aws.String(status)
+	}
+	out, err := svc.ListContainerInstances(t.Context(), in, testAccountID)
+	require.NoError(t, err)
+	return aws.StringValueSlice(out.ContainerInstanceArns)
+}
+
+// A caller that cannot narrow the list has to describe every registration to
+// find one that can take work, and a cluster carrying a drained node reads as
+// if every node were live.
+func TestListContainerInstances_StatusSelectsOnlyThatStatus(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(t.Context(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerInstance(t, svc, "web", "i-live", 1024, 2048)
+	registerInstance(t, svc, "web", "i-drained", 1024, 2048)
+
+	kv, err := svc.bucket(t.Context(), testAccountID)
+	require.NoError(t, err)
+	drained, err := svc.upsertInstance(t.Context(), kv, testAccountID, "web", "i-drained", func(r *InstanceRecord) {
+		r.Status = InstanceStatusDraining
+	})
+	require.NoError(t, err)
+
+	active := listInstanceARNs(t, svc, "web", InstanceStatusActive)
+	assert.Len(t, active, 1)
+	assert.NotContains(t, active, drained.ARN)
+
+	assert.Equal(t, []string{drained.ARN}, listInstanceARNs(t, svc, "web", InstanceStatusDraining))
+	assert.Len(t, listInstanceARNs(t, svc, "web", ""), 2)
+}
+
+// The statuses a registration passes through in AWS never reach a record here,
+// so asking for one is a valid question with an empty answer rather than an
+// error.
+func TestListContainerInstances_TransitionalStatusMatchesNothing(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(t.Context(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerInstance(t, svc, "web", "i-live", 1024, 2048)
+
+	for _, status := range []string{"REGISTERING", "DEREGISTERING", "REGISTRATION_FAILED"} {
+		assert.Empty(t, listInstanceARNs(t, svc, "web", status), status)
+	}
+}
+
+func TestListContainerInstances_UnknownStatusIsRefused(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(t.Context(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+
+	_, err = svc.ListContainerInstances(t.Context(), &ecs.ListContainerInstancesInput{
+		Cluster: aws.String("web"),
+		Status:  aws.String("INACTIVE"),
+	}, testAccountID)
+	require.Error(t, err)
+}
+
+// A filtered list and a describe of what it returned have to agree, since the
+// status the filter selects on is the one describe reports.
+func TestListContainerInstances_FilterAgreesWithDescribe(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(t.Context(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerInstance(t, svc, "web", "i-live", 1024, 2048)
+	registerInstance(t, svc, "web", "i-drained", 1024, 2048)
+
+	kv, err := svc.bucket(t.Context(), testAccountID)
+	require.NoError(t, err)
+	_, err = svc.upsertInstance(t.Context(), kv, testAccountID, "web", "i-drained", func(r *InstanceRecord) {
+		r.Status = InstanceStatusDraining
+	})
+	require.NoError(t, err)
+
+	arns := listInstanceARNs(t, svc, "web", InstanceStatusActive)
+	require.NotEmpty(t, arns)
+	out, err := svc.DescribeContainerInstances(t.Context(), &ecs.DescribeContainerInstancesInput{
+		Cluster:            aws.String("web"),
+		ContainerInstances: aws.StringSlice(arns),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, out.ContainerInstances, len(arns))
+	for _, ci := range out.ContainerInstances {
+		assert.Equal(t, InstanceStatusActive, aws.StringValue(ci.Status))
+	}
+}

spinifex/handlers/ecs/instances.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -119,9 +120,25 @@ func (s *Service) DescribeContainerInstances(ctx context.Context, input *ecs.Des
 	return out, nil
 }
 
-// ListContainerInstances returns the ARNs of all container instances in a cluster.
+// instanceListStatuses are the values the status filter accepts. A registration
+// here is only ever ACTIVE or DRAINING; the other three are states AWS passes a
+// registration through, so they are valid to ask for and match nothing.
+var instanceListStatuses = map[string]bool{
+	InstanceStatusActive:   true,
+	InstanceStatusDraining: true,
+	"REGISTERING":          true,
+	"DEREGISTERING":        true,
+	"REGISTRATION_FAILED":  true,
+}
+
+// ListContainerInstances returns the ARNs of the container instances in a
+// cluster, narrowed to one registration status when the caller names one.
 func (s *Service) ListContainerInstances(ctx context.Context, input *ecs.ListContainerInstancesInput, accountID string) (*ecs.ListContainerInstancesOutput, error) {
 	cluster := ClusterShortName(aws.StringValue(input.Cluster))
+	status := aws.StringValue(input.Status)
+	if status != "" && !instanceListStatuses[status] {
+		return nil, errors.New(awserrors.ErrorECSInvalidParameter)
+	}
 	kv, err := s.bucket(ctx, accountID)
 	if err != nil {
 		return nil, err
@@ -132,6 +149,11 @@ func (s *Service) ListContainerInstances(ctx context.Context, input *ecs.ListCon
 	}
 	arns := make([]string, 0, len(recs))
 	for i := range recs {
+		// The record's status is what DescribeContainerInstances reports, so a
+		// filtered list and a describe of what it returns cannot disagree.
+		if status != "" && recs[i].Status != status {
+			continue
+		}
 		arns = append(arns, recs[i].ARN)
 	}
 	return &ecs.ListContainerInstancesOutput{ContainerInstanceArns: aws.StringSlice(arns)}, nil