gazctl shards prune: treat missing delete permission as non-fatal

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 {

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))
+}

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