shards_prune: skip unhealthy stores when pruning shards

Author: mdibaiee

broker/stores/health_check.go

@@ -62,6 +62,7 @@ func checkLoop(stores map[pb.FragmentStore]*ActiveStore, fs pb.FragmentStore, at
 			"error":    err,
 			"attempt":  attempt,
 			"interval": checkInterval,
+			"authError": active.IsAuthError(err),
 		}).Warn("store health check failed")
 		attempt += 1
 	}

broker/stores/health_check_test.go

@@ -61,6 +61,9 @@ func TestHealthChecks(t *testing.T) {
 							require.Equal(t, ".test/health-check", path)
 							return server.URL + "/signed/test", nil
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},
@@ -77,6 +80,9 @@ func TestHealthChecks(t *testing.T) {
 						GetFunc: func(_ Store, ctx context.Context, path string) (io.ReadCloser, error) {
 							return nil, errors.New("simulated GET failure")
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},
@@ -99,6 +105,9 @@ func TestHealthChecks(t *testing.T) {
 						SignGetFunc: func(_ Store, path string, d time.Duration) (string, error) {
 							return server.URL + "/wrong", nil
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},
@@ -115,6 +124,9 @@ func TestHealthChecks(t *testing.T) {
 						GetFunc: func(_ Store, ctx context.Context, path string) (io.ReadCloser, error) {
 							return nil, nil // Return nil reader without error
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},
@@ -132,6 +144,9 @@ func TestHealthChecks(t *testing.T) {
 							// Return a reader that fails on Read
 							return io.NopCloser(errorReader{}), nil
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},
@@ -160,6 +175,9 @@ func TestHealthChecks(t *testing.T) {
 						SignGetFunc: func(_ Store, path string, d time.Duration) (string, error) {
 							return server.URL + "/signed/test", nil
 						},
+						IsAuthErrorFunc: func(_ Store, err error) bool {
+							return false
+						},
 					}, nil
 				}
 			},

cmd/gazctl/gazctlcmd/shards_prune.go

@@ -55,12 +55,20 @@ func (cmd *cmdShardsPrune) Execute([]string) error {
 	for _, shard := range listShards(rsc, cmd.Selector).Shards {
 		metrics.shardsTotal++
 
+		var recoveryLog = shard.Spec.RecoveryLog()
+
+		// Check if the recovery log's fragment stores are healthy before fetching hints.
+		// Skip this shard if any store is unhealthy to avoid hanging.
+		if !checkRecoveryLogStoresHealth(ctx, rjc, recoveryLog) {
+			skipRecoveryLogs[recoveryLog] = true
+			continue
+		}
+
 		var allHints, err = consumer.FetchHints(ctx, rsc, &pc.GetHintsRequest{
 			Shard: shard.Spec.Id,
 		})
 		mbp.Must(err, "failed to fetch hints")
 
-		var recoveryLog = shard.Spec.RecoveryLog()
 		rawHintsResponses[recoveryLog] = append(rawHintsResponses[recoveryLog], *allHints)
 
 		for _, curHints := range append(allHints.BackupHints, allHints.PrimaryHints) {
@@ -94,7 +102,7 @@ func (cmd *cmdShardsPrune) Execute([]string) error {
 
 	for journal, segments := range logSegmentSets {
 		if skipRecoveryLogs[journal] {
-			log.WithField("journal", journal).Warn("skipping journal because a shard is missing hints that cover it")
+			log.WithField("journal", journal).Warn("skipping journal because a shard is missing hints that cover it or has unlhealthy store")
 			continue
 		}
 		log.WithField("journal", journal).Debug("checking fragments of journal")
@@ -162,6 +170,48 @@ func (cmd *cmdShardsPrune) Execute([]string) error {
 	return nil
 }
 
+// checkRecoveryLogStoresHealth checks if all fragment stores for the given recovery log journal are healthy.
+// Returns true if all stores are healthy, false otherwise.
+func checkRecoveryLogStoresHealth(ctx context.Context, jc pb.JournalClient, recoveryLog pb.Journal) bool {
+	var journalSpec, err = client.GetJournal(ctx, jc, recoveryLog)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"journal": recoveryLog,
+			"error":   err,
+		}).Warn("failed to fetch journal spec for recovery log")
+		return false
+	}
+
+	if len(journalSpec.Fragment.Stores) == 0 {
+		// No stores configured, consider healthy
+		return true
+	}
+
+	for _, store := range journalSpec.Fragment.Stores {
+		var resp, err = client.FragmentStoreHealth(ctx, jc, store)
+		if err != nil {
+			log.WithFields(log.Fields{
+				"journal": recoveryLog,
+				"store":   store,
+				"error":   err,
+			}).Warn("failed to check fragment store health")
+			return false
+		}
+
+		if resp.Status != pb.Status_OK {
+			log.WithFields(log.Fields{
+				"journal":    recoveryLog,
+				"store":      store,
+				"status":     resp.Status,
+				"storeError": resp.StoreHealthError,
+			}).Warn("fragment store is unhealthy")
+			return false
+		}
+	}
+
+	return true
+}
+
 func overlapsAnySegment(segments []recoverylog.Segment, fragment pb.Fragment) bool {
 	for _, seg := range segments {
 		if (seg.FirstOffset < fragment.End || fragment.End == 0) &&