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 {