[#1806] services/control: Allow to flush write-cache

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

pkg/services/control/convert.go

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

pkg/services/control/rpc.go

@@ -17,6 +17,7 @@ const (
 	rpcRestoreShard    = "RestoreShard"
 	rpcSynchronizeTree = "SynchronizeTree"
 	rpcEvacuateShard   = "EvacuateShard"
+	rpcFlushCache      = "FlushCache"
 )
 
 // HealthCheck executes ControlService.HealthCheck RPC.
@@ -177,3 +178,16 @@ func EvacuateShard(cli *client.Client, req *EvacuateShardRequest, opts ...client
 
 	return wResp.EvacuateShardResponse, nil
 }
+
+// FlushCache executes ControlService.FlushCache RPC.
+func FlushCache(cli *client.Client, req *FlushCacheRequest, opts ...client.CallOption) (*FlushCacheResponse, error) {
+	wResp := &flushCacheResponseWrapper{new(FlushCacheResponse)}
+	wReq := &requestWrapper{m: req}
+
+	err := client.SendUnary(cli, common.CallMethodInfoUnary(serviceName, rpcFlushCache), wReq, wResp, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return wResp.FlushCacheResponse, nil
+}

pkg/services/control/server/flush_cache.go

@@ -0,0 +1,36 @@
+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) FlushCache(_ context.Context, req *control.FlushCacheRequest) (*control.FlushCacheResponse, 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.FlushWriteCachePrm
+	prm.SetShardID(shardID)
+
+	_, err = s.s.FlushWriteCache(prm)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	resp := &control.FlushCacheResponse{Body: &control.FlushCacheResponse_Body{}}
+
+	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 {
 
     // EvacuateShard moves all data from one shard to the others.
     rpc EvacuateShard (EvacuateShardRequest) returns (EvacuateShardResponse);
+
+    // FlushCache moves all data from one shard to the others.
+    rpc FlushCache (FlushCacheRequest) returns (FlushCacheResponse);
 }
 
 // Health check request.
@@ -311,3 +314,25 @@ message EvacuateShardResponse {
     Body body = 1;
     Signature signature = 2;
 }
+
+// FlushCache request.
+message FlushCacheRequest {
+    // Request body structure.
+    message Body {
+        // ID of the shard.
+        bytes shard_ID = 1;
+    }
+
+    Body body = 1;
+    Signature signature = 2;
+}
+
+// FlushCache response.
+message FlushCacheResponse {
+    // Response body structure.
+    message Body {
+    }
+
+    Body body = 1;
+    Signature signature = 2;
+}