forked from TrueCloudLab/frostfs-node
[#1086] services/control: implement DumpShard
RPC
Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
parent
aa53418119
commit
0e60b1d6c9
9 changed files with 294 additions and 0 deletions
19
pkg/local_object_storage/engine/dump.go
Normal file
19
pkg/local_object_storage/engine/dump.go
Normal file
|
@ -0,0 +1,19 @@
|
||||||
|
package engine
|
||||||
|
|
||||||
|
import "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
||||||
|
|
||||||
|
// DumpShard dumps objects from the shard with provided identifier.
|
||||||
|
//
|
||||||
|
// Returns an error if shard is not read-only.
|
||||||
|
func (e *StorageEngine) DumpShard(id *shard.ID, prm *shard.DumpPrm) error {
|
||||||
|
e.mtx.RLock()
|
||||||
|
defer e.mtx.RUnlock()
|
||||||
|
|
||||||
|
sh, ok := e.shards[id.String()]
|
||||||
|
if !ok {
|
||||||
|
return errShardNotFound
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := sh.Dump(prm)
|
||||||
|
return err
|
||||||
|
}
|
|
@ -130,3 +130,21 @@ func (w *setShardModeResponseWrapper) FromGRPCMessage(m grpc.Message) error {
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type dumpShardResponseWrapper struct {
|
||||||
|
*DumpShardResponse
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *dumpShardResponseWrapper) ToGRPCMessage() grpc.Message {
|
||||||
|
return w.DumpShardResponse
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *dumpShardResponseWrapper) FromGRPCMessage(m grpc.Message) error {
|
||||||
|
r, ok := m.(*DumpShardResponse)
|
||||||
|
if !ok {
|
||||||
|
return message.NewUnexpectedMessageType(m, (*DumpShardResponse)(nil))
|
||||||
|
}
|
||||||
|
|
||||||
|
w.DumpShardResponse = r
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
|
@ -14,6 +14,7 @@ const (
|
||||||
rpcDropObjects = "DropObjects"
|
rpcDropObjects = "DropObjects"
|
||||||
rpcListShards = "ListShards"
|
rpcListShards = "ListShards"
|
||||||
rpcSetShardMode = "SetShardMode"
|
rpcSetShardMode = "SetShardMode"
|
||||||
|
rpcDumpShard = "DumpShard"
|
||||||
)
|
)
|
||||||
|
|
||||||
// HealthCheck executes ControlService.HealthCheck RPC.
|
// HealthCheck executes ControlService.HealthCheck RPC.
|
||||||
|
@ -144,3 +145,16 @@ func SetShardMode(
|
||||||
|
|
||||||
return wResp.m, nil
|
return wResp.m, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DumpShard executes ControlService.DumpShard RPC.
|
||||||
|
func DumpShard(cli *client.Client, req *DumpShardRequest, opts ...client.CallOption) (*DumpShardResponse, error) {
|
||||||
|
wResp := new(dumpShardResponseWrapper)
|
||||||
|
wReq := &requestWrapper{m: req}
|
||||||
|
|
||||||
|
err := client.SendUnary(cli, common.CallMethodInfoUnary(serviceName, rpcDumpShard), wReq, wResp, opts...)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return wResp.DumpShardResponse, nil
|
||||||
|
}
|
||||||
|
|
37
pkg/services/control/server/dump.go
Normal file
37
pkg/services/control/server/dump.go
Normal file
|
@ -0,0 +1,37 @@
|
||||||
|
package control
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
|
||||||
|
"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) DumpShard(_ context.Context, req *control.DumpShardRequest) (*control.DumpShardResponse, error) {
|
||||||
|
err := s.isValidRequest(req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, status.Error(codes.PermissionDenied, err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
shardID := shard.NewIDFromBytes(req.GetBody().GetShard_ID())
|
||||||
|
|
||||||
|
prm := new(shard.DumpPrm)
|
||||||
|
prm.WithPath(req.GetBody().GetFilepath())
|
||||||
|
prm.WithIgnoreErrors(req.GetBody().GetIgnoreErrors())
|
||||||
|
|
||||||
|
err = s.s.DumpShard(shardID, prm)
|
||||||
|
if err != nil {
|
||||||
|
return nil, status.Error(codes.Internal, err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
resp := new(control.DumpShardResponse)
|
||||||
|
resp.SetBody(new(control.DumpShardResponse_Body))
|
||||||
|
|
||||||
|
err = SignMessage(s.key, resp)
|
||||||
|
if err != nil {
|
||||||
|
return nil, status.Error(codes.Internal, err.Error())
|
||||||
|
}
|
||||||
|
return resp, nil
|
||||||
|
}
|
|
@ -910,3 +910,172 @@ func (x *SetShardModeResponse) ReadSignedData(buf []byte) ([]byte, error) {
|
||||||
func (x *SetShardModeResponse) SignedDataSize() int {
|
func (x *SetShardModeResponse) SignedDataSize() int {
|
||||||
return x.GetBody().StableSize()
|
return x.GetBody().StableSize()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetShardID sets shard ID for the dump shard request.
|
||||||
|
func (x *DumpShardRequest_Body) SetShardID(id []byte) {
|
||||||
|
x.Shard_ID = id
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetFilepath sets filepath for the dump shard request.
|
||||||
|
func (x *DumpShardRequest_Body) SetFilepath(p string) {
|
||||||
|
x.Filepath = p
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetIgnoreErrors sets ignore errors flag for the dump shard request.
|
||||||
|
func (x *DumpShardRequest_Body) SetIgnoreErrors(ignore bool) {
|
||||||
|
x.IgnoreErrors = ignore
|
||||||
|
}
|
||||||
|
|
||||||
|
const (
|
||||||
|
_ = iota
|
||||||
|
dumpShardReqBodyShardIDFNum
|
||||||
|
dumpShardReqBodyFilepathFNum
|
||||||
|
dumpShardReqBodyIgnoreErrorsFNum
|
||||||
|
)
|
||||||
|
|
||||||
|
// StableMarshal reads binary representation of request body binary format.
|
||||||
|
//
|
||||||
|
// If buffer length is less than StableSize(), new buffer is allocated.
|
||||||
|
//
|
||||||
|
// Returns any error encountered which did not allow writing the data completely.
|
||||||
|
// Otherwise, returns the buffer in which the data is written.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same binary format.
|
||||||
|
func (x *DumpShardRequest_Body) StableMarshal(buf []byte) ([]byte, error) {
|
||||||
|
if x == nil {
|
||||||
|
return []byte{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if sz := x.StableSize(); len(buf) < sz {
|
||||||
|
buf = make([]byte, sz)
|
||||||
|
}
|
||||||
|
|
||||||
|
var (
|
||||||
|
offset, n int
|
||||||
|
err error
|
||||||
|
)
|
||||||
|
|
||||||
|
n, err = proto.BytesMarshal(dumpShardReqBodyShardIDFNum, buf, x.Shard_ID)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
offset += n
|
||||||
|
|
||||||
|
n, err = proto.StringMarshal(dumpShardReqBodyFilepathFNum, buf[offset:], x.Filepath)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
offset += n
|
||||||
|
|
||||||
|
_, err = proto.BoolMarshal(dumpShardReqBodyIgnoreErrorsFNum, buf[offset:], x.IgnoreErrors)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return buf, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// StableSize returns binary size of the request body in protobuf binary format.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same binary size.
|
||||||
|
func (x *DumpShardRequest_Body) StableSize() int {
|
||||||
|
if x == nil {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
|
||||||
|
size := 0
|
||||||
|
|
||||||
|
size += proto.BytesSize(dumpShardReqBodyShardIDFNum, x.Shard_ID)
|
||||||
|
size += proto.StringSize(dumpShardReqBodyFilepathFNum, x.Filepath)
|
||||||
|
size += proto.BoolSize(dumpShardReqBodyIgnoreErrorsFNum, x.IgnoreErrors)
|
||||||
|
|
||||||
|
return size
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetBody sets request body.
|
||||||
|
func (x *DumpShardRequest) SetBody(v *DumpShardRequest_Body) {
|
||||||
|
if x != nil {
|
||||||
|
x.Body = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetSignature sets body signature of the request.
|
||||||
|
func (x *DumpShardRequest) SetSignature(v *Signature) {
|
||||||
|
if x != nil {
|
||||||
|
x.Signature = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReadSignedData reads signed data from request to buf.
|
||||||
|
//
|
||||||
|
// If buffer length is less than x.SignedDataSize(), new buffer is allocated.
|
||||||
|
//
|
||||||
|
// Returns any error encountered which did not allow writing the data completely.
|
||||||
|
// Otherwise, returns the buffer in which the data is written.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same signed data.
|
||||||
|
func (x *DumpShardRequest) ReadSignedData(buf []byte) ([]byte, error) {
|
||||||
|
return x.GetBody().StableMarshal(buf)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SignedDataSize returns size of the request signed data in bytes.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same signed data size.
|
||||||
|
func (x *DumpShardRequest) SignedDataSize() int {
|
||||||
|
return x.GetBody().StableSize()
|
||||||
|
}
|
||||||
|
|
||||||
|
// StableMarshal reads binary representation of the response body in protobuf binary format.
|
||||||
|
//
|
||||||
|
// If buffer length is less than x.StableSize(), new buffer is allocated.
|
||||||
|
//
|
||||||
|
// Returns any error encountered which did not allow writing the data completely.
|
||||||
|
// Otherwise, returns the buffer in which the data is written.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same binary format.
|
||||||
|
func (x *DumpShardResponse_Body) StableMarshal(buf []byte) ([]byte, error) {
|
||||||
|
return buf, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// StableSize returns binary size of the response body
|
||||||
|
// in protobuf binary format.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same binary size.
|
||||||
|
func (x *DumpShardResponse_Body) StableSize() int {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetBody sets response body.
|
||||||
|
func (x *DumpShardResponse) SetBody(v *DumpShardResponse_Body) {
|
||||||
|
if x != nil {
|
||||||
|
x.Body = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetSignature sets response body signature.
|
||||||
|
func (x *DumpShardResponse) SetSignature(v *Signature) {
|
||||||
|
if x != nil {
|
||||||
|
x.Signature = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReadSignedData reads signed data from response to buf.
|
||||||
|
//
|
||||||
|
// If buffer length is less than SignedDataSize(), new buffer is allocated.
|
||||||
|
//
|
||||||
|
// Returns any error encountered which did not allow writing the data completely.
|
||||||
|
// Otherwise, returns the buffer in which the data is written.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same signed data.
|
||||||
|
func (x *DumpShardResponse) ReadSignedData(buf []byte) ([]byte, error) {
|
||||||
|
return x.GetBody().StableMarshal(buf)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SignedDataSize returns binary size of the signed data.
|
||||||
|
//
|
||||||
|
// Structures with the same field values have the same signed data size.
|
||||||
|
func (x *DumpShardResponse) SignedDataSize() int {
|
||||||
|
return x.GetBody().StableSize()
|
||||||
|
}
|
||||||
|
|
BIN
pkg/services/control/service.pb.go
generated
BIN
pkg/services/control/service.pb.go
generated
Binary file not shown.
|
@ -25,6 +25,9 @@ service ControlService {
|
||||||
|
|
||||||
// Sets mode of the shard.
|
// Sets mode of the shard.
|
||||||
rpc SetShardMode (SetShardModeRequest) returns (SetShardModeResponse);
|
rpc SetShardMode (SetShardModeRequest) returns (SetShardModeResponse);
|
||||||
|
|
||||||
|
// Dump objects from the shard.
|
||||||
|
rpc DumpShard (DumpShardRequest) returns (DumpShardResponse);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Health check request.
|
// Health check request.
|
||||||
|
@ -201,3 +204,37 @@ message SetShardModeResponse {
|
||||||
// Body signature.
|
// Body signature.
|
||||||
Signature signature = 2;
|
Signature signature = 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DumpShard request.
|
||||||
|
message DumpShardRequest {
|
||||||
|
// Request body structure.
|
||||||
|
message Body {
|
||||||
|
// ID of the shard.
|
||||||
|
bytes shard_ID = 1;
|
||||||
|
|
||||||
|
// Path to the output.
|
||||||
|
string filepath = 2;
|
||||||
|
|
||||||
|
// Flag indicating whether object read errors should be ignored.
|
||||||
|
bool ignore_errors = 3;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Body of dump shard request message.
|
||||||
|
Body body = 1;
|
||||||
|
|
||||||
|
// Body signature.
|
||||||
|
Signature signature = 2;
|
||||||
|
}
|
||||||
|
|
||||||
|
// DumpShard response.
|
||||||
|
message DumpShardResponse {
|
||||||
|
// Response body structure.
|
||||||
|
message Body {
|
||||||
|
}
|
||||||
|
|
||||||
|
// Body of dump shard response message.
|
||||||
|
Body body = 1;
|
||||||
|
|
||||||
|
// Body signature.
|
||||||
|
Signature signature = 2;
|
||||||
|
}
|
||||||
|
|
BIN
pkg/services/control/service_grpc.pb.go
generated
BIN
pkg/services/control/service_grpc.pb.go
generated
Binary file not shown.
BIN
pkg/services/control/types.pb.go
generated
BIN
pkg/services/control/types.pb.go
generated
Binary file not shown.
Loading…
Reference in a new issue