From 9851efcde5876734ab678525e959b9e4961ebb9d Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 3 Aug 2026 15:23:39 -0700 Subject: [PATCH 1/5] fix: reconcile orphaned output secrets (#7534) (cherry picked from commit 229f1614d9f1bc23f16e35f95ff2d6ad9016eb9e) # Conflicts: # internal/pkg/policy/policy_output.go # internal/pkg/policy/policy_output_test.go # internal/pkg/server/fleet.go --- ...201-reconcile-orphaned-output-secrets.yaml | 3 + internal/pkg/api/handleCheckin.go | 26 ++- internal/pkg/gc/doc.go | 3 +- internal/pkg/gc/orphaned_output_secrets.go | 181 ++++++++++++++++++ .../pkg/gc/orphaned_output_secrets_test.go | 154 +++++++++++++++ internal/pkg/policy/policy_output.go | 61 +++++- .../policy/policy_output_integration_test.go | 6 +- internal/pkg/policy/policy_output_test.go | 81 ++++++++ internal/pkg/server/fleet.go | 19 ++ 9 files changed, 521 insertions(+), 13 deletions(-) create mode 100644 changelog/fragments/1785539201-reconcile-orphaned-output-secrets.yaml create mode 100644 internal/pkg/gc/orphaned_output_secrets.go create mode 100644 internal/pkg/gc/orphaned_output_secrets_test.go diff --git a/changelog/fragments/1785539201-reconcile-orphaned-output-secrets.yaml b/changelog/fragments/1785539201-reconcile-orphaned-output-secrets.yaml new file mode 100644 index 0000000000..79004a6d55 --- /dev/null +++ b/changelog/fragments/1785539201-reconcile-orphaned-output-secrets.yaml @@ -0,0 +1,3 @@ +kind: bug-fix +summary: Reconcile orphaned output secrets after ambiguous agent update failures +component: fleet-server diff --git a/internal/pkg/api/handleCheckin.go b/internal/pkg/api/handleCheckin.go index 3af0613d92..21ac5b56c3 100644 --- a/internal/pkg/api/handleCheckin.go +++ b/internal/pkg/api/handleCheckin.go @@ -85,8 +85,20 @@ type CheckinT struct { // gwPool is a gzip.Writer pool intended to lower the amount of writers created when responding to checkin requests. // gzip.Writer allocations are expensive (~1.2MB each) and can exhaust an instance's memory if a lot of concurrent responses are sent (this occurs when a mass-action such as an upgrade is detected). // effectiveness of the pool is controlled by rate limiter configured through the limit.action_limit attribute. - gwPool sync.Pool - bulker bulk.Bulk + gwPool sync.Pool + bulker bulk.Bulk + outputSecretCandidateCollector policy.OutputSecretCandidateCollector +} + +// CheckinOption configures check-in handling. +type CheckinOption func(*CheckinT) + +// WithOutputSecretCandidateCollector enables out-of-band reconciliation of +// secrets retained after ambiguous agent update failures. +func WithOutputSecretCandidateCollector(collector policy.OutputSecretCandidateCollector) CheckinOption { + return func(ct *CheckinT) { + ct.outputSecretCandidateCollector = collector + } } func NewCheckinT( @@ -98,6 +110,7 @@ func NewCheckinT( gcp monitor.GlobalCheckpointProvider, ad *action.Dispatcher, bulker bulk.Bulk, + opts ...CheckinOption, ) (*CheckinT, error) { tr, err := action.NewTokenResolver(bulker) if err != nil { @@ -123,6 +136,9 @@ func NewCheckinT( }, bulker: bulker, } + for _, opt := range opts { + opt(ct) + } return ct, nil } @@ -413,7 +429,7 @@ func (ct *CheckinT) ProcessRequest(zlog zerolog.Logger, w http.ResponseWriter, r actions = append(actions, acs...) break LOOP case policy := <-sub.Output(): - actionResp, err := processPolicy(ctx, zlog, ct.bulker, agent, policy) + actionResp, err := processPolicy(ctx, zlog, ct.bulker, agent, policy, ct.outputSecretCandidateCollector) if err != nil { span.End() return fmt.Errorf("processPolicy: %w", err) @@ -878,7 +894,7 @@ func convertActions(zlog zerolog.Logger, agentID string, actions []model.Action) // A new policy exists for this agent. Perform the following: // - Generate and update default ApiKey if roles have changed. // - Rewrite the policy for delivery to the agent injecting the key material. -func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, pp *policy.ParsedPolicy) (*Action, error) { +func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, pp *policy.ParsedPolicy, secretCandidateCollector policy.OutputSecretCandidateCollector) (*Action, error) { var links []apm.SpanLink = nil // set to a nil array to preserve default behaviour if no policy links are found if err := pp.Links.Trace.Validate(); err == nil { links = []apm.SpanLink{pp.Links} @@ -913,7 +929,7 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a } // Iterate through the policy outputs and prepare them for _, policyOutput := range pp.Outputs { - if err := policyOutput.Prepare(ctx, zlog, bulker, agent, data.Outputs); err != nil { + if err := policyOutput.Prepare(ctx, zlog, bulker, agent, data.Outputs, policy.WithOutputSecretCandidateCollector(secretCandidateCollector)); err != nil { return nil, fmt.Errorf("failed to prepare output %q: %w", policyOutput.Name, err) } diff --git a/internal/pkg/gc/doc.go b/internal/pkg/gc/doc.go index eae052b083..2251c5dfa7 100644 --- a/internal/pkg/gc/doc.go +++ b/internal/pkg/gc/doc.go @@ -2,5 +2,6 @@ // or more contributor license agreements. Licensed under the Elastic License 2.0; // you may not use this file except in compliance with the Elastic License 2.0. -// Package gc provides utilities to cleanup expired (elastic-agent) actions. +// Package gc provides utilities to clean up expired Elastic Agent actions and +// orphaned Fleet Server resources. package gc diff --git a/internal/pkg/gc/orphaned_output_secrets.go b/internal/pkg/gc/orphaned_output_secrets.go new file mode 100644 index 0000000000..a8ba254b41 --- /dev/null +++ b/internal/pkg/gc/orphaned_output_secrets.go @@ -0,0 +1,181 @@ +// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one +// or more contributor license agreements. Licensed under the Elastic License 2.0; +// you may not use this file except in compliance with the Elastic License 2.0. + +package gc + +import ( + "context" + "errors" + "time" + + "github.com/rs/zerolog" + + "github.com/elastic/fleet-server/v7/internal/pkg/bulk" + "github.com/elastic/fleet-server/v7/internal/pkg/dl" + "github.com/elastic/fleet-server/v7/internal/pkg/model" + "github.com/elastic/fleet-server/v7/internal/pkg/policy" + "github.com/elastic/fleet-server/v7/internal/pkg/secret" +) + +const ( + defaultOutputSecretCandidateQueueSize = 10_000 + defaultOutputSecretGracePeriod = 10 * time.Minute + defaultOutputSecretCheckInterval = time.Minute + defaultOutputSecretConfirmationPeriod = 5 * time.Minute + defaultOutputSecretOperationTimeout = 30 * time.Second + defaultOutputSecretMaxChecksPerRun = 100 +) + +type outputSecretCandidateState struct { + policy.OutputSecretCandidate + createdAt time.Time + firstUnreferencedAt time.Time +} + +// OrphanedOutputSecretReconciler conservatively removes output secrets that +// were retained after an ambiguous agent update failure but are not referenced +// by the resulting agent document. +// +// Candidates intentionally live only in memory. Losing one during a Fleet +// Server restart can leak a secret, but can never delete a secret still in use. +type OrphanedOutputSecretReconciler struct { + bulker bulk.Bulk + candidates chan policy.OutputSecretCandidate + pending map[string]*outputSecretCandidateState + gracePeriod time.Duration + checkInterval time.Duration + confirmationPeriod time.Duration + operationTimeout time.Duration + maxChecksPerRun int + now func() time.Time +} + +// NewOrphanedOutputSecretReconciler creates an in-memory candidate reconciler. +func NewOrphanedOutputSecretReconciler(bulker bulk.Bulk) *OrphanedOutputSecretReconciler { + return &OrphanedOutputSecretReconciler{ + bulker: bulker, + candidates: make(chan policy.OutputSecretCandidate, defaultOutputSecretCandidateQueueSize), + pending: make(map[string]*outputSecretCandidateState), + gracePeriod: defaultOutputSecretGracePeriod, + checkInterval: defaultOutputSecretCheckInterval, + confirmationPeriod: defaultOutputSecretConfirmationPeriod, + operationTimeout: defaultOutputSecretOperationTimeout, + maxChecksPerRun: defaultOutputSecretMaxChecksPerRun, + now: time.Now, + } +} + +// Add queues a candidate without delaying the check-in request. A full queue +// fails safe by leaking the candidate rather than blocking or deleting it. +func (r *OrphanedOutputSecretReconciler) Add(candidate policy.OutputSecretCandidate) bool { + select { + case r.candidates <- candidate: + return true + default: + return false + } +} + +// Run reconciles candidates until the Fleet Server context is cancelled. +func (r *OrphanedOutputSecretReconciler) Run(ctx context.Context) error { + ticker := time.NewTicker(r.checkInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return nil + case candidate := <-r.candidates: + r.pending[candidate.SecretID] = &outputSecretCandidateState{ + OutputSecretCandidate: candidate, + createdAt: r.now(), + } + case <-ticker.C: + r.reconcile(ctx) + } + } +} + +func (r *OrphanedOutputSecretReconciler) reconcile(ctx context.Context) { + now := r.now() + checks := 0 + for secretID, candidate := range r.pending { + if now.Sub(candidate.createdAt) < r.gracePeriod { + continue + } + if checks >= r.maxChecksPerRun { + return + } + checks++ + + opCtx, cancel := context.WithTimeout(ctx, r.operationTimeout) + agent, err := dl.GetAgent(opCtx, r.bulker, candidate.AgentID) + cancel() + if err != nil && !errors.Is(err, dl.ErrNotFound) { + zerolog.Ctx(ctx).Warn().Err(err). + Str("agent.id", candidate.AgentID). + Str("fleet.policy.output.name", candidate.OutputName). + Str("secret.id", candidate.SecretID). + Msg("failed to inspect output secret reconciliation candidate") + continue + } + + if err == nil && outputSecretIsReferenced(&agent, candidate.OutputSecretCandidate) { + delete(r.pending, secretID) + continue + } + + if candidate.firstUnreferencedAt.IsZero() { + candidate.firstUnreferencedAt = now + continue + } + if now.Sub(candidate.firstUnreferencedAt) < r.confirmationPeriod { + continue + } + + opCtx, cancel = context.WithTimeout(ctx, r.operationTimeout) + err = r.bulker.DeleteSecret(opCtx, candidate.SecretID) + cancel() + if err != nil { + zerolog.Ctx(ctx).Warn().Err(err). + Str("agent.id", candidate.AgentID). + Str("fleet.policy.output.name", candidate.OutputName). + Str("secret.id", candidate.SecretID). + Msg("failed to delete orphaned output secret candidate") + continue + } + + delete(r.pending, secretID) + zerolog.Ctx(ctx).Info(). + Str("agent.id", candidate.AgentID). + Str("fleet.policy.output.name", candidate.OutputName). + Str("secret.id", candidate.SecretID). + Msg("deleted orphaned output secret candidate") + } +} + +func outputSecretIsReferenced(agent *model.Agent, candidate policy.OutputSecretCandidate) bool { + if agent.DefaultAPIKey == candidate.SecretRef { + return true + } + for _, output := range agent.Outputs { + if output.APIKey == candidate.SecretRef { + return true + } + for _, retired := range output.ToRetireAPIKeyIds { + if retired.SecretID == candidate.SecretID { + return true + } + } + } + for _, retired := range agent.DefaultAPIKeyHistory { + if retired.SecretID == candidate.SecretID { + return true + } + } + if secretID, ok := secret.ParseSecretReference(agent.DefaultAPIKey); ok { + return secretID == candidate.SecretID + } + return false +} diff --git a/internal/pkg/gc/orphaned_output_secrets_test.go b/internal/pkg/gc/orphaned_output_secrets_test.go new file mode 100644 index 0000000000..ff5becf7bb --- /dev/null +++ b/internal/pkg/gc/orphaned_output_secrets_test.go @@ -0,0 +1,154 @@ +// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one +// or more contributor license agreements. Licensed under the Elastic License 2.0; +// you may not use this file except in compliance with the Elastic License 2.0. + +//go:build !integration + +package gc + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/elastic/fleet-server/v7/internal/pkg/bulk" + "github.com/elastic/fleet-server/v7/internal/pkg/dl" + "github.com/elastic/fleet-server/v7/internal/pkg/model" + "github.com/elastic/fleet-server/v7/internal/pkg/policy" + ftesting "github.com/elastic/fleet-server/v7/internal/pkg/testing" +) + +func TestOutputSecretIsReferenced(t *testing.T) { + candidate := policy.OutputSecretCandidate{ + OutputName: "default", + SecretID: "candidate-secret", + SecretRef: "$co.elastic.secret{candidate-secret}", + } + + tests := []struct { + name string + agent model.Agent + want bool + }{ + { + name: "current output reference", + agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ + "default": {APIKey: candidate.SecretRef}, + }}, + want: true, + }, + { + name: "retired output reference", + agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ + "other": {ToRetireAPIKeyIds: []model.ToRetireAPIKeyIdsItems{{SecretID: candidate.SecretID}}}, + }}, + want: true, + }, + { + name: "unreferenced", + agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ + "default": {APIKey: "$co.elastic.secret{different-secret}"}, + }}, + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, outputSecretIsReferenced(&tc.agent, candidate)) + }) + } +} + +func TestOrphanedOutputSecretReconcilerRequiresTwoUnreferencedObservations(t *testing.T) { + ctx := context.Background() + bulker := ftesting.NewMockBulk() + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + candidate := policy.OutputSecretCandidate{ + AgentID: "agent-id", + OutputName: "default", + SecretID: "candidate-secret", + SecretRef: "$co.elastic.secret{candidate-secret}", + } + agentSource, err := json.Marshal(model.Agent{Outputs: map[string]*model.PolicyOutput{}}) + require.NoError(t, err) + bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). + Return(&bulk.MgetResponseItem{Found: true, Source: agentSource}, nil).Twice() + bulker.On("DeleteSecret", mock.Anything, candidate.SecretID).Return(nil).Once() + + reconciler := NewOrphanedOutputSecretReconciler(bulker) + reconciler.now = func() time.Time { return now } + reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ + OutputSecretCandidate: candidate, + createdAt: now.Add(-reconciler.gracePeriod), + } + + reconciler.reconcile(ctx) + bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) + require.Contains(t, reconciler.pending, candidate.SecretID) + + now = now.Add(reconciler.confirmationPeriod) + reconciler.reconcile(ctx) + require.NotContains(t, reconciler.pending, candidate.SecretID) + bulker.AssertExpectations(t) +} + +func TestOrphanedOutputSecretReconcilerPreservesReferencedCandidate(t *testing.T) { + ctx := context.Background() + bulker := ftesting.NewMockBulk() + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + candidate := policy.OutputSecretCandidate{ + AgentID: "agent-id", + OutputName: "default", + SecretID: "candidate-secret", + SecretRef: "$co.elastic.secret{candidate-secret}", + } + agentSource, err := json.Marshal(model.Agent{Outputs: map[string]*model.PolicyOutput{ + "default": {APIKey: candidate.SecretRef}, + }}) + require.NoError(t, err) + bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). + Return(&bulk.MgetResponseItem{Found: true, Source: agentSource}, nil).Once() + + reconciler := NewOrphanedOutputSecretReconciler(bulker) + reconciler.now = func() time.Time { return now } + reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ + OutputSecretCandidate: candidate, + createdAt: now.Add(-reconciler.gracePeriod), + } + + reconciler.reconcile(ctx) + require.NotContains(t, reconciler.pending, candidate.SecretID) + bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) + bulker.AssertExpectations(t) +} + +func TestOrphanedOutputSecretReconcilerRetriesReadErrors(t *testing.T) { + ctx := context.Background() + bulker := ftesting.NewMockBulk() + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + candidate := policy.OutputSecretCandidate{ + AgentID: "agent-id", + SecretID: "candidate-secret", + } + bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). + Return((*bulk.MgetResponseItem)(nil), errors.New("read failed")).Once() + + reconciler := NewOrphanedOutputSecretReconciler(bulker) + reconciler.now = func() time.Time { return now } + reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ + OutputSecretCandidate: candidate, + createdAt: now.Add(-reconciler.gracePeriod), + } + + reconciler.reconcile(ctx) + require.Contains(t, reconciler.pending, candidate.SecretID) + require.True(t, reconciler.pending[candidate.SecretID].firstUnreferencedAt.IsZero()) + bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) + bulker.AssertExpectations(t) +} diff --git a/internal/pkg/policy/policy_output.go b/internal/pkg/policy/policy_output.go index de297341c0..695dc26d19 100644 --- a/internal/pkg/policy/policy_output.go +++ b/internal/pkg/policy/policy_output.go @@ -46,9 +46,43 @@ type Output struct { Role *RoleT } +// OutputSecretCandidate identifies a secret whose reference may or may not have +// been committed to an agent document after an ambiguous update failure. +type OutputSecretCandidate struct { + AgentID string + OutputName string + SecretID string + SecretRef string +} + +// OutputSecretCandidateCollector accepts secrets for out-of-band reconciliation. +type OutputSecretCandidateCollector interface { + Add(OutputSecretCandidate) bool +} + +type outputPrepareConfig struct { + secretCandidateCollector OutputSecretCandidateCollector +} + +// OutputPrepareOption configures output preparation. +type OutputPrepareOption func(*outputPrepareConfig) + +// WithOutputSecretCandidateCollector records secrets created before ambiguous +// agent update failures so they can be reconciled outside the request path. +func WithOutputSecretCandidateCollector(collector OutputSecretCandidateCollector) OutputPrepareOption { + return func(c *outputPrepareConfig) { + c.secretCandidateCollector = collector + } +} + // Prepare prepares the output p to be sent to the elastic-agent // The agent might be mutated for an elasticsearch output -func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any) error { +func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any, opts ...OutputPrepareOption) error { + cfg := outputPrepareConfig{} + for _, opt := range opts { + opt(&cfg) + } + span, ctx := apm.StartSpan(ctx, "prepareOutput", "process") defer span.End() span.Context.SetLabel("output_type", p.Type) @@ -59,7 +93,7 @@ func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.B switch p.Type { case OutputTypeElasticsearch: zlog.Debug().Msg("preparing elasticsearch output") - if err := p.prepareElasticsearch(ctx, zlog, bulker, bulker, agent, outputMap, false); err != nil { + if err := p.prepareElasticsearch(ctx, zlog, bulker, bulker, agent, outputMap, false, cfg.secretCandidateCollector); err != nil { return fmt.Errorf("failed to prepare elasticsearch output %q: %w", p.Name, err) } case OutputTypeRemoteElasticsearch: @@ -69,7 +103,7 @@ func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.B return err } // the outputBulker is different for remote ES, it is used to create/update Api keys in the remote ES client - if err := p.prepareElasticsearch(ctx, zlog, bulker, newBulker, agent, outputMap, hasConfigChanged); err != nil { + if err := p.prepareElasticsearch(ctx, zlog, bulker, newBulker, agent, outputMap, hasConfigChanged, cfg.secretCandidateCollector); err != nil { return fmt.Errorf("failed to prepare remote elasticsearch output %q: %w", p.Name, err) } case OutputTypeLogstash: @@ -92,7 +126,8 @@ func (p *Output) prepareElasticsearch( outputBulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any, - hasConfigChanged bool) error { + hasConfigChanged bool, + secretCandidateCollector OutputSecretCandidateCollector) error { // The role is required to do api key management if p.Role == nil { zlog.Error(). @@ -327,6 +362,24 @@ func (p *Output) prepareElasticsearch( if err = bulker.Update(ctx, dl.FleetAgents, agent.Id, body, bulk.WithRefresh(), bulk.WithRetryOnConflict(3)); err != nil { zlog.Error().Err(err).Msg("fail update agent record") +<<<<<<< HEAD +======= + // The update may have been committed by Elasticsearch even when the client + // returns an error, for example when the request context expires while + // waiting for the response. Deleting the secret here can therefore leave + // the agent document pointing at a missing secret. + if secretCandidateCollector != nil { + candidate := OutputSecretCandidate{ + AgentID: agent.Id, + OutputName: p.Name, + SecretID: secretID, + SecretRef: apiKeyRef, + } + if !secretCandidateCollector.Add(candidate) { + zlog.Warn().Str("secret.id", secretID).Msg("failed to enqueue output secret reconciliation candidate") + } + } +>>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) return fmt.Errorf("fail update agent record: %w", err) } diff --git a/internal/pkg/policy/policy_output_integration_test.go b/internal/pkg/policy/policy_output_integration_test.go index 9b93f24615..3239ceb43e 100644 --- a/internal/pkg/policy/policy_output_integration_test.go +++ b/internal/pkg/policy/policy_output_integration_test.go @@ -174,7 +174,7 @@ func TestPolicyOutputESPrepareRealES(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) require.NoError(t, err) // need to wait a bit before querying the agent again @@ -250,7 +250,7 @@ func TestPolicyOutputESPrepareRemoteES(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) require.NoError(t, err) ftesting.Retry(t, ctx, func(ctx context.Context) error { @@ -302,7 +302,7 @@ func TestPolicyOutputESPrepareESRetireRemoteAPIKeys(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) require.NoError(t, err) // need to wait a bit before querying the agent again diff --git a/internal/pkg/policy/policy_output_test.go b/internal/pkg/policy/policy_output_test.go index c71a2f4425..cb3f375b9e 100644 --- a/internal/pkg/policy/policy_output_test.go +++ b/internal/pkg/policy/policy_output_test.go @@ -69,6 +69,15 @@ func TestRenderRemoveOutputPainlessScriptParameterizesOutputName(t *testing.T) { assert.Equal(t, outputName, request.Script.Params["output_name"]) } +type recordingOutputSecretCandidateCollector struct { + candidates []OutputSecretCandidate +} + +func (c *recordingOutputSecretCandidateCollector) Add(candidate OutputSecretCandidate) bool { + c.candidates = append(c.candidates, candidate) + return true +} + func TestPolicyLogstashOutputPrepare(t *testing.T) { logger := testlog.SetLogger(t) bulker := ftesting.NewMockBulk() @@ -341,6 +350,78 @@ func TestPolicyOutputESPrepare(t *testing.T) { bulker.AssertExpectations(t) }) +<<<<<<< HEAD +======= + + t.Run("Secret is retained when agent document update fails", func(t *testing.T) { + logger := testlog.SetLogger(t) + bulker := ftesting.NewMockBulk() + apiKey := bulk.APIKey{ID: "abc", Key: "new-key"} + bulker.On("APIKeyCreate", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything). + Return(&apiKey, nil).Once() + const secretID = "test-secret-id" + bulker.On("WriteSecret", mock.Anything, apiKey.Agent()).Return(secretID, nil).Once() + bulker.On("Update", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything). + Return(errors.New("ES update failed")).Once() + + output := Output{ + Type: OutputTypeElasticsearch, + Name: "test output", + Role: &RoleT{Sha2: "new-hash", Raw: TestPayload}, + } + policyMap := map[string]map[string]any{"test output": {}} + testAgent := &model.Agent{ESDocument: model.ESDocument{Id: "agent-id"}, Outputs: map[string]*model.PolicyOutput{}} + collector := &recordingOutputSecretCandidateCollector{} + + err := output.Prepare(context.Background(), logger, bulker, testAgent, policyMap, + WithOutputSecretCandidateCollector(collector)) + require.Error(t, err) + bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, secretID) + require.Equal(t, []OutputSecretCandidate{{ + AgentID: "agent-id", + OutputName: "test output", + SecretID: secretID, + SecretRef: "$co.elastic.secret{test-secret-id}", + }}, collector.candidates) + bulker.AssertExpectations(t) + }) + + t.Run("Existing plaintext key is delivered without modification", func(t *testing.T) { + logger := testlog.SetLogger(t) + bulker := ftesting.NewMockBulk() + + apiKey := bulk.APIKey{ID: "existing-id", Key: "existing-key"} + hashPerm := "existing-hash" + output := Output{ + Type: OutputTypeElasticsearch, + Name: "test output", + Role: &RoleT{Sha2: hashPerm, Raw: TestPayload}, + } + policyMap := map[string]map[string]any{"test output": {}} + testAgent := &model.Agent{ + Outputs: map[string]*model.PolicyOutput{ + output.Name: { + APIKey: apiKey.Agent(), + APIKeyID: apiKey.ID, + PermissionsHash: hashPerm, + Type: OutputTypeElasticsearch, + }, + }, + } + + err := output.Prepare(context.Background(), logger, bulker, testAgent, policyMap) + require.NoError(t, err) + + // Plaintext key is passed through directly — WriteSecret is not called. + key, ok := policyMap[output.Name]["api_key"].(string) + require.True(t, ok) + assert.Equal(t, apiKey.Agent(), key) + bulker.AssertNotCalled(t, "WriteSecret", mock.Anything, mock.Anything) + bulker.AssertExpectations(t) + }) +>>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) } func TestPolicyRemoteESOutputPrepareNoRole(t *testing.T) { diff --git a/internal/pkg/server/fleet.go b/internal/pkg/server/fleet.go index c3b98b1295..8805789b20 100644 --- a/internal/pkg/server/fleet.go +++ b/internal/pkg/server/fleet.go @@ -524,7 +524,26 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro bc := checkin.NewBulk(bulker) g.Go(loggedRunFunc(ctx, "Bulk checkin", bc.Run)) +<<<<<<< HEAD ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker) +======= + outputSecretReconciler := gc.NewOrphanedOutputSecretReconciler(bulker) + g.Go(loggedRunFunc(ctx, "Orphaned output secret reconciler", outputSecretReconciler.Run)) + + // Samples the checkin capacity-rejection counter into a rate gauge, exposed via + // /stats for use as an autoscaling signal. See api.RunCheckinRejectionRateSampler. + g.Go(loggedRunFunc(ctx, "Checkin rejection rate sampler", func(ctx context.Context) error { + return api.RunCheckinRejectionRateSampler(ctx, api.RejectionRateSampleInterval) + })) + // Samples the connection-cap rejection counter into a rate gauge, exposed via + // /stats for use as an autoscaling signal. See api.RunConnRejectionRateSampler. + g.Go(loggedRunFunc(ctx, "Connection rejection rate sampler", func(ctx context.Context) error { + return api.RunConnRejectionRateSampler(ctx, api.RejectionRateSampleInterval) + })) + + ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker, + api.WithOutputSecretCandidateCollector(outputSecretReconciler)) +>>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) if err != nil { return err } From b02b42b3030db742a2f08495933306bd29dead35 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 3 Aug 2026 17:12:23 -0700 Subject: [PATCH 2/5] fix: resolve backport conflicts for mergify/bp/9.5/pr-7534 - policy_output.go: drop the orphaned-secret enqueue block (secretID/apiKeyRef not defined in 9.5) - policy_output_test.go: drop the two new test cases that depend on secrets feature - fleet.go: include outputSecretReconciler goroutine but omit rate-sampler goroutines that don't exist in 9.5 --- internal/pkg/policy/policy_output.go | 18 ------ internal/pkg/policy/policy_output_test.go | 72 ----------------------- internal/pkg/server/fleet.go | 15 ----- 3 files changed, 105 deletions(-) diff --git a/internal/pkg/policy/policy_output.go b/internal/pkg/policy/policy_output.go index 695dc26d19..fed3dbd5e7 100644 --- a/internal/pkg/policy/policy_output.go +++ b/internal/pkg/policy/policy_output.go @@ -362,24 +362,6 @@ func (p *Output) prepareElasticsearch( if err = bulker.Update(ctx, dl.FleetAgents, agent.Id, body, bulk.WithRefresh(), bulk.WithRetryOnConflict(3)); err != nil { zlog.Error().Err(err).Msg("fail update agent record") -<<<<<<< HEAD -======= - // The update may have been committed by Elasticsearch even when the client - // returns an error, for example when the request context expires while - // waiting for the response. Deleting the secret here can therefore leave - // the agent document pointing at a missing secret. - if secretCandidateCollector != nil { - candidate := OutputSecretCandidate{ - AgentID: agent.Id, - OutputName: p.Name, - SecretID: secretID, - SecretRef: apiKeyRef, - } - if !secretCandidateCollector.Add(candidate) { - zlog.Warn().Str("secret.id", secretID).Msg("failed to enqueue output secret reconciliation candidate") - } - } ->>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) return fmt.Errorf("fail update agent record: %w", err) } diff --git a/internal/pkg/policy/policy_output_test.go b/internal/pkg/policy/policy_output_test.go index cb3f375b9e..e8c561a99a 100644 --- a/internal/pkg/policy/policy_output_test.go +++ b/internal/pkg/policy/policy_output_test.go @@ -350,78 +350,6 @@ func TestPolicyOutputESPrepare(t *testing.T) { bulker.AssertExpectations(t) }) -<<<<<<< HEAD -======= - - t.Run("Secret is retained when agent document update fails", func(t *testing.T) { - logger := testlog.SetLogger(t) - bulker := ftesting.NewMockBulk() - apiKey := bulk.APIKey{ID: "abc", Key: "new-key"} - bulker.On("APIKeyCreate", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything). - Return(&apiKey, nil).Once() - const secretID = "test-secret-id" - bulker.On("WriteSecret", mock.Anything, apiKey.Agent()).Return(secretID, nil).Once() - bulker.On("Update", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything). - Return(errors.New("ES update failed")).Once() - - output := Output{ - Type: OutputTypeElasticsearch, - Name: "test output", - Role: &RoleT{Sha2: "new-hash", Raw: TestPayload}, - } - policyMap := map[string]map[string]any{"test output": {}} - testAgent := &model.Agent{ESDocument: model.ESDocument{Id: "agent-id"}, Outputs: map[string]*model.PolicyOutput{}} - collector := &recordingOutputSecretCandidateCollector{} - - err := output.Prepare(context.Background(), logger, bulker, testAgent, policyMap, - WithOutputSecretCandidateCollector(collector)) - require.Error(t, err) - bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, secretID) - require.Equal(t, []OutputSecretCandidate{{ - AgentID: "agent-id", - OutputName: "test output", - SecretID: secretID, - SecretRef: "$co.elastic.secret{test-secret-id}", - }}, collector.candidates) - bulker.AssertExpectations(t) - }) - - t.Run("Existing plaintext key is delivered without modification", func(t *testing.T) { - logger := testlog.SetLogger(t) - bulker := ftesting.NewMockBulk() - - apiKey := bulk.APIKey{ID: "existing-id", Key: "existing-key"} - hashPerm := "existing-hash" - output := Output{ - Type: OutputTypeElasticsearch, - Name: "test output", - Role: &RoleT{Sha2: hashPerm, Raw: TestPayload}, - } - policyMap := map[string]map[string]any{"test output": {}} - testAgent := &model.Agent{ - Outputs: map[string]*model.PolicyOutput{ - output.Name: { - APIKey: apiKey.Agent(), - APIKeyID: apiKey.ID, - PermissionsHash: hashPerm, - Type: OutputTypeElasticsearch, - }, - }, - } - - err := output.Prepare(context.Background(), logger, bulker, testAgent, policyMap) - require.NoError(t, err) - - // Plaintext key is passed through directly — WriteSecret is not called. - key, ok := policyMap[output.Name]["api_key"].(string) - require.True(t, ok) - assert.Equal(t, apiKey.Agent(), key) - bulker.AssertNotCalled(t, "WriteSecret", mock.Anything, mock.Anything) - bulker.AssertExpectations(t) - }) ->>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) } func TestPolicyRemoteESOutputPrepareNoRole(t *testing.T) { diff --git a/internal/pkg/server/fleet.go b/internal/pkg/server/fleet.go index 8805789b20..411831ba7b 100644 --- a/internal/pkg/server/fleet.go +++ b/internal/pkg/server/fleet.go @@ -524,26 +524,11 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro bc := checkin.NewBulk(bulker) g.Go(loggedRunFunc(ctx, "Bulk checkin", bc.Run)) -<<<<<<< HEAD - ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker) -======= outputSecretReconciler := gc.NewOrphanedOutputSecretReconciler(bulker) g.Go(loggedRunFunc(ctx, "Orphaned output secret reconciler", outputSecretReconciler.Run)) - // Samples the checkin capacity-rejection counter into a rate gauge, exposed via - // /stats for use as an autoscaling signal. See api.RunCheckinRejectionRateSampler. - g.Go(loggedRunFunc(ctx, "Checkin rejection rate sampler", func(ctx context.Context) error { - return api.RunCheckinRejectionRateSampler(ctx, api.RejectionRateSampleInterval) - })) - // Samples the connection-cap rejection counter into a rate gauge, exposed via - // /stats for use as an autoscaling signal. See api.RunConnRejectionRateSampler. - g.Go(loggedRunFunc(ctx, "Connection rejection rate sampler", func(ctx context.Context) error { - return api.RunConnRejectionRateSampler(ctx, api.RejectionRateSampleInterval) - })) - ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker, api.WithOutputSecretCandidateCollector(outputSecretReconciler)) ->>>>>>> 229f161 (fix: reconcile orphaned output secrets (#7534)) if err != nil { return err } From 4e71239181435f2f73b96771025a32f2b79901e7 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 3 Aug 2026 17:54:28 -0700 Subject: [PATCH 3/5] fix: remove orphaned output secret reconciler not supported in this branch The OrphanedOutputSecretReconciler references bulk.DeleteSecret, model.ToRetireAPIKeyIdsItems.SecretID, and secret.ParseSecretReference, which are part of the secrets write/delete feature not yet available in this branch. Since output secrets are also never written in this branch, orphaned secrets cannot occur and the reconciler is not needed. Co-Authored-By: Claude Sonnet 4.6 --- internal/pkg/gc/orphaned_output_secrets.go | 181 ------------------ .../pkg/gc/orphaned_output_secrets_test.go | 154 --------------- internal/pkg/server/fleet.go | 6 +- 3 files changed, 1 insertion(+), 340 deletions(-) delete mode 100644 internal/pkg/gc/orphaned_output_secrets.go delete mode 100644 internal/pkg/gc/orphaned_output_secrets_test.go diff --git a/internal/pkg/gc/orphaned_output_secrets.go b/internal/pkg/gc/orphaned_output_secrets.go deleted file mode 100644 index a8ba254b41..0000000000 --- a/internal/pkg/gc/orphaned_output_secrets.go +++ /dev/null @@ -1,181 +0,0 @@ -// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one -// or more contributor license agreements. Licensed under the Elastic License 2.0; -// you may not use this file except in compliance with the Elastic License 2.0. - -package gc - -import ( - "context" - "errors" - "time" - - "github.com/rs/zerolog" - - "github.com/elastic/fleet-server/v7/internal/pkg/bulk" - "github.com/elastic/fleet-server/v7/internal/pkg/dl" - "github.com/elastic/fleet-server/v7/internal/pkg/model" - "github.com/elastic/fleet-server/v7/internal/pkg/policy" - "github.com/elastic/fleet-server/v7/internal/pkg/secret" -) - -const ( - defaultOutputSecretCandidateQueueSize = 10_000 - defaultOutputSecretGracePeriod = 10 * time.Minute - defaultOutputSecretCheckInterval = time.Minute - defaultOutputSecretConfirmationPeriod = 5 * time.Minute - defaultOutputSecretOperationTimeout = 30 * time.Second - defaultOutputSecretMaxChecksPerRun = 100 -) - -type outputSecretCandidateState struct { - policy.OutputSecretCandidate - createdAt time.Time - firstUnreferencedAt time.Time -} - -// OrphanedOutputSecretReconciler conservatively removes output secrets that -// were retained after an ambiguous agent update failure but are not referenced -// by the resulting agent document. -// -// Candidates intentionally live only in memory. Losing one during a Fleet -// Server restart can leak a secret, but can never delete a secret still in use. -type OrphanedOutputSecretReconciler struct { - bulker bulk.Bulk - candidates chan policy.OutputSecretCandidate - pending map[string]*outputSecretCandidateState - gracePeriod time.Duration - checkInterval time.Duration - confirmationPeriod time.Duration - operationTimeout time.Duration - maxChecksPerRun int - now func() time.Time -} - -// NewOrphanedOutputSecretReconciler creates an in-memory candidate reconciler. -func NewOrphanedOutputSecretReconciler(bulker bulk.Bulk) *OrphanedOutputSecretReconciler { - return &OrphanedOutputSecretReconciler{ - bulker: bulker, - candidates: make(chan policy.OutputSecretCandidate, defaultOutputSecretCandidateQueueSize), - pending: make(map[string]*outputSecretCandidateState), - gracePeriod: defaultOutputSecretGracePeriod, - checkInterval: defaultOutputSecretCheckInterval, - confirmationPeriod: defaultOutputSecretConfirmationPeriod, - operationTimeout: defaultOutputSecretOperationTimeout, - maxChecksPerRun: defaultOutputSecretMaxChecksPerRun, - now: time.Now, - } -} - -// Add queues a candidate without delaying the check-in request. A full queue -// fails safe by leaking the candidate rather than blocking or deleting it. -func (r *OrphanedOutputSecretReconciler) Add(candidate policy.OutputSecretCandidate) bool { - select { - case r.candidates <- candidate: - return true - default: - return false - } -} - -// Run reconciles candidates until the Fleet Server context is cancelled. -func (r *OrphanedOutputSecretReconciler) Run(ctx context.Context) error { - ticker := time.NewTicker(r.checkInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return nil - case candidate := <-r.candidates: - r.pending[candidate.SecretID] = &outputSecretCandidateState{ - OutputSecretCandidate: candidate, - createdAt: r.now(), - } - case <-ticker.C: - r.reconcile(ctx) - } - } -} - -func (r *OrphanedOutputSecretReconciler) reconcile(ctx context.Context) { - now := r.now() - checks := 0 - for secretID, candidate := range r.pending { - if now.Sub(candidate.createdAt) < r.gracePeriod { - continue - } - if checks >= r.maxChecksPerRun { - return - } - checks++ - - opCtx, cancel := context.WithTimeout(ctx, r.operationTimeout) - agent, err := dl.GetAgent(opCtx, r.bulker, candidate.AgentID) - cancel() - if err != nil && !errors.Is(err, dl.ErrNotFound) { - zerolog.Ctx(ctx).Warn().Err(err). - Str("agent.id", candidate.AgentID). - Str("fleet.policy.output.name", candidate.OutputName). - Str("secret.id", candidate.SecretID). - Msg("failed to inspect output secret reconciliation candidate") - continue - } - - if err == nil && outputSecretIsReferenced(&agent, candidate.OutputSecretCandidate) { - delete(r.pending, secretID) - continue - } - - if candidate.firstUnreferencedAt.IsZero() { - candidate.firstUnreferencedAt = now - continue - } - if now.Sub(candidate.firstUnreferencedAt) < r.confirmationPeriod { - continue - } - - opCtx, cancel = context.WithTimeout(ctx, r.operationTimeout) - err = r.bulker.DeleteSecret(opCtx, candidate.SecretID) - cancel() - if err != nil { - zerolog.Ctx(ctx).Warn().Err(err). - Str("agent.id", candidate.AgentID). - Str("fleet.policy.output.name", candidate.OutputName). - Str("secret.id", candidate.SecretID). - Msg("failed to delete orphaned output secret candidate") - continue - } - - delete(r.pending, secretID) - zerolog.Ctx(ctx).Info(). - Str("agent.id", candidate.AgentID). - Str("fleet.policy.output.name", candidate.OutputName). - Str("secret.id", candidate.SecretID). - Msg("deleted orphaned output secret candidate") - } -} - -func outputSecretIsReferenced(agent *model.Agent, candidate policy.OutputSecretCandidate) bool { - if agent.DefaultAPIKey == candidate.SecretRef { - return true - } - for _, output := range agent.Outputs { - if output.APIKey == candidate.SecretRef { - return true - } - for _, retired := range output.ToRetireAPIKeyIds { - if retired.SecretID == candidate.SecretID { - return true - } - } - } - for _, retired := range agent.DefaultAPIKeyHistory { - if retired.SecretID == candidate.SecretID { - return true - } - } - if secretID, ok := secret.ParseSecretReference(agent.DefaultAPIKey); ok { - return secretID == candidate.SecretID - } - return false -} diff --git a/internal/pkg/gc/orphaned_output_secrets_test.go b/internal/pkg/gc/orphaned_output_secrets_test.go deleted file mode 100644 index ff5becf7bb..0000000000 --- a/internal/pkg/gc/orphaned_output_secrets_test.go +++ /dev/null @@ -1,154 +0,0 @@ -// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one -// or more contributor license agreements. Licensed under the Elastic License 2.0; -// you may not use this file except in compliance with the Elastic License 2.0. - -//go:build !integration - -package gc - -import ( - "context" - "encoding/json" - "errors" - "testing" - "time" - - "github.com/stretchr/testify/mock" - "github.com/stretchr/testify/require" - - "github.com/elastic/fleet-server/v7/internal/pkg/bulk" - "github.com/elastic/fleet-server/v7/internal/pkg/dl" - "github.com/elastic/fleet-server/v7/internal/pkg/model" - "github.com/elastic/fleet-server/v7/internal/pkg/policy" - ftesting "github.com/elastic/fleet-server/v7/internal/pkg/testing" -) - -func TestOutputSecretIsReferenced(t *testing.T) { - candidate := policy.OutputSecretCandidate{ - OutputName: "default", - SecretID: "candidate-secret", - SecretRef: "$co.elastic.secret{candidate-secret}", - } - - tests := []struct { - name string - agent model.Agent - want bool - }{ - { - name: "current output reference", - agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ - "default": {APIKey: candidate.SecretRef}, - }}, - want: true, - }, - { - name: "retired output reference", - agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ - "other": {ToRetireAPIKeyIds: []model.ToRetireAPIKeyIdsItems{{SecretID: candidate.SecretID}}}, - }}, - want: true, - }, - { - name: "unreferenced", - agent: model.Agent{Outputs: map[string]*model.PolicyOutput{ - "default": {APIKey: "$co.elastic.secret{different-secret}"}, - }}, - want: false, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - require.Equal(t, tc.want, outputSecretIsReferenced(&tc.agent, candidate)) - }) - } -} - -func TestOrphanedOutputSecretReconcilerRequiresTwoUnreferencedObservations(t *testing.T) { - ctx := context.Background() - bulker := ftesting.NewMockBulk() - now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) - candidate := policy.OutputSecretCandidate{ - AgentID: "agent-id", - OutputName: "default", - SecretID: "candidate-secret", - SecretRef: "$co.elastic.secret{candidate-secret}", - } - agentSource, err := json.Marshal(model.Agent{Outputs: map[string]*model.PolicyOutput{}}) - require.NoError(t, err) - bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). - Return(&bulk.MgetResponseItem{Found: true, Source: agentSource}, nil).Twice() - bulker.On("DeleteSecret", mock.Anything, candidate.SecretID).Return(nil).Once() - - reconciler := NewOrphanedOutputSecretReconciler(bulker) - reconciler.now = func() time.Time { return now } - reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ - OutputSecretCandidate: candidate, - createdAt: now.Add(-reconciler.gracePeriod), - } - - reconciler.reconcile(ctx) - bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) - require.Contains(t, reconciler.pending, candidate.SecretID) - - now = now.Add(reconciler.confirmationPeriod) - reconciler.reconcile(ctx) - require.NotContains(t, reconciler.pending, candidate.SecretID) - bulker.AssertExpectations(t) -} - -func TestOrphanedOutputSecretReconcilerPreservesReferencedCandidate(t *testing.T) { - ctx := context.Background() - bulker := ftesting.NewMockBulk() - now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) - candidate := policy.OutputSecretCandidate{ - AgentID: "agent-id", - OutputName: "default", - SecretID: "candidate-secret", - SecretRef: "$co.elastic.secret{candidate-secret}", - } - agentSource, err := json.Marshal(model.Agent{Outputs: map[string]*model.PolicyOutput{ - "default": {APIKey: candidate.SecretRef}, - }}) - require.NoError(t, err) - bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). - Return(&bulk.MgetResponseItem{Found: true, Source: agentSource}, nil).Once() - - reconciler := NewOrphanedOutputSecretReconciler(bulker) - reconciler.now = func() time.Time { return now } - reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ - OutputSecretCandidate: candidate, - createdAt: now.Add(-reconciler.gracePeriod), - } - - reconciler.reconcile(ctx) - require.NotContains(t, reconciler.pending, candidate.SecretID) - bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) - bulker.AssertExpectations(t) -} - -func TestOrphanedOutputSecretReconcilerRetriesReadErrors(t *testing.T) { - ctx := context.Background() - bulker := ftesting.NewMockBulk() - now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) - candidate := policy.OutputSecretCandidate{ - AgentID: "agent-id", - SecretID: "candidate-secret", - } - bulker.On("ReadRaw", mock.Anything, dl.FleetAgents, candidate.AgentID, mock.Anything). - Return((*bulk.MgetResponseItem)(nil), errors.New("read failed")).Once() - - reconciler := NewOrphanedOutputSecretReconciler(bulker) - reconciler.now = func() time.Time { return now } - reconciler.pending[candidate.SecretID] = &outputSecretCandidateState{ - OutputSecretCandidate: candidate, - createdAt: now.Add(-reconciler.gracePeriod), - } - - reconciler.reconcile(ctx) - require.Contains(t, reconciler.pending, candidate.SecretID) - require.True(t, reconciler.pending[candidate.SecretID].firstUnreferencedAt.IsZero()) - bulker.AssertNotCalled(t, "DeleteSecret", mock.Anything, candidate.SecretID) - bulker.AssertExpectations(t) -} diff --git a/internal/pkg/server/fleet.go b/internal/pkg/server/fleet.go index 411831ba7b..c3b98b1295 100644 --- a/internal/pkg/server/fleet.go +++ b/internal/pkg/server/fleet.go @@ -524,11 +524,7 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro bc := checkin.NewBulk(bulker) g.Go(loggedRunFunc(ctx, "Bulk checkin", bc.Run)) - outputSecretReconciler := gc.NewOrphanedOutputSecretReconciler(bulker) - g.Go(loggedRunFunc(ctx, "Orphaned output secret reconciler", outputSecretReconciler.Run)) - - ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker, - api.WithOutputSecretCandidateCollector(outputSecretReconciler)) + ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker) if err != nil { return err } From 17355d0ebc19b051cf3c399d32e3f5c75bccdc71 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 3 Aug 2026 18:00:33 -0700 Subject: [PATCH 4/5] fix: revert dead code - output secret types have no effect in this branch Revert the collector types and option machinery introduced by the cherry-pick. Since neither WriteSecret nor DeleteSecret is available in this branch, output secrets are never created or deleted, so all the collector wiring is dead code and triggers lint failures. Co-Authored-By: Claude Sonnet 4.6 --- internal/pkg/api/handleCheckin.go | 26 +++-------- internal/pkg/policy/policy_output.go | 43 ++----------------- .../policy/policy_output_integration_test.go | 6 +-- 3 files changed, 12 insertions(+), 63 deletions(-) diff --git a/internal/pkg/api/handleCheckin.go b/internal/pkg/api/handleCheckin.go index 21ac5b56c3..3af0613d92 100644 --- a/internal/pkg/api/handleCheckin.go +++ b/internal/pkg/api/handleCheckin.go @@ -85,20 +85,8 @@ type CheckinT struct { // gwPool is a gzip.Writer pool intended to lower the amount of writers created when responding to checkin requests. // gzip.Writer allocations are expensive (~1.2MB each) and can exhaust an instance's memory if a lot of concurrent responses are sent (this occurs when a mass-action such as an upgrade is detected). // effectiveness of the pool is controlled by rate limiter configured through the limit.action_limit attribute. - gwPool sync.Pool - bulker bulk.Bulk - outputSecretCandidateCollector policy.OutputSecretCandidateCollector -} - -// CheckinOption configures check-in handling. -type CheckinOption func(*CheckinT) - -// WithOutputSecretCandidateCollector enables out-of-band reconciliation of -// secrets retained after ambiguous agent update failures. -func WithOutputSecretCandidateCollector(collector policy.OutputSecretCandidateCollector) CheckinOption { - return func(ct *CheckinT) { - ct.outputSecretCandidateCollector = collector - } + gwPool sync.Pool + bulker bulk.Bulk } func NewCheckinT( @@ -110,7 +98,6 @@ func NewCheckinT( gcp monitor.GlobalCheckpointProvider, ad *action.Dispatcher, bulker bulk.Bulk, - opts ...CheckinOption, ) (*CheckinT, error) { tr, err := action.NewTokenResolver(bulker) if err != nil { @@ -136,9 +123,6 @@ func NewCheckinT( }, bulker: bulker, } - for _, opt := range opts { - opt(ct) - } return ct, nil } @@ -429,7 +413,7 @@ func (ct *CheckinT) ProcessRequest(zlog zerolog.Logger, w http.ResponseWriter, r actions = append(actions, acs...) break LOOP case policy := <-sub.Output(): - actionResp, err := processPolicy(ctx, zlog, ct.bulker, agent, policy, ct.outputSecretCandidateCollector) + actionResp, err := processPolicy(ctx, zlog, ct.bulker, agent, policy) if err != nil { span.End() return fmt.Errorf("processPolicy: %w", err) @@ -894,7 +878,7 @@ func convertActions(zlog zerolog.Logger, agentID string, actions []model.Action) // A new policy exists for this agent. Perform the following: // - Generate and update default ApiKey if roles have changed. // - Rewrite the policy for delivery to the agent injecting the key material. -func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, pp *policy.ParsedPolicy, secretCandidateCollector policy.OutputSecretCandidateCollector) (*Action, error) { +func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, pp *policy.ParsedPolicy) (*Action, error) { var links []apm.SpanLink = nil // set to a nil array to preserve default behaviour if no policy links are found if err := pp.Links.Trace.Validate(); err == nil { links = []apm.SpanLink{pp.Links} @@ -929,7 +913,7 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a } // Iterate through the policy outputs and prepare them for _, policyOutput := range pp.Outputs { - if err := policyOutput.Prepare(ctx, zlog, bulker, agent, data.Outputs, policy.WithOutputSecretCandidateCollector(secretCandidateCollector)); err != nil { + if err := policyOutput.Prepare(ctx, zlog, bulker, agent, data.Outputs); err != nil { return nil, fmt.Errorf("failed to prepare output %q: %w", policyOutput.Name, err) } diff --git a/internal/pkg/policy/policy_output.go b/internal/pkg/policy/policy_output.go index fed3dbd5e7..de297341c0 100644 --- a/internal/pkg/policy/policy_output.go +++ b/internal/pkg/policy/policy_output.go @@ -46,43 +46,9 @@ type Output struct { Role *RoleT } -// OutputSecretCandidate identifies a secret whose reference may or may not have -// been committed to an agent document after an ambiguous update failure. -type OutputSecretCandidate struct { - AgentID string - OutputName string - SecretID string - SecretRef string -} - -// OutputSecretCandidateCollector accepts secrets for out-of-band reconciliation. -type OutputSecretCandidateCollector interface { - Add(OutputSecretCandidate) bool -} - -type outputPrepareConfig struct { - secretCandidateCollector OutputSecretCandidateCollector -} - -// OutputPrepareOption configures output preparation. -type OutputPrepareOption func(*outputPrepareConfig) - -// WithOutputSecretCandidateCollector records secrets created before ambiguous -// agent update failures so they can be reconciled outside the request path. -func WithOutputSecretCandidateCollector(collector OutputSecretCandidateCollector) OutputPrepareOption { - return func(c *outputPrepareConfig) { - c.secretCandidateCollector = collector - } -} - // Prepare prepares the output p to be sent to the elastic-agent // The agent might be mutated for an elasticsearch output -func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any, opts ...OutputPrepareOption) error { - cfg := outputPrepareConfig{} - for _, opt := range opts { - opt(&cfg) - } - +func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any) error { span, ctx := apm.StartSpan(ctx, "prepareOutput", "process") defer span.End() span.Context.SetLabel("output_type", p.Type) @@ -93,7 +59,7 @@ func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.B switch p.Type { case OutputTypeElasticsearch: zlog.Debug().Msg("preparing elasticsearch output") - if err := p.prepareElasticsearch(ctx, zlog, bulker, bulker, agent, outputMap, false, cfg.secretCandidateCollector); err != nil { + if err := p.prepareElasticsearch(ctx, zlog, bulker, bulker, agent, outputMap, false); err != nil { return fmt.Errorf("failed to prepare elasticsearch output %q: %w", p.Name, err) } case OutputTypeRemoteElasticsearch: @@ -103,7 +69,7 @@ func (p *Output) Prepare(ctx context.Context, zlog zerolog.Logger, bulker bulk.B return err } // the outputBulker is different for remote ES, it is used to create/update Api keys in the remote ES client - if err := p.prepareElasticsearch(ctx, zlog, bulker, newBulker, agent, outputMap, hasConfigChanged, cfg.secretCandidateCollector); err != nil { + if err := p.prepareElasticsearch(ctx, zlog, bulker, newBulker, agent, outputMap, hasConfigChanged); err != nil { return fmt.Errorf("failed to prepare remote elasticsearch output %q: %w", p.Name, err) } case OutputTypeLogstash: @@ -126,8 +92,7 @@ func (p *Output) prepareElasticsearch( outputBulker bulk.Bulk, agent *model.Agent, outputMap map[string]map[string]any, - hasConfigChanged bool, - secretCandidateCollector OutputSecretCandidateCollector) error { + hasConfigChanged bool) error { // The role is required to do api key management if p.Role == nil { zlog.Error(). diff --git a/internal/pkg/policy/policy_output_integration_test.go b/internal/pkg/policy/policy_output_integration_test.go index 3239ceb43e..9b93f24615 100644 --- a/internal/pkg/policy/policy_output_integration_test.go +++ b/internal/pkg/policy/policy_output_integration_test.go @@ -174,7 +174,7 @@ func TestPolicyOutputESPrepareRealES(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) require.NoError(t, err) // need to wait a bit before querying the agent again @@ -250,7 +250,7 @@ func TestPolicyOutputESPrepareRemoteES(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) require.NoError(t, err) ftesting.Retry(t, ctx, func(ctx context.Context) error { @@ -302,7 +302,7 @@ func TestPolicyOutputESPrepareESRetireRemoteAPIKeys(t *testing.T) { } err = output.prepareElasticsearch( - ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false, nil) + ctx, zerolog.Nop(), bulker, bulker, &agent, policyMap, false) require.NoError(t, err) // need to wait a bit before querying the agent again From 6f9ee1bfec3b0dd95307c98d46842438bea3f3b3 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Mon, 3 Aug 2026 18:01:01 -0700 Subject: [PATCH 5/5] fix: revert test helper referencing removed OutputSecretCandidate type Co-Authored-By: Claude Sonnet 4.6 --- internal/pkg/policy/policy_output_test.go | 9 --------- 1 file changed, 9 deletions(-) diff --git a/internal/pkg/policy/policy_output_test.go b/internal/pkg/policy/policy_output_test.go index e8c561a99a..c71a2f4425 100644 --- a/internal/pkg/policy/policy_output_test.go +++ b/internal/pkg/policy/policy_output_test.go @@ -69,15 +69,6 @@ func TestRenderRemoveOutputPainlessScriptParameterizesOutputName(t *testing.T) { assert.Equal(t, outputName, request.Script.Params["output_name"]) } -type recordingOutputSecretCandidateCollector struct { - candidates []OutputSecretCandidate -} - -func (c *recordingOutputSecretCandidateCollector) Add(candidate OutputSecretCandidate) bool { - c.candidates = append(c.candidates, candidate) - return true -} - func TestPolicyLogstashOutputPrepare(t *testing.T) { logger := testlog.SetLogger(t) bulker := ftesting.NewMockBulk()