feat(access): support gradual repair message routing to shardnode

with: #1000959605

Signed-off-by: xiejian <xiejian3@oppo.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

blobstore/access/stream/stream.go

@@ -160,6 +160,10 @@ type StreamConfig struct {
 	DeleteIntoShardnodePercentage int64 `json:"delete_into_shardnode_percentage"`
 	deleteRoundrobin              int64 `json:"-"`
 
+	// RepairIntoShardnodePercentage roundrobin percentage [1-100]
+	RepairIntoShardnodePercentage int64 `json:"repair_into_shardnode_percentage"`
+	repairRoundrobin              int64 `json:"-"`
+
 	MemPoolSizeClasses map[int]int `json:"mem_pool_size_classes"`
 
 	// CodeModesPutQuorums
@@ -264,6 +268,10 @@ func confCheck(cfg *StreamConfig) error {
 	if cfg.DeleteIntoShardnodePercentage > 100 {
 		cfg.DeleteIntoShardnodePercentage = 100
 	}
+	defaulter.IntegerLess(&cfg.RepairIntoShardnodePercentage, 0)
+	if cfg.RepairIntoShardnodePercentage > 100 {
+		cfg.RepairIntoShardnodePercentage = 100
+	}
 
 	defaulter.LessOrEqual(&cfg.ClusterConfig.CMClientConfig.Config.ClientTimeoutMs, defaultTimeoutClusterMgr)
 	defaulter.LessOrEqual(&cfg.BlobnodeConfig.ClientTimeoutMs, defaultTimeoutBlobnode)
@@ -315,8 +323,9 @@ func NewStreamHandler(cfg *StreamConfig, stopCh <-chan struct{}) (h StreamHandle
 		// Do not use rpc retry, because the stream blob handles retries itself
 		defaulter.LessOrEqual(&cfg.ShardnodeConfig.Config.Retry, int(1))
 		handler.shardnodeClient = shardnode.New(cfg.ShardnodeConfig.Config)
-	} else { // disable write delete msg to shardnode
+	} else { // disable write delete/repair msg to shardnode
 		handler.StreamConfig.DeleteIntoShardnodePercentage = 0
+		handler.StreamConfig.RepairIntoShardnodePercentage = 0
 	}
 
 	if err = clustermgr.LoadExtendCodemode(context.Background(), handler.clusterController); err != nil {
@@ -411,6 +420,15 @@ func (h *Handler) sendRepairMsg(ctx context.Context, blob blobIdent, badIdxes []
 	span := trace.SpanFromContextSafe(ctx)
 	span.Infof("to repair %s indexes(%+v)", blob.String(), badIdxes)
 
+	if h.RepairIntoShardnodePercentage > 0 {
+		percentage := (atomic.AddInt64(&h.repairRoundrobin, 1) % 100) + 1
+		if percentage <= h.RepairIntoShardnodePercentage {
+			span.Debugf("to repair into shardnode %s", blob.String())
+			h.sendRepairMsgIntoShardnode(ctx, blob, badIdxes)
+			return
+		}
+	}
+
 	clusterID := blob.cid
 	serviceController, err := h.clusterController.GetServiceController(clusterID)
 	if err != nil {

blobstore/access/stream/stream_blob.go

@@ -665,6 +665,66 @@ func (h *Handler) fixCreateBlobArgs(ctx context.Context, args *acapi.CreateBlobA
 	return nil
 }
 
+func (h *Handler) getRepairMessageShardnode(ctx context.Context,
+	shardController controller.IShardController, clusterID proto.ClusterID, blob blobIdent, badIdxes []uint8,
+) (args shardnode.RepairSliceArgs, host string, err error) {
+	args.Vid = blob.vid
+	args.Bid = blob.bid
+	args.Reason = "access-repair"
+	for _, idx := range badIdxes {
+		args.BadIdx = append(args.BadIdx, uint32(idx))
+	}
+	tagNum := shardController.GetShardSubRangeCount(ctx)
+	var shard controller.Shard
+	shard, err = shardController.GetShard(ctx, args.GetShardKeys(tagNum))
+	if err != nil {
+		return
+	}
+	args.Header, err = h.getOpHeaderByShard(ctx, shardController, shard, acapi.GetShardModeLeader)
+	if err != nil {
+		return
+	}
+	host, err = h.getShardHost(ctx, clusterID, args.Header.DiskID)
+	return
+}
+
+func (h *Handler) sendRepairMsgIntoShardnode(ctx context.Context, blob blobIdent, badIdxes []uint8) {
+	span := trace.SpanFromContextSafe(ctx)
+	clusterID := blob.cid
+	shardController, err := h.clusterController.GetShardController(clusterID)
+	if err != nil {
+		span.Error(errors.Detail(err))
+		return
+	}
+
+	if err := retry.Timed(3, 100).On(func() error {
+		args, host, err := h.getRepairMessageShardnode(ctx, shardController, clusterID, blob, badIdxes)
+		if err != nil {
+			reportUnhealth(clusterID, "repair.msg", serviceShard, "-", "failed")
+			span.Warn(err)
+			return err
+		}
+		if err = h.shardnodeClient.RepairSlice(ctx, host, args); err != nil {
+			span.Warnf("send to shardnode %s repair message(%+v) %s", host, args, err.Error())
+			reportUnhealth(clusterID, "repair.msg", serviceShard, host, "failed")
+			_, err = h.punishAndUpdate(ctx, &punishArgs{
+				ShardOpHeader: args.Header,
+				clusterID:     clusterID,
+				host:          host,
+				mode:          acapi.GetShardModeLeader,
+				err:           err,
+			})
+			err = errors.Base(err, host)
+		}
+		return err
+	}); err != nil {
+		span.Errorf("send shardnode repair message(%+v) failed %s", blob, errors.Detail(err))
+		return
+	}
+
+	span.Infof("send shardnode repair message(%+v)", blob)
+}
+
 func (h *Handler) getDeleteMessageShardnode(ctx context.Context,
 	shardController controller.IShardController, clusterID proto.ClusterID, slice proto.Slice,
 ) (args shardnode.DeleteBlobRawArgs, host string, err error) {

blobstore/api/shardnode/client.go

@@ -45,6 +45,7 @@ type AccessAPI interface {
 	AllocSlice(ctx context.Context, host string, args AllocSliceArgs) (ret AllocSliceRet, err error)
 	FindAndDeleteBlob(ctx context.Context, host string, args DeleteBlobArgs) (ret GetBlobRet, err error)
 	DeleteBlobRaw(ctx context.Context, host string, args DeleteBlobRawArgs) error
+	RepairSlice(ctx context.Context, host string, args RepairSliceArgs) error
 	GetShardStats(ctx context.Context, host string, args GetShardArgs) (ret ShardStats, err error)
 }
 
@@ -97,3 +98,7 @@ func (c *FakeClient) FindAndDeleteBlob(ctx context.Context, host string, args De
 func (c *FakeClient) DeleteBlobRaw(ctx context.Context, host string, args DeleteBlobRawArgs) error {
 	return errcode.ErrShardNodeUnsupport
 }
+
+func (c *FakeClient) RepairSlice(ctx context.Context, host string, args RepairSliceArgs) error {
+	return errcode.ErrShardNodeUnsupport
+}

blobstore/testing/mocks/api_shardnode.go

@@ -153,6 +153,20 @@ func (mr *MockShardnodeAccessMockRecorder) ListBlob(arg0, arg1, arg2 interface{}
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListBlob", reflect.TypeOf((*MockShardnodeAccess)(nil).ListBlob), arg0, arg1, arg2)
 }
 
+// RepairSlice mocks base method.
+func (m *MockShardnodeAccess) RepairSlice(arg0 context.Context, arg1 string, arg2 shardnode.RepairSliceArgs) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RepairSlice", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RepairSlice indicates an expected call of RepairSlice.
+func (mr *MockShardnodeAccessMockRecorder) RepairSlice(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RepairSlice", reflect.TypeOf((*MockShardnodeAccess)(nil).RepairSlice), arg0, arg1, arg2)
+}
+
 // SealBlob mocks base method.
 func (m *MockShardnodeAccess) SealBlob(arg0 context.Context, arg1 string, arg2 shardnode.SealBlobArgs) error {
 	m.ctrl.T.Helper()