From 7f1cad6faf43405d498daad04546e5e200ce7d69 Mon Sep 17 00:00:00 2001 From: paul cannon Date: Mon, 19 Sep 2022 16:16:48 -0500 Subject: [PATCH] satellite/repair: better handling of piece fetch errors We have an alert on `repair_too_many_nodes_failed` which fires too frequently. Every time so far, it has been because of a network blip of some nature on the satellite side. Satellite operators are expected to have other means in place for alerting on network problems and fixing them, so it's not necessary for the repair framework to act in that way. Instead, in this change, we change the way that `repair_too_many_nodes_failed` works. When a repair fails, we collect piece fetch errors by type and determine from them whether it looks like we are having network problems (most errors are connection failures, possibly also some successful connections which subsequently time out) or whether something else has happened. We will now only emit `repair_too_many_nodes_failed` when the outcome does not look like a network failure. In the network failure case, we will instead emit `repair_suspected_network_problem`. Refs: https://github.com/storj/storj/issues/4669 Change-Id: I49df98da5df9c606b95ad08a2bdfec8092fba926 --- monkit.lock | 1 + satellite/repair/repair_test.go | 28 +++---- satellite/repair/repairer/ec.go | 28 +++---- satellite/repair/repairer/segments.go | 113 +++++++++++++++++++++----- 4 files changed, 117 insertions(+), 53 deletions(-) diff --git a/monkit.lock b/monkit.lock index 6dfcee397..989b9e834 100644 --- a/monkit.lock +++ b/monkit.lock @@ -127,6 +127,7 @@ storj.io/storj/satellite/repair/repairer."repair_segment_pieces_successful" IntV storj.io/storj/satellite/repair/repairer."repair_segment_pieces_total" IntVal storj.io/storj/satellite/repair/repairer."repair_segment_size" IntVal storj.io/storj/satellite/repair/repairer."repair_success" Meter +storj.io/storj/satellite/repair/repairer."repair_suspected_network_problem" Meter storj.io/storj/satellite/repair/repairer."repair_too_many_nodes_failed" Meter storj.io/storj/satellite/repair/repairer."repair_unnecessary" Meter storj.io/storj/satellite/repair/repairer."repairer_segments_below_min_req" Counter diff --git a/satellite/repair/repair_test.go b/satellite/repair/repair_test.go index 8baf079f9..c0011cb3e 100644 --- a/satellite/repair/repair_test.go +++ b/satellite/repair/repair_test.go @@ -1018,7 +1018,7 @@ func TestMissingPieceDataRepair(t *testing.T) { nodesReputation[piece.StorageNode] = *info } - var successful metabase.Pieces + var successful []repairer.PieceFetchResult satellite.Repairer.SegmentRepairer.OnTestingPiecesReportHook = func(pieces repairer.FetchResultReport) { successful = pieces.Successful } @@ -1040,8 +1040,8 @@ func TestMissingPieceDataRepair(t *testing.T) { // repair shouldn't update audit status for _, result := range successful { - successfulNodeReputation := nodesReputation[result.StorageNode] - successfulNodeReputationAfter := nodesReputationAfter[result.StorageNode] + successfulNodeReputation := nodesReputation[result.Piece.StorageNode] + successfulNodeReputationAfter := nodesReputationAfter[result.Piece.StorageNode] require.Equal(t, successfulNodeReputation.TotalAuditCount, successfulNodeReputationAfter.TotalAuditCount) require.Equal(t, successfulNodeReputation.AuditSuccessCount, successfulNodeReputationAfter.AuditSuccessCount) require.Equal(t, successfulNodeReputation.AuditReputationAlpha, successfulNodeReputationAfter.AuditReputationAlpha) @@ -1248,7 +1248,7 @@ func TestCorruptDataRepair_Failed(t *testing.T) { nodesReputation[piece.StorageNode] = *info } - var successful metabase.Pieces + var successful []repairer.PieceFetchResult satellite.Repairer.SegmentRepairer.OnTestingPiecesReportHook = func(report repairer.FetchResultReport) { successful = report.Successful } @@ -1270,8 +1270,8 @@ func TestCorruptDataRepair_Failed(t *testing.T) { // repair shouldn't update audit status for _, result := range successful { - successfulNodeReputation := nodesReputation[result.StorageNode] - successfulNodeReputationAfter := nodesReputationAfter[result.StorageNode] + successfulNodeReputation := nodesReputation[result.Piece.StorageNode] + successfulNodeReputationAfter := nodesReputationAfter[result.Piece.StorageNode] require.Equal(t, successfulNodeReputation.TotalAuditCount, successfulNodeReputationAfter.TotalAuditCount) require.Equal(t, successfulNodeReputation.AuditSuccessCount, successfulNodeReputationAfter.AuditSuccessCount) require.Equal(t, successfulNodeReputation.AuditReputationAlpha, successfulNodeReputationAfter.AuditReputationAlpha) @@ -2547,7 +2547,7 @@ func TestECRepairerGetCorrupted(t *testing.T) { require.Equal(t, 0, len(piecesReport.Contained)) require.Equal(t, 0, len(piecesReport.Unknown)) require.Equal(t, int(segment.Redundancy.RequiredShares), len(piecesReport.Successful)) - require.Equal(t, corruptedPiece, piecesReport.Failed[0]) + require.Equal(t, corruptedPiece, piecesReport.Failed[0].Piece) }) } @@ -2616,7 +2616,7 @@ func TestECRepairerGetMissingPiece(t *testing.T) { require.Equal(t, 0, len(piecesReport.Contained)) require.Equal(t, 0, len(piecesReport.Unknown)) require.Equal(t, int(segment.Redundancy.RequiredShares), len(piecesReport.Successful)) - require.Equal(t, missingPiece, piecesReport.Failed[0]) + require.Equal(t, missingPiece, piecesReport.Failed[0].Piece) }) } @@ -2682,7 +2682,7 @@ func TestECRepairerGetOffline(t *testing.T) { require.Equal(t, 0, len(piecesReport.Contained)) require.Equal(t, 0, len(piecesReport.Unknown)) require.Equal(t, int(segment.Redundancy.RequiredShares), len(piecesReport.Successful)) - require.Equal(t, offlinePiece, piecesReport.Offline[0]) + require.Equal(t, offlinePiece, piecesReport.Offline[0].Piece) }) } @@ -2752,7 +2752,7 @@ func TestECRepairerGetUnknown(t *testing.T) { require.Equal(t, 0, len(piecesReport.Contained)) require.Equal(t, 1, len(piecesReport.Unknown)) require.Equal(t, int(segment.Redundancy.RequiredShares), len(piecesReport.Successful)) - require.Equal(t, unknownPiece, piecesReport.Unknown[0]) + require.Equal(t, unknownPiece, piecesReport.Unknown[0].Piece) }) } @@ -2837,10 +2837,10 @@ func TestECRepairerGetFailure(t *testing.T) { require.Equal(t, 0, len(piecesReport.Contained)) require.Equal(t, 1, len(piecesReport.Unknown)) require.Equal(t, 1, len(piecesReport.Successful)) - require.Equal(t, offlinePiece, piecesReport.Offline[0]) - require.Equal(t, corruptedPiece, piecesReport.Failed[0]) - require.Equal(t, unknownPiece, piecesReport.Unknown[0]) - require.Equal(t, successfulPiece, piecesReport.Successful[0]) + require.Equal(t, offlinePiece, piecesReport.Offline[0].Piece) + require.Equal(t, corruptedPiece, piecesReport.Failed[0].Piece) + require.Equal(t, unknownPiece, piecesReport.Unknown[0].Piece) + require.Equal(t, successfulPiece, piecesReport.Successful[0].Piece) }) } diff --git a/satellite/repair/repairer/ec.go b/satellite/repair/repairer/ec.go index 2b5047e5c..ea3581407 100644 --- a/satellite/repair/repairer/ec.go +++ b/satellite/repair/repairer/ec.go @@ -7,7 +7,6 @@ import ( "bytes" "context" "errors" - "fmt" "hash" "io" "io/ioutil" @@ -95,9 +94,6 @@ func (ec *ECRepairer) Get(ctx context.Context, limits []*pb.AddressedOrderLimit, limiter := sync2.NewLimiter(es.RequiredCount()) cond := sync.NewCond(&sync.Mutex{}) - var errlist errs.Group - var mu sync.Mutex - for currentLimitIndex, limit := range limits { if limit == nil { continue @@ -165,49 +161,46 @@ func (ec *ECRepairer) Get(ctx context.Context, limits []*pb.AddressedOrderLimit, zap.Stringer("node ID", limit.GetLimit().StorageNodeId), zap.Stringer("Piece ID", limit.Limit.PieceId), zap.String("reason", err.Error())) - pieces.Failed = append(pieces.Failed, piece) + pieces.Failed = append(pieces.Failed, PieceFetchResult{Piece: piece, Err: err}) return } pieceAudit := audit.PieceAuditFromErr(err) switch pieceAudit { case audit.PieceAuditFailure: - ec.log.Debug("Failed to download pieces for repair: piece not found (audit failed)", + ec.log.Debug("Failed to download piece for repair: piece not found (audit failed)", zap.Stringer("Node ID", limit.GetLimit().StorageNodeId), zap.Stringer("Piece ID", limit.Limit.PieceId), zap.Error(err)) - pieces.Failed = append(pieces.Failed, piece) + pieces.Failed = append(pieces.Failed, PieceFetchResult{Piece: piece, Err: err}) case audit.PieceAuditOffline: - ec.log.Debug("Failed to download pieces for repair: dial timeout (offline)", + ec.log.Debug("Failed to download piece for repair: dial timeout (offline)", zap.Stringer("Node ID", limit.GetLimit().StorageNodeId), zap.Stringer("Piece ID", limit.Limit.PieceId), zap.Error(err)) - pieces.Offline = append(pieces.Offline, piece) + pieces.Offline = append(pieces.Offline, PieceFetchResult{Piece: piece, Err: err}) case audit.PieceAuditContained: - ec.log.Info("Failed to download pieces for repair: download timeout (contained)", + ec.log.Info("Failed to download piece for repair: download timeout (contained)", zap.Stringer("Node ID", limit.GetLimit().StorageNodeId), zap.Stringer("Piece ID", limit.Limit.PieceId), zap.Error(err)) - pieces.Contained = append(pieces.Contained, piece) + pieces.Contained = append(pieces.Contained, PieceFetchResult{Piece: piece, Err: err}) case audit.PieceAuditUnknown: - ec.log.Info("Failed to download pieces for repair: unknown transport error (skipped)", + ec.log.Info("Failed to download piece for repair: unknown transport error (skipped)", zap.Stringer("Node ID", limit.GetLimit().StorageNodeId), zap.Stringer("Piece ID", limit.Limit.PieceId), zap.Error(err)) - pieces.Unknown = append(pieces.Unknown, piece) + pieces.Unknown = append(pieces.Unknown, PieceFetchResult{Piece: piece, Err: err}) } - mu.Lock() - errlist.Add(fmt.Errorf("node id: %s, error: %w", limit.GetLimit().StorageNodeId.String(), err)) - mu.Unlock() return } pieceReaders[currentLimitIndex] = pieceReadCloser - pieces.Successful = append(pieces.Successful, piece) + pieces.Successful = append(pieces.Successful, PieceFetchResult{Piece: piece}) successfulPieces++ return } @@ -221,7 +214,6 @@ func (ec *ECRepairer) Get(ctx context.Context, limits []*pb.AddressedOrderLimit, return nil, pieces, &irreparableError{ piecesAvailable: int32(successfulPieces), piecesRequired: int32(es.RequiredCount()), - errlist: errlist, } } diff --git a/satellite/repair/repairer/segments.go b/satellite/repair/repairer/segments.go index 563f70c8a..042ac8ac3 100644 --- a/satellite/repair/repairer/segments.go +++ b/satellite/repair/repairer/segments.go @@ -9,6 +9,7 @@ import ( "fmt" "io" "math" + "strings" "time" "github.com/zeebo/errs" @@ -49,21 +50,27 @@ var ( type irreparableError struct { piecesAvailable int32 piecesRequired int32 - errlist []error } func (ie *irreparableError) Error() string { return fmt.Sprintf("%d available pieces < %d required", ie.piecesAvailable, ie.piecesRequired) } +// PieceFetchResult combines a piece pointer with the error we got when we tried +// to acquire that piece. +type PieceFetchResult struct { + Piece metabase.Piece + Err error +} + // FetchResultReport contains a categorization of a set of pieces based on the results of // GET operations. type FetchResultReport struct { - Successful metabase.Pieces - Failed metabase.Pieces - Offline metabase.Pieces - Contained metabase.Pieces - Unknown metabase.Pieces + Successful []PieceFetchResult + Failed []PieceFetchResult + Offline []PieceFetchResult + Contained []PieceFetchResult + Unknown []PieceFetchResult } // SegmentRepairer for segments. @@ -152,6 +159,8 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue // ignore segment if expired if segment.Expired(repairer.nowFn()) { + mon.Meter("repair_unnecessary").Mark(1) + mon.Meter("segment_expired_before_repair").Mark(1) repairer.log.Debug("segment has expired", zap.Stringer("Stream ID", segment.StreamID), zap.Uint64("Position", queueSegment.Position.Encode())) return true, nil } @@ -183,7 +192,7 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue } numHealthy := len(pieces) - len(missingPieces) - // irreparable piece + // irreparable segment if numHealthy < int(segment.Redundancy.RequiredShares) { mon.Counter("repairer_segments_below_min_req").Inc(1) //mon:locked stats.repairerSegmentsBelowMinReq.Inc(1) @@ -260,9 +269,11 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue mon.Meter("repair_nodes_unavailable").Mark(1) //mon:locked stats.repairerNodesUnavailable.Mark(1) - repairer.log.Warn("irreparable segment", + repairer.log.Warn("irreparable segment: too many nodes offline", zap.String("StreamID", queueSegment.StreamID.String()), zap.Uint64("Position", queueSegment.Position.Encode()), + zap.Int("piecesAvailable", len(healthyPieces)), + zap.Int16("piecesRequired", segment.Redundancy.RequiredShares), zap.Error(err), ) } @@ -309,7 +320,8 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue segmentReader, piecesReport, err := repairer.ec.Get(ctx, getOrderLimits, cachedNodesInfo, getPrivateKey, redundancy, int64(segment.EncryptedSize)) // ensure we get values, even if only zero values, so that redash can have an alert based on this - mon.Meter("repair_too_many_nodes_failed").Mark(0) //mon:locked + mon.Meter("repair_too_many_nodes_failed").Mark(0) //mon:locked + mon.Meter("repair_suspected_network_problem").Mark(0) //mon:locked stats.repairTooManyNodesFailed.Mark(0) if repairer.OnTestingPiecesReportHook != nil { @@ -348,16 +360,65 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue // to wait for nodes to come back online. var irreparableErr *irreparableError if errors.As(err, &irreparableErr) { - mon.Meter("repair_too_many_nodes_failed").Mark(1) //mon:locked + // piecesReport.Offline: + // Nodes which were online recently, but which we couldn't contact for + // this operation. + // + // piecesReport.Failed: + // Nodes which we contacted successfully but which indicated they + // didn't have the piece we wanted. + // + // piecesReport.Contained: + // Nodes which we contacted successfully but timed out after we asked + // for the piece. + // + // piecesReport.Unknown: + // Something else went wrong, and we don't know what. + // + // In a network failure scenario, we expect more than half of the outcomes + // will be in Offline or Contained. + if len(piecesReport.Offline)+len(piecesReport.Contained) > len(piecesReport.Successful)+len(piecesReport.Failed)+len(piecesReport.Unknown) { + mon.Meter("repair_suspected_network_problem").Mark(1) //mon:locked + } else { + mon.Meter("repair_too_many_nodes_failed").Mark(1) //mon:locked + } stats.repairTooManyNodesFailed.Mark(1) - repairer.log.Warn("irreparable segment", + failedNodeIDs := make([]string, 0, len(piecesReport.Failed)) + offlineNodeIDs := make([]string, 0, len(piecesReport.Offline)) + timedOutNodeIDs := make([]string, 0, len(piecesReport.Contained)) + unknownErrs := make([]string, 0, len(piecesReport.Unknown)) + for _, outcome := range piecesReport.Failed { + failedNodeIDs = append(failedNodeIDs, outcome.Piece.StorageNode.String()) + } + for _, outcome := range piecesReport.Offline { + offlineNodeIDs = append(offlineNodeIDs, outcome.Piece.StorageNode.String()) + } + for _, outcome := range piecesReport.Contained { + timedOutNodeIDs = append(timedOutNodeIDs, outcome.Piece.StorageNode.String()) + } + for _, outcome := range piecesReport.Unknown { + // We are purposefully using the error's string here, as opposed + // to wrapping the error. It is not likely that we need the local-side + // traceback of where this error was initially wrapped, and this will + // keep the logs more readable. + unknownErrs = append(unknownErrs, fmt.Sprintf("node ID [%s] err: %v", outcome.Piece.StorageNode, outcome.Err)) + } + + repairer.log.Warn("irreparable segment: could not acquire enough shares", zap.String("StreamID", queueSegment.StreamID.String()), zap.Uint64("Position", queueSegment.Position.Encode()), zap.Int32("piecesAvailable", irreparableErr.piecesAvailable), zap.Int32("piecesRequired", irreparableErr.piecesRequired), - zap.Error(errs.Combine(irreparableErr.errlist...)), + zap.Int("numFailedNodes", len(failedNodeIDs)), + zap.Stringer("failedNodes", commaSeparatedArray(failedNodeIDs)), + zap.Int("numOfflineNodes", len(offlineNodeIDs)), + zap.Stringer("offlineNodes", commaSeparatedArray(offlineNodeIDs)), + zap.Int("numTimedOutNodes", len(timedOutNodeIDs)), + zap.Stringer("timedOutNodes", commaSeparatedArray(timedOutNodeIDs)), + zap.Stringer("unknownErrors", commaSeparatedArray(unknownErrs)), ) + // repair will be attempted again if the segment remains unhealthy. return false, nil } // The segment's redundancy strategy is invalid, or else there was an internal error. @@ -375,17 +436,17 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue NodesReputation: cachedNodesReputation, } - for _, piece := range piecesReport.Successful { - report.Successes = append(report.Successes, piece.StorageNode) + for _, outcome := range piecesReport.Successful { + report.Successes = append(report.Successes, outcome.Piece.StorageNode) } - for _, piece := range piecesReport.Failed { - report.Fails = append(report.Fails, piece.StorageNode) + for _, outcome := range piecesReport.Failed { + report.Fails = append(report.Fails, outcome.Piece.StorageNode) } - for _, piece := range piecesReport.Offline { - report.Offlines = append(report.Offlines, piece.StorageNode) + for _, outcome := range piecesReport.Offline { + report.Offlines = append(report.Offlines, outcome.Piece.StorageNode) } - for _, piece := range piecesReport.Unknown { - report.Unknown = append(report.Unknown, piece.StorageNode) + for _, outcome := range piecesReport.Unknown { + report.Unknown = append(report.Unknown, outcome.Piece.StorageNode) } _, reportErr := repairer.reporter.RecordAudits(ctx, report) if reportErr != nil { @@ -462,7 +523,9 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment *queue } // add pieces that failed piece hashes verification to the removal list - toRemove = append(toRemove, piecesReport.Failed...) + for _, outcome := range piecesReport.Failed { + toRemove = append(toRemove, outcome.Piece) + } newPieces, err := segment.Pieces.Update(repairedPieces, toRemove) if err != nil { @@ -636,3 +699,11 @@ func sliceToSet(slice []uint16) map[uint16]bool { } return set } + +// commaSeparatedArray concatenates an array into a comma-separated string, +// lazily. +type commaSeparatedArray []string + +func (c commaSeparatedArray) String() string { + return strings.Join(c, ", ") +}