From dd7a3e02a0ae7bf91fb0c5285c572c4079522536 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 16 Sep 2026 08:53:37 +1000 Subject: [PATCH] fix(acm,elbv2): serialise target group writes and fail fast on a missing tenant CA (#1041) RegisterTargets read the target group, appended in memory and wrote the whole record back with no revision guard, so two concurrent registrations lost one. Terraform parallelises by default, so any target group with more than one member hit it, and the apply exited 0 with half the fleet not behind the load balancer. Every writer of the record now goes through the store's existing CAS rather than a wholesale Put, because a CAS loop on one writer is defeated by a non-CAS writer on the same key. Contention stays per target group; no global lock is introduced. RequestCertificate returned InternalErrorException for a missing tenant CA, which is a client-actionable precondition rather than a server fault. Worse, that code is classified retryable, so the provider backed off and retried instead of failing, and a single certificate sat at Still creating for over nine minutes. It now returns ACM's not-found code, which is terminal, and carries a message naming the missing CA and how to create one. ListCertificates omitted CertificateSummaryList entirely on an empty account, since the SDK marshaller drops a nil slice unconditionally. Closes: mulga-c6khe Closes: mulga-wbjal Closes: mulga-l6yft Closes: mulga-61h3h --- spinifex/handlers/acm/service_impl.go | 17 +- spinifex/handlers/acm/service_impl_test.go | 64 +++++ spinifex/handlers/elbv2/health_checker.go | 24 +- spinifex/handlers/elbv2/service_impl.go | 265 +++++++++++------- .../handlers/elbv2/service_impl_attributes.go | 12 +- spinifex/handlers/elbv2/service_impl_tags.go | 9 +- spinifex/handlers/elbv2/service_impl_test.go | 172 ++++++++++++ spinifex/handlers/elbv2/store.go | 45 ++- 8 files changed, 497 insertions(+), 111 deletions(-) diff --git a/spinifex/handlers/acm/service_impl.go b/spinifex/handlers/acm/service_impl.go index 2d02f8ad5..a5226f67b 100644 --- a/spinifex/handlers/acm/service_impl.go +++ b/spinifex/handlers/acm/service_impl.go @@ -298,7 +298,12 @@ func (s *ACMServiceImpl) RequestCertificate(ctx context.Context, input *acm.Requ if mode == ValidationModePrivateCA { if s.TenantCA == nil { slog.ErrorContext(ctx, "RequestCertificate: PRIVATE_CA mode derived but no tenant CA is wired", "domain", domain) - return nil, errors.New(awserrors.ErrorInternalError) + // An absent tenant CA is an unmet precondition the caller can act + // on, not a server fault: ResourceNotFound is a 404 the provider + // fails fast on, where InternalError is retried with backoff. + return nil, awserrors.Errorf(awserrors.ErrorResourceNotFound, + "no tenant CA is configured, so PRIVATE_CA issuance for %q is unavailable; create one with %q", + domain, tenantCACreateCommandHint) } for _, d := range allDomains { if !s.TenantCA.Authorized(d) { @@ -499,14 +504,18 @@ func (s *ACMServiceImpl) GetCertificate(ctx context.Context, input *acm.GetCerti // ListCertificates returns summaries for every cert owned by accountID. // Summaries carry no key material, so this takes the metadata-only accessor. +// +// The list is built with make, never appended into a nil field: the SDK's JSON +// marshaller drops a nil slice whatever the struct tag says, so an account with +// no certificates would answer {} and hand an SDK caller a nil to range over. func (s *ACMServiceImpl) ListCertificates(ctx context.Context, input *acm.ListCertificatesInput, accountID string) (*acm.ListCertificatesOutput, error) { recs, err := s.store.ListCertMetadata(ctx, accountID) if err != nil { return nil, errors.New(awserrors.ErrorInternalError) } - out := &acm.ListCertificatesOutput{} + summaries := make([]*acm.CertificateSummary, 0, len(recs)) for _, rec := range recs { - out.CertificateSummaryList = append(out.CertificateSummaryList, &acm.CertificateSummary{ + summaries = append(summaries, &acm.CertificateSummary{ CertificateArn: aws.String(rec.CertificateArn), DomainName: aws.String(rec.DomainName), Status: aws.String(certStatusOrDefault(rec)), @@ -518,7 +527,7 @@ func (s *ACMServiceImpl) ListCertificates(ctx context.Context, input *acm.ListCe InUse: aws.Bool(len(rec.InUseBy) > 0), }) } - return out, nil + return &acm.ListCertificatesOutput{CertificateSummaryList: summaries}, nil } // DeleteCertificate removes an owned cert; unknown ARN → ResourceNotFound. diff --git a/spinifex/handlers/acm/service_impl_test.go b/spinifex/handlers/acm/service_impl_test.go index 5c7df5b0d..2490f1c36 100644 --- a/spinifex/handlers/acm/service_impl_test.go +++ b/spinifex/handlers/acm/service_impl_test.go @@ -18,6 +18,7 @@ import ( "time" "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/private/protocol/json/jsonutil" "github.com/aws/aws-sdk-go/service/acm" "github.com/mulgadc/spinifex/spinifex/arn" "github.com/mulgadc/spinifex/spinifex/awserrors" @@ -716,3 +717,66 @@ func TestMintedArnRoundTripsThroughTheGatesFormatter(t *testing.T) { require.True(t, ok) assert.Equal(t, minted, arn.FormatACMCertificate(svc.region, testAccountID, id)) } + +// An account with no certificates must still answer with the key present and +// empty: the SDK marshaller drops a nil slice, so a caller that ranges over the +// result without a nil check breaks on every account's first state. +func TestListCertificates_EmptyAccountReturnsEmptyList(t *testing.T) { + svc := setupACMService(t) + + list, err := svc.ListCertificates(context.Background(), &acm.ListCertificatesInput{}, testAccountID) + require.NoError(t, err) + require.NotNil(t, list.CertificateSummaryList, "CertificateSummaryList must be present, not nil") + assert.Empty(t, list.CertificateSummaryList) + + body, err := jsonutil.BuildJSON(list) + require.NoError(t, err) + assert.Contains(t, string(body), "CertificateSummaryList", "the wire body must carry the key, not {}") +} + +// The populated case must keep working: an empty slice that never fills is just +// a different way of losing the list. +func TestListCertificates_PopulatedAccountReturnsSummaries(t *testing.T) { + svc := setupACMService(t) + c1, k1 := genCert(t, "filled-one.example.com") + c2, k2 := genCert(t, "filled-two.example.com") + + _, err := svc.ImportCertificate(context.Background(), &acm.ImportCertificateInput{Certificate: c1, PrivateKey: k1}, testAccountID) + require.NoError(t, err) + _, err = svc.ImportCertificate(context.Background(), &acm.ImportCertificateInput{Certificate: c2, PrivateKey: k2}, testAccountID) + require.NoError(t, err) + + list, err := svc.ListCertificates(context.Background(), &acm.ListCertificatesInput{}, testAccountID) + require.NoError(t, err) + require.Len(t, list.CertificateSummaryList, 2) + domains := []string{ + aws.StringValue(list.CertificateSummaryList[0].DomainName), + aws.StringValue(list.CertificateSummaryList[1].DomainName), + } + assert.ElementsMatch(t, []string{"filled-one.example.com", "filled-two.example.com"}, domains) +} + +// A missing tenant CA is a precondition the operator can fix, not a server +// fault. InternalErrorException is a retryable 500, so Terraform sat on +// "Still creating..." instead of failing; this asserts the 4xx and the reason. +func TestRequestCertificate_NoTenantCAIsAClientErrorNamingTheCause(t *testing.T) { + svc := setupACMService(t) + require.Nil(t, svc.TenantCA, "this test only means anything with no tenant CA wired") + + _, err := svc.RequestCertificate(context.Background(), &acm.RequestCertificateInput{ + DomainName: aws.String("no-ca.example.com"), + }, testAccountID) + require.Error(t, err) + + code, message, ok := awserrors.ResolveErrorDetail(err) + require.True(t, ok, "the error must carry a registered AWS code") + assert.Equal(t, awserrors.ErrorResourceNotFound, code) + assert.NotEqual(t, awserrors.ErrorInternalError, code, "a retryable 500 hangs the provider's create") + + assert.Equal(t, 404, awserrors.ErrorLookup[code].HTTPCode) + assert.True(t, awserrors.IsTerminal(err), "the provider must fail fast, not retry with backoff") + + assert.Contains(t, message, "tenant CA", "the message must name the missing tenant CA") + assert.Contains(t, message, "no-ca.example.com") + assert.Contains(t, message, tenantCACreateCommandHint, "the message must name the command that fixes it") +} diff --git a/spinifex/handlers/elbv2/health_checker.go b/spinifex/handlers/elbv2/health_checker.go index fbc683b43..75394590c 100644 --- a/spinifex/handlers/elbv2/health_checker.go +++ b/spinifex/handlers/elbv2/health_checker.go @@ -2,6 +2,7 @@ package handlers_elbv2 import ( "context" + "errors" "fmt" "log/slog" "sync" @@ -114,8 +115,29 @@ func (hc *healthChecker) handleHealthReportDirect(ctx context.Context, report lb } hc.mu.Unlock() + // Apply the new states field-by-field under CAS. A whole-record Put here + // would discard any target registered since this scan read the group, which + // is the one write on this key that runs on its own schedule. for _, tg := range changedTGs { - if err := hc.store.PutTargetGroup(ctx, tg); err != nil { + states := make(map[string]Target, len(tg.Targets)) + for _, t := range tg.Targets { + states[targetKey(t.Id, t.Port)] = t + } + _, err := hc.store.UpdateTargetGroup(ctx, tg.TargetGroupID, func(rec *TargetGroupRecord) (bool, error) { + changed := false + for i := range rec.Targets { + cur := &rec.Targets[i] + want, ok := states[targetKey(cur.Id, cur.Port)] + if !ok || cur.HealthState == want.HealthState { + continue + } + cur.HealthState = want.HealthState + cur.HealthDesc = want.HealthDesc + changed = true + } + return changed, nil + }) + if err != nil && !errors.Is(err, ErrTargetGroupNotFound) { slog.ErrorContext(ctx, "healthChecker: failed to persist target group", "tgId", tg.TargetGroupID, "err", err) } } diff --git a/spinifex/handlers/elbv2/service_impl.go b/spinifex/handlers/elbv2/service_impl.go index d4d5b5966..caa8e1b9b 100644 --- a/spinifex/handlers/elbv2/service_impl.go +++ b/spinifex/handlers/elbv2/service_impl.go @@ -220,27 +220,35 @@ func (s *ELBv2ServiceImpl) ResetTargetHealthOnStartup(ctx context.Context) error if ctx.Err() != nil { return ctx.Err() } - changed := false - for i := range tg.Targets { - t := &tg.Targets[i] - // Draining state is driven by DeregisterTargets; don't clobber it. - if t.HealthState == TargetHealthDraining { - continue + // Counted inside the CAS body and only added on commit, so a retried + // attempt does not count the same target twice. + reset := 0 + updated, err := s.store.UpdateTargetGroup(ctx, tg.TargetGroupID, func(rec *TargetGroupRecord) (bool, error) { + reset = 0 + for i := range rec.Targets { + t := &rec.Targets[i] + // Draining state is driven by DeregisterTargets; don't clobber it. + if t.HealthState == TargetHealthDraining { + continue + } + if t.HealthState == TargetHealthInitial { + continue + } + t.HealthState = TargetHealthInitial + t.HealthDesc = "Target registration is in progress" + reset++ } - if t.HealthState == TargetHealthInitial { - continue - } - t.HealthState = TargetHealthInitial - t.HealthDesc = "Target registration is in progress" - changed = true - resetTargets++ - } - if changed { - if err := s.store.PutTargetGroup(ctx, tg); err != nil { + return reset > 0, nil + }) + if err != nil { + if !errors.Is(err, ErrTargetGroupNotFound) { slog.Error("ResetTargetHealthOnStartup: persist failed", "tgId", tg.TargetGroupID, "err", err) - continue } + continue + } + if updated != nil && reset > 0 { + resetTargets += reset resetTGs++ } } @@ -2011,54 +2019,51 @@ func (s *ELBv2ServiceImpl) ModifyTargetGroup(ctx context.Context, input *elbv2.M return nil, errors.New(awserrors.ErrorMissingParameter) } - tg, err := s.store.GetTargetGroupByArn(ctx, *input.TargetGroupArn) + // Under CAS, so a health-check change and a concurrent target registration + // on the same group cannot discard each other. + tg, err := s.updateTargetGroupCAS(ctx, *input.TargetGroupArn, accountID, "ModifyTargetGroup", func(rec *TargetGroupRecord) (bool, error) { + hc := rec.HealthCheck + if input.HealthCheckEnabled != nil { + hc.Enabled = *input.HealthCheckEnabled + } + if input.HealthCheckProtocol != nil { + hc.Protocol = *input.HealthCheckProtocol + } + if input.HealthCheckPort != nil { + hc.Port = *input.HealthCheckPort + } + if input.HealthCheckPath != nil { + if err := validateHealthCheckPath(*input.HealthCheckPath); err != nil { + return false, err + } + hc.Path = *input.HealthCheckPath + } + if input.HealthCheckIntervalSeconds != nil { + hc.IntervalSeconds = *input.HealthCheckIntervalSeconds + } + if input.HealthCheckTimeoutSeconds != nil { + hc.TimeoutSeconds = *input.HealthCheckTimeoutSeconds + } + if input.HealthyThresholdCount != nil { + hc.HealthyThreshold = *input.HealthyThresholdCount + } + if input.UnhealthyThresholdCount != nil { + hc.UnhealthyThreshold = *input.UnhealthyThresholdCount + } + if input.Matcher != nil && input.Matcher.HttpCode != nil { + if err := validateHealthCheckMatcher(*input.Matcher.HttpCode); err != nil { + return false, err + } + hc.Matcher = *input.Matcher.HttpCode + } + if hc == rec.HealthCheck { + return false, nil + } + rec.HealthCheck = hc + return true, nil + }) if err != nil { - slog.ErrorContext(ctx, "ModifyTargetGroup: failed to get TG", "arn", *input.TargetGroupArn, "err", err) - return nil, errors.New(awserrors.ErrorServerInternal) - } - if tg == nil || tg.AccountID != accountID { - return nil, errors.New(awserrors.ErrorELBv2TargetGroupNotFound) - } - - hc := tg.HealthCheck - if input.HealthCheckEnabled != nil { - hc.Enabled = *input.HealthCheckEnabled - } - if input.HealthCheckProtocol != nil { - hc.Protocol = *input.HealthCheckProtocol - } - if input.HealthCheckPort != nil { - hc.Port = *input.HealthCheckPort - } - if input.HealthCheckPath != nil { - if err := validateHealthCheckPath(*input.HealthCheckPath); err != nil { - return nil, err - } - hc.Path = *input.HealthCheckPath - } - if input.HealthCheckIntervalSeconds != nil { - hc.IntervalSeconds = *input.HealthCheckIntervalSeconds - } - if input.HealthCheckTimeoutSeconds != nil { - hc.TimeoutSeconds = *input.HealthCheckTimeoutSeconds - } - if input.HealthyThresholdCount != nil { - hc.HealthyThreshold = *input.HealthyThresholdCount - } - if input.UnhealthyThresholdCount != nil { - hc.UnhealthyThreshold = *input.UnhealthyThresholdCount - } - if input.Matcher != nil && input.Matcher.HttpCode != nil { - if err := validateHealthCheckMatcher(*input.Matcher.HttpCode); err != nil { - return nil, err - } - hc.Matcher = *input.Matcher.HttpCode - } - tg.HealthCheck = hc - - if err := s.store.PutTargetGroup(ctx, tg); err != nil { - slog.ErrorContext(ctx, "ModifyTargetGroup: failed to persist record", "arn", tg.TargetGroupArn, "err", err) - return nil, errors.New(awserrors.ErrorServerInternal) + return nil, err } slog.InfoContext(ctx, "ModifyTargetGroup completed", "arn", tg.TargetGroupArn, "accountID", accountID) @@ -2197,27 +2202,61 @@ func (s *ELBv2ServiceImpl) DescribeTargetGroups(ctx context.Context, input *elbv // --- Target registration --- +// targetKey identifies a registered target within its group. Port is always the +// resolved one (the group default when the caller sent none), so the same target +// never appears under two keys. +func targetKey(id string, port int64) string { + return fmt.Sprintf("%s:%d", id, port) +} + +// updateTargetGroupCAS mutates the target group at arn under optimistic +// concurrency, mapping an absent or cross-account record onto the ELBv2 +// not-found error and anything unclassified onto an internal error. +func (s *ELBv2ServiceImpl) updateTargetGroupCAS(ctx context.Context, arn, accountID, opName string, mutate func(*TargetGroupRecord) (bool, error)) (*TargetGroupRecord, error) { + tg, err := s.store.UpdateTargetGroupByArn(ctx, arn, func(rec *TargetGroupRecord) (bool, error) { + if rec.AccountID != accountID { + return false, ErrTargetGroupNotFound + } + return mutate(rec) + }) + switch { + case err == nil: + return tg, nil + case errors.Is(err, ErrTargetGroupNotFound): + return nil, errors.New(awserrors.ErrorELBv2TargetGroupNotFound) + } + // An error the mutation raised already carries its own AWS code; only an + // unclassified store or contention failure becomes an internal error. + if _, ok := awserrors.ResolveErrorCode(err); ok { + return nil, err + } + slog.ErrorContext(ctx, opName+": failed to update TG", "arn", arn, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) +} + func (s *ELBv2ServiceImpl) RegisterTargets(ctx context.Context, input *elbv2.RegisterTargetsInput, accountID string) (*elbv2.RegisterTargetsOutput, error) { if input.TargetGroupArn == nil || *input.TargetGroupArn == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } + tgArn := *input.TargetGroupArn - tg, err := s.store.GetTargetGroupByArn(ctx, *input.TargetGroupArn) + tg, err := s.store.GetTargetGroupByArn(ctx, tgArn) if err != nil { - slog.ErrorContext(ctx, "RegisterTargets: failed to get TG", "arn", *input.TargetGroupArn, "err", err) + slog.ErrorContext(ctx, "RegisterTargets: failed to get TG", "arn", tgArn, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } if tg == nil || tg.AccountID != accountID { return nil, errors.New(awserrors.ErrorELBv2TargetGroupNotFound) } - // Build map of existing targets for dedup - existing := make(map[string]int) // id:port -> index - for i, t := range tg.Targets { - key := fmt.Sprintf("%s:%d", t.Id, t.Port) - existing[key] = i + // Resolve private IPs before the CAS loop: an ENI lookup is network I/O and + // must not repeat on every retry. An already-registered target is skipped + // without one, so re-registering never depends on the instance still existing. + registered := make(map[string]bool, len(tg.Targets)) + for _, t := range tg.Targets { + registered[targetKey(t.Id, t.Port)] = true } - + pending := make([]Target, 0, len(input.Targets)) for _, td := range input.Targets { if td.Id == nil { continue @@ -2226,18 +2265,14 @@ func (s *ELBv2ServiceImpl) RegisterTargets(ctx context.Context, input *elbv2.Reg if td.Port != nil { port = *td.Port } - key := fmt.Sprintf("%s:%d", *td.Id, port) - if _, exists := existing[key]; exists { + if registered[targetKey(*td.Id, port)] { continue // Already registered } - - // Resolve target ID → private IP (instance ENI lookup, or raw IP for ip-type TGs) privateIP, err := s.resolveRegisteredTargetIP(ctx, tg.TargetType, *td.Id, accountID) if err != nil { return nil, err } - - tg.Targets = append(tg.Targets, Target{ + pending = append(pending, Target{ Id: *td.Id, Port: port, HealthState: TargetHealthInitial, @@ -2246,18 +2281,35 @@ func (s *ELBv2ServiceImpl) RegisterTargets(ctx context.Context, input *elbv2.Reg }) } - if err := s.store.PutTargetGroup(ctx, tg); err != nil { - slog.ErrorContext(ctx, "RegisterTargets: failed to persist TG", "arn", *input.TargetGroupArn, "err", err) - return nil, errors.New(awserrors.ErrorServerInternal) + // The append runs against the record as committed, not the copy read above, + // so two concurrent registrations on one group both land. + if _, err := s.updateTargetGroupCAS(ctx, tgArn, accountID, "RegisterTargets", func(rec *TargetGroupRecord) (bool, error) { + present := make(map[string]bool, len(rec.Targets)) + for _, t := range rec.Targets { + present[targetKey(t.Id, t.Port)] = true + } + changed := false + for _, t := range pending { + key := targetKey(t.Id, t.Port) + if present[key] { + continue + } + present[key] = true + rec.Targets = append(rec.Targets, t) + changed = true + } + return changed, nil + }); err != nil { + return nil, err } // Reload HAProxy for any LBs that reference this target group - if err := s.updateStoredConfigForTargetGroup(ctx, tg.TargetGroupArn); err != nil { - slog.ErrorContext(ctx, "RegisterTargets: failed to update config", "arn", *input.TargetGroupArn, "err", err) + if err := s.updateStoredConfigForTargetGroup(ctx, tgArn); err != nil { + slog.ErrorContext(ctx, "RegisterTargets: failed to update config", "arn", tgArn, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } - slog.InfoContext(ctx, "RegisterTargets completed", "tgArn", *input.TargetGroupArn, "targetsAdded", len(input.Targets), "accountID", accountID) + slog.InfoContext(ctx, "RegisterTargets completed", "tgArn", tgArn, "targetsAdded", len(input.Targets), "accountID", accountID) return &elbv2.RegisterTargetsOutput{}, nil } @@ -2266,10 +2318,11 @@ func (s *ELBv2ServiceImpl) DeregisterTargets(ctx context.Context, input *elbv2.D if input.TargetGroupArn == nil || *input.TargetGroupArn == "" { return nil, errors.New(awserrors.ErrorMissingParameter) } + tgArn := *input.TargetGroupArn - tg, err := s.store.GetTargetGroupByArn(ctx, *input.TargetGroupArn) + tg, err := s.store.GetTargetGroupByArn(ctx, tgArn) if err != nil { - slog.ErrorContext(ctx, "DeregisterTargets: failed to get TG", "arn", *input.TargetGroupArn, "err", err) + slog.ErrorContext(ctx, "DeregisterTargets: failed to get TG", "arn", tgArn, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } if tg == nil || tg.AccountID != accountID { @@ -2286,32 +2339,42 @@ func (s *ELBv2ServiceImpl) DeregisterTargets(ctx context.Context, input *elbv2.D if td.Port != nil { port = *td.Port } - removeSet[fmt.Sprintf("%s:%d", *td.Id, port)] = true + removeSet[targetKey(*td.Id, port)] = true } - var remaining []Target - for _, t := range tg.Targets { - key := fmt.Sprintf("%s:%d", t.Id, t.Port) - if removeSet[key] { - s.hc.removeTarget(tg.TargetGroupID, t.Id, t.Port) - } else { + // removed is rebuilt on every attempt, so a lost CAS race cannot leave a + // target from an abandoned attempt in the health-checker eviction list. + var removed []Target + updated, err := s.updateTargetGroupCAS(ctx, tgArn, accountID, "DeregisterTargets", func(rec *TargetGroupRecord) (bool, error) { + removed = nil + remaining := make([]Target, 0, len(rec.Targets)) + for _, t := range rec.Targets { + if removeSet[targetKey(t.Id, t.Port)] { + removed = append(removed, t) + continue + } remaining = append(remaining, t) } + if len(removed) == 0 { + return false, nil + } + rec.Targets = remaining + return true, nil + }) + if err != nil { + return nil, err } - tg.Targets = remaining - - if err := s.store.PutTargetGroup(ctx, tg); err != nil { - slog.ErrorContext(ctx, "DeregisterTargets: failed to persist TG", "arn", *input.TargetGroupArn, "err", err) - return nil, errors.New(awserrors.ErrorServerInternal) + for _, t := range removed { + s.hc.removeTarget(updated.TargetGroupID, t.Id, t.Port) } // Reload HAProxy for any LBs that reference this target group - if err := s.updateStoredConfigForTargetGroup(ctx, tg.TargetGroupArn); err != nil { - slog.ErrorContext(ctx, "DeregisterTargets: failed to update config", "arn", *input.TargetGroupArn, "err", err) + if err := s.updateStoredConfigForTargetGroup(ctx, tgArn); err != nil { + slog.ErrorContext(ctx, "DeregisterTargets: failed to update config", "arn", tgArn, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } - slog.InfoContext(ctx, "DeregisterTargets completed", "tgArn", *input.TargetGroupArn, "targetsRemoved", len(input.Targets), "accountID", accountID) + slog.InfoContext(ctx, "DeregisterTargets completed", "tgArn", tgArn, "targetsRemoved", len(input.Targets), "accountID", accountID) return &elbv2.DeregisterTargetsOutput{}, nil } diff --git a/spinifex/handlers/elbv2/service_impl_attributes.go b/spinifex/handlers/elbv2/service_impl_attributes.go index f18e13bab..9c925b97c 100644 --- a/spinifex/handlers/elbv2/service_impl_attributes.go +++ b/spinifex/handlers/elbv2/service_impl_attributes.go @@ -166,8 +166,16 @@ func (s *ELBv2ServiceImpl) tgAttrResource(ctx context.Context, arn, accountID, o notFound: awserrors.ErrorELBv2TargetGroupNotFound, // fetch/save close over ctx: attrResource is consumed by the generic // attribute helpers, which have no context of their own to pass down. - fetch: func(arn string) (*TargetGroupRecord, error) { return s.store.GetTargetGroupByArn(ctx, arn) }, - save: func(tg *TargetGroupRecord) error { return s.store.PutTargetGroup(ctx, tg) }, + fetch: func(arn string) (*TargetGroupRecord, error) { return s.store.GetTargetGroupByArn(ctx, arn) }, + // CAS the attribute field alone: a whole-record Put here would discard + // targets registered since fetch read this copy. + save: func(tg *TargetGroupRecord) error { + _, err := s.store.UpdateTargetGroupByArn(ctx, tg.TargetGroupArn, func(rec *TargetGroupRecord) (bool, error) { + rec.Attributes = tg.Attributes + return true, nil + }) + return err + }, accountIDOf: func(r *TargetGroupRecord) string { return r.AccountID }, defaults: func(*TargetGroupRecord) map[string]string { return DefaultTargetGroupAttributes() }, attrsOf: func(r *TargetGroupRecord) map[string]string { return r.Attributes }, diff --git a/spinifex/handlers/elbv2/service_impl_tags.go b/spinifex/handlers/elbv2/service_impl_tags.go index 5ee89d6ba..95cd119d0 100644 --- a/spinifex/handlers/elbv2/service_impl_tags.go +++ b/spinifex/handlers/elbv2/service_impl_tags.go @@ -52,9 +52,14 @@ func (s *ELBv2ServiceImpl) resolveTaggable(ctx context.Context, arn string) (h t } if tg != nil { found = true + // CAS the tag field alone: a whole-record Put here would discard + // targets registered since this copy was read. h = taggable{tags: tg.Tags, owner: tg.AccountID, save: func(t map[string]string) error { - tg.Tags = t - return s.store.PutTargetGroup(ctx, tg) + _, uerr := s.store.UpdateTargetGroupByArn(ctx, arn, func(rec *TargetGroupRecord) (bool, error) { + rec.Tags = t + return true, nil + }) + return uerr }} } case resourcearn.ELBv2Listener: diff --git a/spinifex/handlers/elbv2/service_impl_test.go b/spinifex/handlers/elbv2/service_impl_test.go index 0c13e01d0..2d2c32c6e 100644 --- a/spinifex/handlers/elbv2/service_impl_test.go +++ b/spinifex/handlers/elbv2/service_impl_test.go @@ -3,6 +3,7 @@ package handlers_elbv2 import ( "context" "encoding/xml" + "fmt" "sort" "strings" "sync" @@ -3647,3 +3648,174 @@ func TestDNSWatchBucket_WithStore(t *testing.T) { require.NotNil(t, bucket) assert.Equal(t, KVBucketELBv2, bucket.Name()) } + +// Terraform creates aws_lb_target_group_attachment resources in parallel, so a +// read-modify-write on the group's target list drops registrations — and the +// apply still exits 0 with part of the fleet not behind the load balancer. +func TestRegisterTargets_ConcurrentRegistrationsAllLand(t *testing.T) { + t.Parallel() + svc := setupTestService(t) + + const targetsPerTrial = 8 + // A single trial passes often enough to prove nothing, so each trial races + // on a fresh target group and every one of them has to keep all its targets. + for trial := range 5 { + tgOut, err := svc.CreateTargetGroup(context.Background(), &elbv2.CreateTargetGroupInput{ + Name: aws.String(fmt.Sprintf("race-tg-%d", trial)), + Port: aws.Int64(80), + }, testAccountID) + require.NoError(t, err) + tgArn := tgOut.TargetGroups[0].TargetGroupArn + + want := make([]string, 0, targetsPerTrial) + for i := range targetsPerTrial { + want = append(want, fmt.Sprintf("i-race%02d", i)) + } + + start := make(chan struct{}) + errs := make([]error, targetsPerTrial) + var wg sync.WaitGroup + for i := range targetsPerTrial { + wg.Go(func() { + <-start + _, errs[i] = svc.RegisterTargets(context.Background(), &elbv2.RegisterTargetsInput{ + TargetGroupArn: tgArn, + Targets: []*elbv2.TargetDescription{{Id: aws.String(want[i])}}, + }, testAccountID) + }) + } + close(start) + wg.Wait() + + for i, regErr := range errs { + require.NoError(t, regErr, "trial %d: registration of %s failed", trial, want[i]) + } + + tg, err := svc.store.GetTargetGroupByArn(t.Context(), *tgArn) + require.NoError(t, err) + got := make([]string, 0, len(tg.Targets)) + for _, target := range tg.Targets { + got = append(got, target.Id) + } + assert.ElementsMatch(t, want, got, "trial %d lost a concurrent registration", trial) + } +} + +// Deregistration is the same read-modify-write on the same record, so it has to +// hold under the same contention or a destroy leaves orphaned targets behind. +func TestDeregisterTargets_ConcurrentDeregistrationsAllLand(t *testing.T) { + t.Parallel() + svc := setupTestService(t) + + const registered = 8 + const removing = 6 + for trial := range 3 { + tgOut, err := svc.CreateTargetGroup(context.Background(), &elbv2.CreateTargetGroupInput{ + Name: aws.String(fmt.Sprintf("dereg-race-tg-%d", trial)), + Port: aws.Int64(80), + }, testAccountID) + require.NoError(t, err) + tgArn := tgOut.TargetGroups[0].TargetGroupArn + + ids := make([]*elbv2.TargetDescription, 0, registered) + for i := range registered { + ids = append(ids, &elbv2.TargetDescription{Id: aws.String(fmt.Sprintf("i-drace%02d", i))}) + } + _, err = svc.RegisterTargets(context.Background(), &elbv2.RegisterTargetsInput{ + TargetGroupArn: tgArn, Targets: ids, + }, testAccountID) + require.NoError(t, err) + + start := make(chan struct{}) + errs := make([]error, removing) + var wg sync.WaitGroup + for i := range removing { + wg.Go(func() { + <-start + _, errs[i] = svc.DeregisterTargets(context.Background(), &elbv2.DeregisterTargetsInput{ + TargetGroupArn: tgArn, + Targets: []*elbv2.TargetDescription{ids[i]}, + }, testAccountID) + }) + } + close(start) + wg.Wait() + + for i, deregErr := range errs { + require.NoError(t, deregErr, "trial %d: deregistration %d failed", trial, i) + } + + tg, err := svc.store.GetTargetGroupByArn(t.Context(), *tgArn) + require.NoError(t, err) + got := make([]string, 0, len(tg.Targets)) + for _, target := range tg.Targets { + got = append(got, target.Id) + } + assert.ElementsMatch(t, []string{"i-drace06", "i-drace07"}, got, + "trial %d: a concurrent deregistration was lost or over-applied", trial) + } +} + +// A CAS loop on one writer is defeated by a blind whole-record write from +// another, so the tag and attribute paths have to leave the target list alone. +func TestRegisterTargets_SurvivesConcurrentTagAndAttributeWrites(t *testing.T) { + t.Parallel() + svc := setupTestService(t) + + tgOut, err := svc.CreateTargetGroup(context.Background(), &elbv2.CreateTargetGroupInput{ + Name: aws.String("clobber-tg"), + Port: aws.Int64(80), + }, testAccountID) + require.NoError(t, err) + tgArn := tgOut.TargetGroups[0].TargetGroupArn + + const targets = 6 + start := make(chan struct{}) + errs := make([]error, targets+2) + var wg sync.WaitGroup + for i := range targets { + wg.Go(func() { + <-start + _, errs[i] = svc.RegisterTargets(context.Background(), &elbv2.RegisterTargetsInput{ + TargetGroupArn: tgArn, + Targets: []*elbv2.TargetDescription{{Id: aws.String(fmt.Sprintf("i-clob%02d", i))}}, + }, testAccountID) + }) + } + wg.Go(func() { + <-start + _, errs[targets] = svc.AddTags(context.Background(), &elbv2.AddTagsInput{ + ResourceArns: []*string{tgArn}, + Tags: []*elbv2.Tag{{Key: aws.String("Name"), Value: aws.String("clobber")}}, + }, testAccountID) + }) + wg.Go(func() { + <-start + _, errs[targets+1] = svc.ModifyTargetGroupAttributes(context.Background(), &elbv2.ModifyTargetGroupAttributesInput{ + TargetGroupArn: tgArn, + Attributes: []*elbv2.TargetGroupAttribute{ + {Key: aws.String("deregistration_delay.timeout_seconds"), Value: aws.String("17")}, + }, + }, testAccountID) + }) + close(start) + wg.Wait() + + for i, writeErr := range errs { + require.NoError(t, writeErr, "writer %d failed", i) + } + + tg, err := svc.store.GetTargetGroupByArn(t.Context(), *tgArn) + require.NoError(t, err) + got := make([]string, 0, len(tg.Targets)) + for _, target := range tg.Targets { + got = append(got, target.Id) + } + want := make([]string, 0, targets) + for i := range targets { + want = append(want, fmt.Sprintf("i-clob%02d", i)) + } + assert.ElementsMatch(t, want, got, "a tag or attribute write discarded a registration") + assert.Equal(t, "clobber", tg.Tags["Name"]) + assert.Equal(t, "17", tg.Attributes["deregistration_delay.timeout_seconds"]) +} diff --git a/spinifex/handlers/elbv2/store.go b/spinifex/handlers/elbv2/store.go index e304a8338..818eb522e 100644 --- a/spinifex/handlers/elbv2/store.go +++ b/spinifex/handlers/elbv2/store.go @@ -35,6 +35,11 @@ const ( // maxLBNameClaimRetries bounds the crash-orphan CAS-reclaim loop in ClaimLBName. maxLBNameClaimRetries = 5 + // targetGroupCASAttempts bounds a target-group read-modify-write. It sits + // above the kvutil default because Terraform registers targets at its own + // parallelism (10 by default), every one contending for the same record. + targetGroupCASAttempts = 12 + // lbNameClaimTTL bounds how long an unresolved claim blocks other creators // before ClaimLBName treats it as a crash orphan. A legitimate create finishes // well under this window; it only exists to reclaim an abandoned name. @@ -264,7 +269,45 @@ func (s *Store) GetLoadBalancerByName(ctx context.Context, name, accountID strin // --- Target Group CRUD --- -// PutTargetGroup stores a target group record. +// ErrTargetGroupNotFound is what a target-group CAS update reports when the +// record is absent or its stored ARN does not match the one addressed, so a +// caller can map it onto its own API error. +var ErrTargetGroupNotFound = errors.New("elbv2: target group not found") + +// UpdateTargetGroup applies mutate to one target group under optimistic +// concurrency: a writer that loses the race re-reads and retries instead of +// overwriting the winner. mutate returns false to commit nothing. +func (s *Store) UpdateTargetGroup(ctx context.Context, tgID string, mutate func(*TargetGroupRecord) (bool, error)) (*TargetGroupRecord, error) { + return kvutil.Update(ctx, s.kv, KeyPrefixTG+tgID, kvutil.CASConfig{ + Attempts: targetGroupCASAttempts, + NotFound: ErrTargetGroupNotFound, + Exhausted: func(_ string, attempts int) error { + return fmt.Errorf("elbv2: target group %s contended, CAS exhausted after %d attempts", tgID, attempts) + }, + }, mutate) +} + +// UpdateTargetGroupByArn is UpdateTargetGroup addressed by ARN. The short ID +// comes from the ARN's final segment (see GetTargetGroupByArn) and the stored +// ARN is re-checked under the CAS so a mismatched record is never mutated. +func (s *Store) UpdateTargetGroupByArn(ctx context.Context, arn string, mutate func(*TargetGroupRecord) (bool, error)) (*TargetGroupRecord, error) { + idx := strings.LastIndex(arn, "/") + if idx < 0 || idx == len(arn)-1 { + return nil, ErrTargetGroupNotFound + } + return s.UpdateTargetGroup(ctx, arn[idx+1:], func(tg *TargetGroupRecord) (bool, error) { + if tg.TargetGroupArn != arn { + slog.Error("target group KV record ARN mismatch", + "requested_arn", arn, "stored_arn", tg.TargetGroupArn) + return false, ErrTargetGroupNotFound + } + return mutate(tg) + }) +} + +// PutTargetGroup stores a target group record wholesale. Use UpdateTargetGroup +// for any read-modify-write: a blind Put of a stale record silently discards +// every field a concurrent writer changed. func (s *Store) PutTargetGroup(ctx context.Context, tg *TargetGroupRecord) error { data, err := json.Marshal(tg) if err != nil {