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