diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index da1b941bd..deb34228d 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -708,6 +708,7 @@ message ScrubEcVolumeRequest { VolumeScrubMode mode = 1; // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. repeated uint32 volume_ids = 2; + bool force_deleted_needles_check = 3; // FULL mode only; may report false positives when EC indexes disagree } message ScrubEcVolumeResponse { uint64 total_volumes = 1; diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 889b4dc73..abf9da1cf 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -711,6 +711,7 @@ message ScrubEcVolumeRequest { VolumeScrubMode mode = 1; // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. repeated uint32 volume_ids = 2; + bool force_deleted_needles_check = 3; // FULL mode only; may report false positives when EC indexes disagree } message ScrubEcVolumeResponse { uint64 total_volumes = 1; diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index d02b35bb1..0bc292e23 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -6018,9 +6018,10 @@ type ScrubEcVolumeRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Mode VolumeScrubMode `protobuf:"varint,1,opt,name=mode,proto3,enum=volume_server_pb.VolumeScrubMode" json:"mode,omitempty"` // optional list of volume IDs to scrub. if empty, all EC volumes for the server are scrubbed. - VolumeIds []uint32 `protobuf:"varint,2,rep,packed,name=volume_ids,json=volumeIds,proto3" json:"volume_ids,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + VolumeIds []uint32 `protobuf:"varint,2,rep,packed,name=volume_ids,json=volumeIds,proto3" json:"volume_ids,omitempty"` + ForceDeletedNeedlesCheck bool `protobuf:"varint,3,opt,name=force_deleted_needles_check,json=forceDeletedNeedlesCheck,proto3" json:"force_deleted_needles_check,omitempty"` // FULL mode only; may report false positives when EC indexes disagree + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ScrubEcVolumeRequest) Reset() { @@ -6067,6 +6068,13 @@ func (x *ScrubEcVolumeRequest) GetVolumeIds() []uint32 { return nil } +func (x *ScrubEcVolumeRequest) GetForceDeletedNeedlesCheck() bool { + if x != nil { + return x.ForceDeletedNeedlesCheck + } + return false +} + type ScrubEcVolumeResponse struct { state protoimpl.MessageState `protogen:"open.v1"` TotalVolumes uint64 `protobuf:"varint,1,opt,name=total_volumes,json=totalVolumes,proto3" json:"total_volumes,omitempty"` @@ -7510,11 +7518,12 @@ const file_volume_server_proto_rawDesc = "" + "\vtotal_files\x18\x02 \x01(\x04R\n" + "totalFiles\x12*\n" + "\x11broken_volume_ids\x18\x03 \x03(\rR\x0fbrokenVolumeIds\x12\x18\n" + - "\adetails\x18\x04 \x03(\tR\adetails\"l\n" + + "\adetails\x18\x04 \x03(\tR\adetails\"\xab\x01\n" + "\x14ScrubEcVolumeRequest\x125\n" + "\x04mode\x18\x01 \x01(\x0e2!.volume_server_pb.VolumeScrubModeR\x04mode\x12\x1d\n" + "\n" + - "volume_ids\x18\x02 \x03(\rR\tvolumeIds\"\xf0\x01\n" + + "volume_ids\x18\x02 \x03(\rR\tvolumeIds\x12=\n" + + "\x1bforce_deleted_needles_check\x18\x03 \x01(\bR\x18forceDeletedNeedlesCheck\"\xf0\x01\n" + "\x15ScrubEcVolumeResponse\x12#\n" + "\rtotal_volumes\x18\x01 \x01(\x04R\ftotalVolumes\x12\x1f\n" + "\vtotal_files\x18\x02 \x01(\x04R\n" + diff --git a/weed/server/volume_grpc_scrub.go b/weed/server/volume_grpc_scrub.go index 3b967e35c..412f0bbd2 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -93,6 +93,13 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S } func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb.ScrubEcVolumeRequest) (*volume_server_pb.ScrubEcVolumeResponse, error) { + if err := vs.checkGrpcAdminAuth(ctx); err != nil { + return nil, err + } + if req.GetForceDeletedNeedlesCheck() && req.GetMode() != volume_server_pb.VolumeScrubMode_FULL { + return nil, fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + } + vids := []needle.VolumeId{} if len(req.GetVolumeIds()) == 0 { for _, l := range vs.store.Locations { @@ -124,8 +131,7 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb case volume_server_pb.VolumeScrubMode_LOCAL: files, shardInfos, serrs = v.ScrubLocal() case volume_server_pb.VolumeScrubMode_FULL: - // TODO: expose force_deleted_needles_check in the RPC and weed shell. - files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, false) + files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, req.GetForceDeletedNeedlesCheck()) case volume_server_pb.VolumeScrubMode_CHECKSUM: // Verify each local shard's raw bytes against the bitrot sidecar, // exercising cold parity shards. Read-only. ChecksumScrub's first diff --git a/weed/shell/command_ec_scrub.go b/weed/shell/command_ec_scrub.go index e9bd36f6e..13b6897ab 100644 --- a/weed/shell/command_ec_scrub.go +++ b/weed/shell/command_ec_scrub.go @@ -20,11 +20,12 @@ func init() { } type commandEcVolumeScrub struct { - env *CommandEnv - volumeServerAddrs []pb.ServerAddress - volumeIDs []uint32 - mode volume_server_pb.VolumeScrubMode - grpcDialOption grpc.DialOption + env *CommandEnv + volumeServerAddrs []pb.ServerAddress + volumeIDs []uint32 + mode volume_server_pb.VolumeScrubMode + forceDeletedNeedlesCheck bool + grpcDialOption grpc.DialOption } func (c *commandEcVolumeScrub) Name() string { @@ -52,6 +53,7 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer mode := volScrubCommand.String("mode", "local", "scrubbing mode (index/local/full/checksum)") maxParallelization := volScrubCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") showDetails := volScrubCommand.Bool("details", false, "display scrub result details, if available") + forceDeletedNeedlesCheck := volScrubCommand.Bool("forceDeletedNeedlesCheck", false, "force strict verification of deleted needles (full mode only); may report false positives when EC indexes disagree") if err = volScrubCommand.Parse(args); err != nil { return err @@ -104,6 +106,10 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer } fmt.Fprintf(writer, "using %s mode\n", c.mode.String()) c.env = commandEnv + c.forceDeletedNeedlesCheck = *forceDeletedNeedlesCheck + if c.forceDeletedNeedlesCheck && c.mode != volume_server_pb.VolumeScrubMode_FULL { + return fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + } return c.scrubEcVolumes(writer, *maxParallelization, *showDetails) } @@ -125,8 +131,9 @@ func (c *commandEcVolumeScrub) scrubEcVolumes(writer io.Writer, maxParallelizati err := operation.WithVolumeServerClient(false, addr, c.env.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { res, err := volumeServerClient.ScrubEcVolume(context.Background(), &volume_server_pb.ScrubEcVolumeRequest{ - Mode: c.mode, - VolumeIds: c.volumeIDs, + Mode: c.mode, + VolumeIds: c.volumeIDs, + ForceDeletedNeedlesCheck: c.forceDeletedNeedlesCheck, }) if err != nil { return err