Contrib/volume scrub local (#8226)

weed/pb/volume_server.proto

@@ -643,6 +643,7 @@ enum VolumeScrubMode {
     UNKNOWN = 0;
     INDEX = 1;
     FULL = 2;
+    LOCAL = 3;
 }
 
 message ScrubVolumeRequest {

weed/pb/volume_server_pb/volume_server.pb.go

@@ -28,6 +28,7 @@ const (
 	VolumeScrubMode_UNKNOWN VolumeScrubMode = 0
 	VolumeScrubMode_INDEX   VolumeScrubMode = 1
 	VolumeScrubMode_FULL    VolumeScrubMode = 2
+	VolumeScrubMode_LOCAL   VolumeScrubMode = 3
 )
 
 // Enum value maps for VolumeScrubMode.
@@ -36,11 +37,13 @@ var (
 		0: "UNKNOWN",
 		1: "INDEX",
 		2: "FULL",
+		3: "LOCAL",
 	}
 	VolumeScrubMode_value = map[string]int32{
 		"UNKNOWN": 0,
 		"INDEX":   1,
 		"FULL":    2,
+		"LOCAL":   3,
 	}
 )
 
@@ -7180,11 +7183,12 @@ const file_volume_server_proto_rawDesc = "" +
 	"\rstart_time_ns\x18\x01 \x01(\x03R\vstartTimeNs\x12$\n" +
 	"\x0eremote_time_ns\x18\x02 \x01(\x03R\fremoteTimeNs\x12 \n" +
 	"\fstop_time_ns\x18\x03 \x01(\x03R\n" +
-	"stopTimeNs*3\n" +
+	"stopTimeNs*>\n" +
 	"\x0fVolumeScrubMode\x12\v\n" +
 	"\aUNKNOWN\x10\x00\x12\t\n" +
 	"\x05INDEX\x10\x01\x12\b\n" +
-	"\x04FULL\x10\x022\xfb(\n" +
+	"\x04FULL\x10\x02\x12\t\n" +
+	"\x05LOCAL\x10\x032\xfb(\n" +
 	"\fVolumeServer\x12\\\n" +
 	"\vBatchDelete\x12$.volume_server_pb.BatchDeleteRequest\x1a%.volume_server_pb.BatchDeleteResponse\"\x00\x12n\n" +
 	"\x11VacuumVolumeCheck\x12*.volume_server_pb.VacuumVolumeCheckRequest\x1a+.volume_server_pb.VacuumVolumeCheckResponse\"\x00\x12v\n" +

weed/server/volume_grpc_scrub.go

@@ -36,6 +36,8 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S
 		switch m := req.GetMode(); m {
 		case volume_server_pb.VolumeScrubMode_INDEX:
 			files, serrs = v.CheckIndex()
+		case volume_server_pb.VolumeScrubMode_LOCAL:
+			files, serrs = scrubVolumeLocal(ctx, v)
 		case volume_server_pb.VolumeScrubMode_FULL:
 			files, serrs = scrubVolumeFull(ctx, v)
 		default:
@@ -61,8 +63,12 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S
 	return res, nil
 }
 
+func scrubVolumeLocal(ctx context.Context, v *storage.Volume) (int64, []error) {
+	return 0, []error{fmt.Errorf("scrubVolumeLocal(): not implemented, see https://github.com/seaweedfs/seaweedfs/issues/8018")}
+}
+
 func scrubVolumeFull(ctx context.Context, v *storage.Volume) (int64, []error) {
-	return 0, []error{fmt.Errorf("scrubVolumeFull(): not implemented")}
+	return 0, []error{fmt.Errorf("scrubVolumeFull(): not implemented, see https://github.com/seaweedfs/seaweedfs/issues/8018")}
 }
 
 func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb.ScrubEcVolumeRequest) (*volume_server_pb.ScrubEcVolumeResponse, error) {
@@ -94,6 +100,8 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb
 		case volume_server_pb.VolumeScrubMode_INDEX:
 			// index scrubs do not verify individual EC shards
 			files, serrs = v.CheckIndex()
+		case volume_server_pb.VolumeScrubMode_LOCAL:
+			files, shardInfos, serrs = scrubEcVolumeLocal(ctx, v)
 		case volume_server_pb.VolumeScrubMode_FULL:
 			files, shardInfos, serrs = scrubEcVolumeFull(ctx, v)
 		default:
@@ -121,6 +129,10 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb
 	return res, nil
 }
 
-func scrubEcVolumeFull(ctx context.Context, ecv *erasure_coding.EcVolume) (int64, []*volume_server_pb.EcShardInfo, []error) {
+func scrubEcVolumeLocal(ctx context.Context, v *erasure_coding.EcVolume) (int64, []*volume_server_pb.EcShardInfo, []error) {
-	return 0, nil, []error{fmt.Errorf("scrubEcVolumeFull(): not implemented")}
+	return 0, nil, []error{fmt.Errorf("scrubEcVolumeLocal(): not implemented, see https://github.com/seaweedfs/seaweedfs/issues/8018")}
+}
+
+func scrubEcVolumeFull(ctx context.Context, v *erasure_coding.EcVolume) (int64, []*volume_server_pb.EcShardInfo, []error) {
+	return 0, nil, []error{fmt.Errorf("scrubEcVolumeFull(): not implemented, see https://github.com/seaweedfs/seaweedfs/issues/8018")}
 }

weed/shell/command_ec_scrub.go

@@ -49,7 +49,7 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer
 	nodesStr := volScrubCommand.String("node", "", "comma-separated list of volume server <host>:<port> (optional)")
 	volumeIDsStr := volScrubCommand.String("volumeId", "", "comma-separated EC volume IDs to process (optional)")
 	// TODO: switch default mode to LOCAL, once implemented.
-	mode := volScrubCommand.String("mode", "INDEX", "scrubbing mode (INDEX/FULL)")
+	mode := volScrubCommand.String("mode", "index", "scrubbing mode (index/local/full)")
 	// TODO: add per-node parallelization
 
 	if err = volScrubCommand.Parse(args); err != nil {
@@ -92,6 +92,8 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer
 	switch strings.ToUpper(*mode) {
 	case "INDEX":
 		c.mode = volume_server_pb.VolumeScrubMode_INDEX
+	case "LOCAL":
+		c.mode = volume_server_pb.VolumeScrubMode_LOCAL
 	case "FULL":
 		c.mode = volume_server_pb.VolumeScrubMode_FULL
 	default:

weed/shell/command_volume_scrub.go

@@ -50,7 +50,7 @@ func (c *commandVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer io
 	nodesStr := volScrubCommand.String("node", "", "comma-separated list of volume server <host>:<port> (optional)")
 	volumeIDsStr := volScrubCommand.String("volumeId", "", "comma-separated volume IDs to process (optional)")
 	// TODO: switch default mode to LOCAL, once implemented.
-	mode := volScrubCommand.String("mode", "INDEX", "scrubbing mode (INDEX/FULL)")
+	mode := volScrubCommand.String("mode", "index", "scrubbing mode (index/local/full)")
 	// TODO: add per-node parallelization
 
 	if err = volScrubCommand.Parse(args); err != nil {
@@ -93,6 +93,8 @@ func (c *commandVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer io
 	switch strings.ToUpper(*mode) {
 	case "INDEX":
 		c.mode = volume_server_pb.VolumeScrubMode_INDEX
+	case "LOCAL":
+		c.mode = volume_server_pb.VolumeScrubMode_LOCAL
 	case "FULL":
 		c.mode = volume_server_pb.VolumeScrubMode_FULL
 	default: