[#1731] services/control: Allow to evacuate data from shard

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>

pkg/services/control/convert.go

@@ -184,3 +184,21 @@ func (w *synchronizeTreeResponseWrapper) FromGRPCMessage(m grpc.Message) error {
 	w.SynchronizeTreeResponse = r
 	return nil
 }
+
+type evacuateShardResponseWrapper struct {
+	*EvacuateShardResponse
+}
+
+func (w *evacuateShardResponseWrapper) ToGRPCMessage() grpc.Message {
+	return w.EvacuateShardResponse
+}
+
+func (w *evacuateShardResponseWrapper) FromGRPCMessage(m grpc.Message) error {
+	r, ok := m.(*EvacuateShardResponse)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, (*EvacuateShardResponse)(nil))
+	}
+
+	w.EvacuateShardResponse = r
+	return nil
+}

pkg/services/control/rpc.go

@@ -17,6 +17,7 @@ const (
 	rpcDumpShard       = "DumpShard"
 	rpcRestoreShard    = "RestoreShard"
 	rpcSynchronizeTree = "SynchronizeTree"
+	rpcEvacuateShard   = "EvacuateShard"
 )
 
 // HealthCheck executes ControlService.HealthCheck RPC.
@@ -186,3 +187,16 @@ func SynchronizeTree(cli *client.Client, req *SynchronizeTreeRequest, opts ...cl
 
 	return wResp.SynchronizeTreeResponse, nil
 }
+
+// EvacuateShard executes ControlService.EvacuateShard RPC.
+func EvacuateShard(cli *client.Client, req *EvacuateShardRequest, opts ...client.CallOption) (*EvacuateShardResponse, error) {
+	wResp := &evacuateShardResponseWrapper{new(EvacuateShardResponse)}
+	wReq := &requestWrapper{m: req}
+
+	err := client.SendUnary(cli, common.CallMethodInfoUnary(serviceName, rpcEvacuateShard), wReq, wResp, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return wResp.EvacuateShardResponse, nil
+}

pkg/services/control/server/evacuate.go

@@ -0,0 +1,41 @@
+package control
+
+import (
+	"context"
+
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
+	"github.com/nspcc-dev/neofs-node/pkg/services/control"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func (s *Server) EvacuateShard(_ context.Context, req *control.EvacuateShardRequest) (*control.EvacuateShardResponse, error) {
+	err := s.isValidRequest(req)
+	if err != nil {
+		return nil, status.Error(codes.PermissionDenied, err.Error())
+	}
+
+	shardID := shard.NewIDFromBytes(req.GetBody().GetShard_ID())
+
+	var prm engine.EvacuateShardPrm
+	prm.WithShardID(shardID)
+	prm.WithIgnoreErrors(req.GetBody().GetIgnoreErrors())
+
+	res, err := s.s.Evacuate(prm)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	resp := &control.EvacuateShardResponse{
+		Body: &control.EvacuateShardResponse_Body{
+			Count: uint32(res.Count()),
+		},
+	}
+
+	err = SignMessage(s.key, resp)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	return resp, nil
+}

pkg/services/control/service.proto

@@ -34,6 +34,9 @@ service ControlService {
 
     // Synchronizes all log operations for the specified tree.
     rpc SynchronizeTree (SynchronizeTreeRequest) returns (SynchronizeTreeResponse);
+
+    // EvacuateShard moves all data from one shard to the others.
+    rpc EvacuateShard (EvacuateShardRequest) returns (EvacuateShardResponse);
 }
 
 // Health check request.
@@ -312,3 +315,30 @@ message SynchronizeTreeResponse {
     // Body signature.
     Signature signature = 2;
 }
+
+
+// EvacuateShard request.
+message EvacuateShardRequest {
+    // Request body structure.
+    message Body {
+        // ID of the shard.
+        bytes shard_ID = 1;
+
+        // Flag indicating whether object read errors should be ignored.
+        bool ignore_errors = 2;
+    }
+
+    Body body = 1;
+    Signature signature = 2;
+}
+
+// EvacuateShard response.
+message EvacuateShardResponse {
+    // Response body structure.
+    message Body {
+        uint32 count = 1;
+    }
+
+    Body body = 1;
+    Signature signature = 2;
+}