From e1d0d6284564fd23a10c134be1d4af128ae37523 Mon Sep 17 00:00:00 2001 From: Will Baker Date: Wed, 15 Jul 2026 18:46:01 +0000 Subject: [PATCH] gazctl shards prune: treat missing delete permission as non-fatal Recovery-log pruning is best-effort cleanup of superseded fragments. When a tenant brings their own storage bucket without granting delete permission, every prune fails with AccessDenied on the recovery/ prefix. This hard-errors the command (exit 1) and fires recurring alerts for a bucket we can't clean up and don't otherwise depend on. Classify a failed fragment removal: authorization failures (missing delete permission, 403, bucket-not-found) are now logged and counted in a new non-fatal permissionDenied metric, so the prune still completes successfully. All other removal errors retain the existing failedToRemove behavior and continue to fail the command, so genuine problems keep surfacing. This reuses the per-backend IsAuthError classification already used by fragment.Persist for graceful shutdown, exposed via a new fragment.IsAuthError helper. --- broker/fragment/stores.go | 10 ++++++ broker/fragment/stores_test.go | 47 ++++++++++++++++++++++++++++ cmd/gazctl/gazctlcmd/shards_prune.go | 42 +++++++++++++++---------- 3 files changed, 83 insertions(+), 16 deletions(-) diff --git a/broker/fragment/stores.go b/broker/fragment/stores.go index 0f59d5a1..f09d55c3 100644 --- a/broker/fragment/stores.go +++ b/broker/fragment/stores.go @@ -111,6 +111,16 @@ func Remove(ctx context.Context, fragment pb.Fragment) error { return store.Remove(ctx, fragment.ContentPath()) } +// IsAuthError reports whether `err`, as returned by an operation such as Remove +// or Open against `fragment`, stems from an authorization failure of the +// fragment's BackingStore (for example, a missing delete permission on a +// customer-owned bucket). Callers use this to distinguish "we're not allowed to +// touch this object" from transient or otherwise unexpected failures. +func IsAuthError(fragment pb.Fragment, err error) bool { + var store = stores.Get(fragment.BackingStore) + return store.Store != nil && store.Store.IsAuthError(err) +} + func evalPathPostfix(spool Spool, spec *pb.JournalSpec) (string, error) { var tpl, err = template.New("").Parse(spec.Fragment.PathPostfixTemplate) if err != nil { diff --git a/broker/fragment/stores_test.go b/broker/fragment/stores_test.go index f3e9eced..c272f62a 100644 --- a/broker/fragment/stores_test.go +++ b/broker/fragment/stores_test.go @@ -104,3 +104,50 @@ func TestPersistAuthErrorFallback(t *testing.T) { require.False(t, exists) } } + +func TestIsAuthError(t *testing.T) { + var authError = errors.New("access denied") + var otherError = errors.New("network timeout") + + stores.RegisterProviders(map[string]stores.Constructor{ + // A store which denies deletes with an authorization error. + "s3": func(u *url.URL) (stores.Store, error) { + return &stores.CallbackStore{ + Fallback: stores.NewMemoryStore(u), + RemoveFunc: func(stores.Store, context.Context, string) error { + return authError + }, + IsAuthErrorFunc: func(_ stores.Store, err error) bool { + return err == authError + }, + }, nil + }, + // A store which fails to initialize, so its ActiveStore has a nil Store. + "gs": func(u *url.URL) (stores.Store, error) { + return nil, errors.New("cannot initialize store") + }, + }) + + var frag = pb.Fragment{ + Journal: "test/journal", + Begin: 0, + End: 12, + CompressionCodec: pb.CompressionCodec_NONE, + BackingStore: "s3://bucket/", + } + + // The auth error surfaced by Remove is classified as an auth error... + require.Equal(t, authError, Remove(context.Background(), frag)) + require.True(t, IsAuthError(frag, authError)) + + // ...while an unrelated error against the same store is not. + require.False(t, IsAuthError(frag, otherError)) + + // A store which failed to initialize never reports an auth error (and does not panic). + var uninitFrag = pb.Fragment{ + Journal: "test/journal", + CompressionCodec: pb.CompressionCodec_NONE, + BackingStore: "gs://bucket/", + } + require.False(t, IsAuthError(uninitFrag, authError)) +} diff --git a/cmd/gazctl/gazctlcmd/shards_prune.go b/cmd/gazctl/gazctlcmd/shards_prune.go index 0b98d169..1f054adc 100644 --- a/cmd/gazctl/gazctlcmd/shards_prune.go +++ b/cmd/gazctl/gazctlcmd/shards_prune.go @@ -164,6 +164,14 @@ func (cmd *cmdShardsPrune) Execute([]string) error { defer mu.Unlock() if err != nil { + if fragment.IsAuthError(spec, err) { + log.WithFields(log.Fields{ + "fragment": spec, + "error": err, + }).Warn("skipping fragment removal: no permission to delete") + metrics.permissionDenied++ + return nil + } log.WithFields(log.Fields{ "fragment": spec, "error": err, @@ -275,26 +283,28 @@ func foldHintsIntoSegments(hints recoverylog.FSMHints, sets map[pb.Journal][]rec } type shardsPruneMetrics struct { - shardsTotal int64 - fragmentsTotal int64 - fragmentsPruned int64 - bytesTotal int64 - bytesPruned int64 - skippedJournals int64 - failedToRemove int64 + shardsTotal int64 + fragmentsTotal int64 + fragmentsPruned int64 + bytesTotal int64 + bytesPruned int64 + skippedJournals int64 + failedToRemove int64 + permissionDenied int64 } func logShardsPruneMetrics(m shardsPruneMetrics, journal, message string) { var fields = log.Fields{ - "shardsTotal": m.shardsTotal, - "fragmentsTotal": m.fragmentsTotal, - "fragmentsPruned": m.fragmentsPruned, - "fragmentsKept": m.fragmentsTotal - m.fragmentsPruned, - "bytesTotal": m.bytesTotal, - "bytesPruned": m.bytesPruned, - "bytesKept": m.bytesTotal - m.bytesPruned, - "skippedJournals": m.skippedJournals, - "failedToRemove": m.failedToRemove, + "shardsTotal": m.shardsTotal, + "fragmentsTotal": m.fragmentsTotal, + "fragmentsPruned": m.fragmentsPruned, + "fragmentsKept": m.fragmentsTotal - m.fragmentsPruned, + "bytesTotal": m.bytesTotal, + "bytesPruned": m.bytesPruned, + "bytesKept": m.bytesTotal - m.bytesPruned, + "skippedJournals": m.skippedJournals, + "failedToRemove": m.failedToRemove, + "permissionDenied": m.permissionDenied, } if journal != "" { fields["journal"] = journal