From 9e31cb249f10c06cd0908db1269f600f4bd51cda Mon Sep 17 00:00:00 2001
From: Ekaterina Lebedeva <ekaterina.lebedeva@yadro.com>
Date: Tue, 4 Feb 2025 21:21:31 +0300
Subject: [PATCH] [#1635] control: Add method to search shards by object
Added method `ListShardsForObject` to ControlService and to
StorageEngine. It returns information about shards storing
object on the node.
Signed-off-by: Ekaterina Lebedeva <ekaterina.lebedeva@yadro.com>
---
internal/logs/logs.go | 1 +
pkg/local_object_storage/engine/shards.go | 46 ++
pkg/services/control/rpc.go | 20 +
.../control/server/list_shards_for_object.go | 66 ++
pkg/services/control/service.proto | 23 +
pkg/services/control/service_frostfs.pb.go | 724 ++++++++++++++++++
pkg/services/control/service_grpc.pb.go | 39 +
7 files changed, 919 insertions(+)
create mode 100644 pkg/services/control/server/list_shards_for_object.go
diff --git a/internal/logs/logs.go b/internal/logs/logs.go
index d48a4da9..d07f47fb 100644
--- a/internal/logs/logs.go
+++ b/internal/logs/logs.go
@@ -252,6 +252,7 @@ const (
ShardFailureToMarkLockersAsGarbage = "failure to mark lockers as garbage"
ShardFailureToGetExpiredUnlockedObjects = "failure to get expired unlocked objects"
ShardCouldNotMarkObjectToDeleteInMetabase = "could not mark object to delete in metabase"
+ ShardCouldNotFindObject = "could not find object"
WritecacheWaitingForChannelsToFlush = "waiting for channels to flush"
WritecacheCantRemoveObjectFromWritecache = "can't remove object from write-cache"
BlobovniczatreeCouldNotGetObjectFromLevel = "could not get object from level"
diff --git a/pkg/local_object_storage/engine/shards.go b/pkg/local_object_storage/engine/shards.go
index 8e191f72..28f0287b 100644
--- a/pkg/local_object_storage/engine/shards.go
+++ b/pkg/local_object_storage/engine/shards.go
@@ -11,6 +11,9 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
+ "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
+ apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
+ objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/hrw"
"github.com/google/uuid"
@@ -442,3 +445,46 @@ func (e *StorageEngine) deleteShards(ctx context.Context, ids []*shard.ID) ([]ha
func (s hashedShard) Hash() uint64 {
return s.hash
}
+
+func (e *StorageEngine) ListShardsForObject(ctx context.Context, obj oid.Address) ([]shard.Info, error) {
+ var err error
+ var info []shard.Info
+ prm := shard.ExistsPrm{
+ Address: obj,
+ }
+ var siErr *objectSDK.SplitInfoError
+ var ecErr *objectSDK.ECInfoError
+
+ e.iterateOverUnsortedShards(func(hs hashedShard) (stop bool) {
+ res, exErr := hs.Exists(ctx, prm)
+ if exErr != nil {
+ if client.IsErrObjectAlreadyRemoved(exErr) {
+ err = new(apistatus.ObjectAlreadyRemoved)
+ return true
+ }
+
+ // Check if error is either SplitInfoError or ECInfoError.
+ // True means the object is virtual.
+ if errors.As(exErr, &siErr) || errors.As(exErr, &ecErr) {
+ info = append(info, hs.DumpInfo())
+ return false
+ }
+
+ if shard.IsErrObjectExpired(exErr) {
+ err = exErr
+ return true
+ }
+
+ if !client.IsErrObjectNotFound(exErr) {
+ e.reportShardError(ctx, hs, "could not check existence of object in shard", exErr, zap.Stringer("address", prm.Address))
+ }
+
+ return false
+ }
+ if res.Exists() {
+ info = append(info, hs.DumpInfo())
+ }
+ return false
+ })
+ return info, err
+}
diff --git a/pkg/services/control/rpc.go b/pkg/services/control/rpc.go
index bbf2cf0c..0c4236d0 100644
--- a/pkg/services/control/rpc.go
+++ b/pkg/services/control/rpc.go
@@ -32,6 +32,7 @@ const (
rpcListTargetsLocalOverrides = "ListTargetsLocalOverrides"
rpcDetachShards = "DetachShards"
rpcStartShardRebuild = "StartShardRebuild"
+ rpcListShardsForObject = "ListShardsForObject"
)
// HealthCheck executes ControlService.HealthCheck RPC.
@@ -364,3 +365,22 @@ func StartShardRebuild(cli *client.Client, req *StartShardRebuildRequest, opts .
return wResp.message, nil
}
+
+// ListShardsForObject executes ControlService.ListShardsForObject RPC.
+func ListShardsForObject(
+ cli *client.Client,
+ req *ListShardsForObjectRequest,
+ opts ...client.CallOption,
+) (*ListShardsForObjectResponse, error) {
+ wResp := newResponseWrapper[ListShardsForObjectResponse]()
+
+ wReq := &requestWrapper{
+ m: req,
+ }
+ err := client.SendUnary(cli, common.CallMethodInfoUnary(serviceName, rpcListShardsForObject), wReq, wResp, opts...)
+ if err != nil {
+ return nil, err
+ }
+
+ return wResp.message, nil
+}
diff --git a/pkg/services/control/server/list_shards_for_object.go b/pkg/services/control/server/list_shards_for_object.go
new file mode 100644
index 00000000..84469772
--- /dev/null
+++ b/pkg/services/control/server/list_shards_for_object.go
@@ -0,0 +1,66 @@
+package control
+
+import (
+ "context"
+
+ "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
+ "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
+ "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/control"
+ "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/control/server/ctrlmessage"
+
+ cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
+ oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+func (s *Server) ListShardsForObject(ctx context.Context, req *control.ListShardsForObjectRequest) (*control.ListShardsForObjectResponse, error) {
+ err := s.isValidRequest(req)
+ if err != nil {
+ return nil, status.Error(codes.PermissionDenied, err.Error())
+ }
+
+ var obj oid.ID
+ err = obj.DecodeString(req.GetBody().GetObjectId())
+ if err != nil {
+ return nil, status.Error(codes.InvalidArgument, err.Error())
+ }
+
+ var cnr cid.ID
+ err = cnr.DecodeString(req.GetBody().GetContainerId())
+ if err != nil {
+ return nil, status.Error(codes.InvalidArgument, err.Error())
+ }
+
+ resp := new(control.ListShardsForObjectResponse)
+ body := new(control.ListShardsForObjectResponse_Body)
+ resp.SetBody(body)
+
+ var objAddr oid.Address
+ objAddr.SetContainer(cnr)
+ objAddr.SetObject(obj)
+ info, err := s.s.ListShardsForObject(ctx, objAddr)
+ if err != nil {
+ return nil, status.Error(codes.Internal, err.Error())
+ }
+ if len(info) == 0 {
+ return nil, status.Error(codes.NotFound, logs.ShardCouldNotFindObject)
+ }
+
+ body.SetShard_ID(shardInfoToProto(info))
+
+ // Sign the response
+ if err := ctrlmessage.Sign(s.key, resp); err != nil {
+ return nil, status.Error(codes.Internal, err.Error())
+ }
+ return resp, nil
+}
+
+func shardInfoToProto(infos []shard.Info) [][]byte {
+ shardInfos := make([][]byte, 0, len(infos))
+ for _, info := range infos {
+ shardInfos = append(shardInfos, *info.ID)
+ }
+
+ return shardInfos
+}
diff --git a/pkg/services/control/service.proto b/pkg/services/control/service.proto
index 97ecf9a8..4c539acf 100644
--- a/pkg/services/control/service.proto
+++ b/pkg/services/control/service.proto
@@ -89,6 +89,9 @@ service ControlService {
// StartShardRebuild starts shard rebuild process.
rpc StartShardRebuild(StartShardRebuildRequest) returns (StartShardRebuildResponse);
+
+ // ListShardsForObject returns shard info where object is stored.
+ rpc ListShardsForObject(ListShardsForObjectRequest) returns (ListShardsForObjectResponse);
}
// Health check request.
@@ -729,3 +732,23 @@ message StartShardRebuildResponse {
Signature signature = 2;
}
+
+message ListShardsForObjectRequest {
+ message Body {
+ string object_id = 1;
+ string container_id = 2;
+ }
+
+ Body body = 1;
+ Signature signature = 2;
+}
+
+message ListShardsForObjectResponse {
+ message Body {
+ // List of the node's shards storing object.
+ repeated bytes shard_ID = 1;
+ }
+
+ Body body = 1;
+ Signature signature = 2;
+}
diff --git a/pkg/services/control/service_frostfs.pb.go b/pkg/services/control/service_frostfs.pb.go
index 0b4e3cf3..44849d59 100644
--- a/pkg/services/control/service_frostfs.pb.go
+++ b/pkg/services/control/service_frostfs.pb.go
@@ -17303,3 +17303,727 @@ func (x *StartShardRebuildResponse) UnmarshalEasyJSON(in *jlexer.Lexer) {
in.Consumed()
}
}
+
+type ListShardsForObjectRequest_Body struct {
+ ObjectId string `json:"objectId"`
+ ContainerId string `json:"containerId"`
+}
+
+var (
+ _ encoding.ProtoMarshaler = (*ListShardsForObjectRequest_Body)(nil)
+ _ encoding.ProtoUnmarshaler = (*ListShardsForObjectRequest_Body)(nil)
+ _ json.Marshaler = (*ListShardsForObjectRequest_Body)(nil)
+ _ json.Unmarshaler = (*ListShardsForObjectRequest_Body)(nil)
+)
+
+// StableSize returns the size of x in protobuf format.
+//
+// Structures with the same field values have the same binary size.
+func (x *ListShardsForObjectRequest_Body) StableSize() (size int) {
+ if x == nil {
+ return 0
+ }
+ size += proto.StringSize(1, x.ObjectId)
+ size += proto.StringSize(2, x.ContainerId)
+ return size
+}
+
+// MarshalProtobuf implements the encoding.ProtoMarshaler interface.
+func (x *ListShardsForObjectRequest_Body) MarshalProtobuf(dst []byte) []byte {
+ m := pool.MarshalerPool.Get()
+ defer pool.MarshalerPool.Put(m)
+ x.EmitProtobuf(m.MessageMarshaler())
+ dst = m.Marshal(dst)
+ return dst
+}
+
+func (x *ListShardsForObjectRequest_Body) EmitProtobuf(mm *easyproto.MessageMarshaler) {
+ if x == nil {
+ return
+ }
+ if len(x.ObjectId) != 0 {
+ mm.AppendString(1, x.ObjectId)
+ }
+ if len(x.ContainerId) != 0 {
+ mm.AppendString(2, x.ContainerId)
+ }
+}
+
+// UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
+func (x *ListShardsForObjectRequest_Body) UnmarshalProtobuf(src []byte) (err error) {
+ var fc easyproto.FieldContext
+ for len(src) > 0 {
+ src, err = fc.NextField(src)
+ if err != nil {
+ return fmt.Errorf("cannot read next field in %s", "ListShardsForObjectRequest_Body")
+ }
+ switch fc.FieldNum {
+ case 1: // ObjectId
+ data, ok := fc.String()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "ObjectId")
+ }
+ x.ObjectId = data
+ case 2: // ContainerId
+ data, ok := fc.String()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "ContainerId")
+ }
+ x.ContainerId = data
+ }
+ }
+ return nil
+}
+func (x *ListShardsForObjectRequest_Body) GetObjectId() string {
+ if x != nil {
+ return x.ObjectId
+ }
+ return ""
+}
+func (x *ListShardsForObjectRequest_Body) SetObjectId(v string) {
+ x.ObjectId = v
+}
+func (x *ListShardsForObjectRequest_Body) GetContainerId() string {
+ if x != nil {
+ return x.ContainerId
+ }
+ return ""
+}
+func (x *ListShardsForObjectRequest_Body) SetContainerId(v string) {
+ x.ContainerId = v
+}
+
+// MarshalJSON implements the json.Marshaler interface.
+func (x *ListShardsForObjectRequest_Body) MarshalJSON() ([]byte, error) {
+ w := jwriter.Writer{}
+ x.MarshalEasyJSON(&w)
+ return w.Buffer.BuildBytes(), w.Error
+}
+func (x *ListShardsForObjectRequest_Body) MarshalEasyJSON(out *jwriter.Writer) {
+ if x == nil {
+ out.RawString("null")
+ return
+ }
+ first := true
+ out.RawByte('{')
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"objectId\":"
+ out.RawString(prefix)
+ out.String(x.ObjectId)
+ }
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"containerId\":"
+ out.RawString(prefix)
+ out.String(x.ContainerId)
+ }
+ out.RawByte('}')
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface.
+func (x *ListShardsForObjectRequest_Body) UnmarshalJSON(data []byte) error {
+ r := jlexer.Lexer{Data: data}
+ x.UnmarshalEasyJSON(&r)
+ return r.Error()
+}
+func (x *ListShardsForObjectRequest_Body) UnmarshalEasyJSON(in *jlexer.Lexer) {
+ isTopLevel := in.IsStart()
+ if in.IsNull() {
+ if isTopLevel {
+ in.Consumed()
+ }
+ in.Skip()
+ return
+ }
+ in.Delim('{')
+ for !in.IsDelim('}') {
+ key := in.UnsafeFieldName(false)
+ in.WantColon()
+ if in.IsNull() {
+ in.Skip()
+ in.WantComma()
+ continue
+ }
+ switch key {
+ case "objectId":
+ {
+ var f string
+ f = in.String()
+ x.ObjectId = f
+ }
+ case "containerId":
+ {
+ var f string
+ f = in.String()
+ x.ContainerId = f
+ }
+ }
+ in.WantComma()
+ }
+ in.Delim('}')
+ if isTopLevel {
+ in.Consumed()
+ }
+}
+
+type ListShardsForObjectRequest struct {
+ Body *ListShardsForObjectRequest_Body `json:"body"`
+ Signature *Signature `json:"signature"`
+}
+
+var (
+ _ encoding.ProtoMarshaler = (*ListShardsForObjectRequest)(nil)
+ _ encoding.ProtoUnmarshaler = (*ListShardsForObjectRequest)(nil)
+ _ json.Marshaler = (*ListShardsForObjectRequest)(nil)
+ _ json.Unmarshaler = (*ListShardsForObjectRequest)(nil)
+)
+
+// StableSize returns the size of x in protobuf format.
+//
+// Structures with the same field values have the same binary size.
+func (x *ListShardsForObjectRequest) StableSize() (size int) {
+ if x == nil {
+ return 0
+ }
+ size += proto.NestedStructureSize(1, x.Body)
+ size += proto.NestedStructureSize(2, x.Signature)
+ return size
+}
+
+// ReadSignedData fills buf with signed data of x.
+// 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 *ListShardsForObjectRequest) SignedDataSize() int {
+ return x.GetBody().StableSize()
+}
+
+// SignedDataSize returns size of the request signed data in bytes.
+//
+// Structures with the same field values have the same signed data size.
+func (x *ListShardsForObjectRequest) ReadSignedData(buf []byte) ([]byte, error) {
+ return x.GetBody().MarshalProtobuf(buf), nil
+}
+
+// MarshalProtobuf implements the encoding.ProtoMarshaler interface.
+func (x *ListShardsForObjectRequest) MarshalProtobuf(dst []byte) []byte {
+ m := pool.MarshalerPool.Get()
+ defer pool.MarshalerPool.Put(m)
+ x.EmitProtobuf(m.MessageMarshaler())
+ dst = m.Marshal(dst)
+ return dst
+}
+
+func (x *ListShardsForObjectRequest) EmitProtobuf(mm *easyproto.MessageMarshaler) {
+ if x == nil {
+ return
+ }
+ if x.Body != nil {
+ x.Body.EmitProtobuf(mm.AppendMessage(1))
+ }
+ if x.Signature != nil {
+ x.Signature.EmitProtobuf(mm.AppendMessage(2))
+ }
+}
+
+// UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
+func (x *ListShardsForObjectRequest) UnmarshalProtobuf(src []byte) (err error) {
+ var fc easyproto.FieldContext
+ for len(src) > 0 {
+ src, err = fc.NextField(src)
+ if err != nil {
+ return fmt.Errorf("cannot read next field in %s", "ListShardsForObjectRequest")
+ }
+ switch fc.FieldNum {
+ case 1: // Body
+ data, ok := fc.MessageData()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "Body")
+ }
+ x.Body = new(ListShardsForObjectRequest_Body)
+ if err := x.Body.UnmarshalProtobuf(data); err != nil {
+ return fmt.Errorf("unmarshal: %w", err)
+ }
+ case 2: // Signature
+ data, ok := fc.MessageData()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "Signature")
+ }
+ x.Signature = new(Signature)
+ if err := x.Signature.UnmarshalProtobuf(data); err != nil {
+ return fmt.Errorf("unmarshal: %w", err)
+ }
+ }
+ }
+ return nil
+}
+func (x *ListShardsForObjectRequest) GetBody() *ListShardsForObjectRequest_Body {
+ if x != nil {
+ return x.Body
+ }
+ return nil
+}
+func (x *ListShardsForObjectRequest) SetBody(v *ListShardsForObjectRequest_Body) {
+ x.Body = v
+}
+func (x *ListShardsForObjectRequest) GetSignature() *Signature {
+ if x != nil {
+ return x.Signature
+ }
+ return nil
+}
+func (x *ListShardsForObjectRequest) SetSignature(v *Signature) {
+ x.Signature = v
+}
+
+// MarshalJSON implements the json.Marshaler interface.
+func (x *ListShardsForObjectRequest) MarshalJSON() ([]byte, error) {
+ w := jwriter.Writer{}
+ x.MarshalEasyJSON(&w)
+ return w.Buffer.BuildBytes(), w.Error
+}
+func (x *ListShardsForObjectRequest) MarshalEasyJSON(out *jwriter.Writer) {
+ if x == nil {
+ out.RawString("null")
+ return
+ }
+ first := true
+ out.RawByte('{')
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"body\":"
+ out.RawString(prefix)
+ x.Body.MarshalEasyJSON(out)
+ }
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"signature\":"
+ out.RawString(prefix)
+ x.Signature.MarshalEasyJSON(out)
+ }
+ out.RawByte('}')
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface.
+func (x *ListShardsForObjectRequest) UnmarshalJSON(data []byte) error {
+ r := jlexer.Lexer{Data: data}
+ x.UnmarshalEasyJSON(&r)
+ return r.Error()
+}
+func (x *ListShardsForObjectRequest) UnmarshalEasyJSON(in *jlexer.Lexer) {
+ isTopLevel := in.IsStart()
+ if in.IsNull() {
+ if isTopLevel {
+ in.Consumed()
+ }
+ in.Skip()
+ return
+ }
+ in.Delim('{')
+ for !in.IsDelim('}') {
+ key := in.UnsafeFieldName(false)
+ in.WantColon()
+ if in.IsNull() {
+ in.Skip()
+ in.WantComma()
+ continue
+ }
+ switch key {
+ case "body":
+ {
+ var f *ListShardsForObjectRequest_Body
+ f = new(ListShardsForObjectRequest_Body)
+ f.UnmarshalEasyJSON(in)
+ x.Body = f
+ }
+ case "signature":
+ {
+ var f *Signature
+ f = new(Signature)
+ f.UnmarshalEasyJSON(in)
+ x.Signature = f
+ }
+ }
+ in.WantComma()
+ }
+ in.Delim('}')
+ if isTopLevel {
+ in.Consumed()
+ }
+}
+
+type ListShardsForObjectResponse_Body struct {
+ Shard_ID [][]byte `json:"shardID"`
+}
+
+var (
+ _ encoding.ProtoMarshaler = (*ListShardsForObjectResponse_Body)(nil)
+ _ encoding.ProtoUnmarshaler = (*ListShardsForObjectResponse_Body)(nil)
+ _ json.Marshaler = (*ListShardsForObjectResponse_Body)(nil)
+ _ json.Unmarshaler = (*ListShardsForObjectResponse_Body)(nil)
+)
+
+// StableSize returns the size of x in protobuf format.
+//
+// Structures with the same field values have the same binary size.
+func (x *ListShardsForObjectResponse_Body) StableSize() (size int) {
+ if x == nil {
+ return 0
+ }
+ size += proto.RepeatedBytesSize(1, x.Shard_ID)
+ return size
+}
+
+// MarshalProtobuf implements the encoding.ProtoMarshaler interface.
+func (x *ListShardsForObjectResponse_Body) MarshalProtobuf(dst []byte) []byte {
+ m := pool.MarshalerPool.Get()
+ defer pool.MarshalerPool.Put(m)
+ x.EmitProtobuf(m.MessageMarshaler())
+ dst = m.Marshal(dst)
+ return dst
+}
+
+func (x *ListShardsForObjectResponse_Body) EmitProtobuf(mm *easyproto.MessageMarshaler) {
+ if x == nil {
+ return
+ }
+ for j := range x.Shard_ID {
+ mm.AppendBytes(1, x.Shard_ID[j])
+ }
+}
+
+// UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
+func (x *ListShardsForObjectResponse_Body) UnmarshalProtobuf(src []byte) (err error) {
+ var fc easyproto.FieldContext
+ for len(src) > 0 {
+ src, err = fc.NextField(src)
+ if err != nil {
+ return fmt.Errorf("cannot read next field in %s", "ListShardsForObjectResponse_Body")
+ }
+ switch fc.FieldNum {
+ case 1: // Shard_ID
+ data, ok := fc.Bytes()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "Shard_ID")
+ }
+ x.Shard_ID = append(x.Shard_ID, data)
+ }
+ }
+ return nil
+}
+func (x *ListShardsForObjectResponse_Body) GetShard_ID() [][]byte {
+ if x != nil {
+ return x.Shard_ID
+ }
+ return nil
+}
+func (x *ListShardsForObjectResponse_Body) SetShard_ID(v [][]byte) {
+ x.Shard_ID = v
+}
+
+// MarshalJSON implements the json.Marshaler interface.
+func (x *ListShardsForObjectResponse_Body) MarshalJSON() ([]byte, error) {
+ w := jwriter.Writer{}
+ x.MarshalEasyJSON(&w)
+ return w.Buffer.BuildBytes(), w.Error
+}
+func (x *ListShardsForObjectResponse_Body) MarshalEasyJSON(out *jwriter.Writer) {
+ if x == nil {
+ out.RawString("null")
+ return
+ }
+ first := true
+ out.RawByte('{')
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"shardID\":"
+ out.RawString(prefix)
+ out.RawByte('[')
+ for i := range x.Shard_ID {
+ if i != 0 {
+ out.RawByte(',')
+ }
+ if x.Shard_ID[i] != nil {
+ out.Base64Bytes(x.Shard_ID[i])
+ } else {
+ out.String("")
+ }
+ }
+ out.RawByte(']')
+ }
+ out.RawByte('}')
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface.
+func (x *ListShardsForObjectResponse_Body) UnmarshalJSON(data []byte) error {
+ r := jlexer.Lexer{Data: data}
+ x.UnmarshalEasyJSON(&r)
+ return r.Error()
+}
+func (x *ListShardsForObjectResponse_Body) UnmarshalEasyJSON(in *jlexer.Lexer) {
+ isTopLevel := in.IsStart()
+ if in.IsNull() {
+ if isTopLevel {
+ in.Consumed()
+ }
+ in.Skip()
+ return
+ }
+ in.Delim('{')
+ for !in.IsDelim('}') {
+ key := in.UnsafeFieldName(false)
+ in.WantColon()
+ if in.IsNull() {
+ in.Skip()
+ in.WantComma()
+ continue
+ }
+ switch key {
+ case "shardID":
+ {
+ var f []byte
+ var list [][]byte
+ in.Delim('[')
+ for !in.IsDelim(']') {
+ {
+ tmp := in.Bytes()
+ if len(tmp) == 0 {
+ tmp = nil
+ }
+ f = tmp
+ }
+ list = append(list, f)
+ in.WantComma()
+ }
+ x.Shard_ID = list
+ in.Delim(']')
+ }
+ }
+ in.WantComma()
+ }
+ in.Delim('}')
+ if isTopLevel {
+ in.Consumed()
+ }
+}
+
+type ListShardsForObjectResponse struct {
+ Body *ListShardsForObjectResponse_Body `json:"body"`
+ Signature *Signature `json:"signature"`
+}
+
+var (
+ _ encoding.ProtoMarshaler = (*ListShardsForObjectResponse)(nil)
+ _ encoding.ProtoUnmarshaler = (*ListShardsForObjectResponse)(nil)
+ _ json.Marshaler = (*ListShardsForObjectResponse)(nil)
+ _ json.Unmarshaler = (*ListShardsForObjectResponse)(nil)
+)
+
+// StableSize returns the size of x in protobuf format.
+//
+// Structures with the same field values have the same binary size.
+func (x *ListShardsForObjectResponse) StableSize() (size int) {
+ if x == nil {
+ return 0
+ }
+ size += proto.NestedStructureSize(1, x.Body)
+ size += proto.NestedStructureSize(2, x.Signature)
+ return size
+}
+
+// ReadSignedData fills buf with signed data of x.
+// 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 *ListShardsForObjectResponse) SignedDataSize() int {
+ return x.GetBody().StableSize()
+}
+
+// SignedDataSize returns size of the request signed data in bytes.
+//
+// Structures with the same field values have the same signed data size.
+func (x *ListShardsForObjectResponse) ReadSignedData(buf []byte) ([]byte, error) {
+ return x.GetBody().MarshalProtobuf(buf), nil
+}
+
+// MarshalProtobuf implements the encoding.ProtoMarshaler interface.
+func (x *ListShardsForObjectResponse) MarshalProtobuf(dst []byte) []byte {
+ m := pool.MarshalerPool.Get()
+ defer pool.MarshalerPool.Put(m)
+ x.EmitProtobuf(m.MessageMarshaler())
+ dst = m.Marshal(dst)
+ return dst
+}
+
+func (x *ListShardsForObjectResponse) EmitProtobuf(mm *easyproto.MessageMarshaler) {
+ if x == nil {
+ return
+ }
+ if x.Body != nil {
+ x.Body.EmitProtobuf(mm.AppendMessage(1))
+ }
+ if x.Signature != nil {
+ x.Signature.EmitProtobuf(mm.AppendMessage(2))
+ }
+}
+
+// UnmarshalProtobuf implements the encoding.ProtoUnmarshaler interface.
+func (x *ListShardsForObjectResponse) UnmarshalProtobuf(src []byte) (err error) {
+ var fc easyproto.FieldContext
+ for len(src) > 0 {
+ src, err = fc.NextField(src)
+ if err != nil {
+ return fmt.Errorf("cannot read next field in %s", "ListShardsForObjectResponse")
+ }
+ switch fc.FieldNum {
+ case 1: // Body
+ data, ok := fc.MessageData()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "Body")
+ }
+ x.Body = new(ListShardsForObjectResponse_Body)
+ if err := x.Body.UnmarshalProtobuf(data); err != nil {
+ return fmt.Errorf("unmarshal: %w", err)
+ }
+ case 2: // Signature
+ data, ok := fc.MessageData()
+ if !ok {
+ return fmt.Errorf("cannot unmarshal field %s", "Signature")
+ }
+ x.Signature = new(Signature)
+ if err := x.Signature.UnmarshalProtobuf(data); err != nil {
+ return fmt.Errorf("unmarshal: %w", err)
+ }
+ }
+ }
+ return nil
+}
+func (x *ListShardsForObjectResponse) GetBody() *ListShardsForObjectResponse_Body {
+ if x != nil {
+ return x.Body
+ }
+ return nil
+}
+func (x *ListShardsForObjectResponse) SetBody(v *ListShardsForObjectResponse_Body) {
+ x.Body = v
+}
+func (x *ListShardsForObjectResponse) GetSignature() *Signature {
+ if x != nil {
+ return x.Signature
+ }
+ return nil
+}
+func (x *ListShardsForObjectResponse) SetSignature(v *Signature) {
+ x.Signature = v
+}
+
+// MarshalJSON implements the json.Marshaler interface.
+func (x *ListShardsForObjectResponse) MarshalJSON() ([]byte, error) {
+ w := jwriter.Writer{}
+ x.MarshalEasyJSON(&w)
+ return w.Buffer.BuildBytes(), w.Error
+}
+func (x *ListShardsForObjectResponse) MarshalEasyJSON(out *jwriter.Writer) {
+ if x == nil {
+ out.RawString("null")
+ return
+ }
+ first := true
+ out.RawByte('{')
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"body\":"
+ out.RawString(prefix)
+ x.Body.MarshalEasyJSON(out)
+ }
+ {
+ if !first {
+ out.RawByte(',')
+ } else {
+ first = false
+ }
+ const prefix string = "\"signature\":"
+ out.RawString(prefix)
+ x.Signature.MarshalEasyJSON(out)
+ }
+ out.RawByte('}')
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface.
+func (x *ListShardsForObjectResponse) UnmarshalJSON(data []byte) error {
+ r := jlexer.Lexer{Data: data}
+ x.UnmarshalEasyJSON(&r)
+ return r.Error()
+}
+func (x *ListShardsForObjectResponse) UnmarshalEasyJSON(in *jlexer.Lexer) {
+ isTopLevel := in.IsStart()
+ if in.IsNull() {
+ if isTopLevel {
+ in.Consumed()
+ }
+ in.Skip()
+ return
+ }
+ in.Delim('{')
+ for !in.IsDelim('}') {
+ key := in.UnsafeFieldName(false)
+ in.WantColon()
+ if in.IsNull() {
+ in.Skip()
+ in.WantComma()
+ continue
+ }
+ switch key {
+ case "body":
+ {
+ var f *ListShardsForObjectResponse_Body
+ f = new(ListShardsForObjectResponse_Body)
+ f.UnmarshalEasyJSON(in)
+ x.Body = f
+ }
+ case "signature":
+ {
+ var f *Signature
+ f = new(Signature)
+ f.UnmarshalEasyJSON(in)
+ x.Signature = f
+ }
+ }
+ in.WantComma()
+ }
+ in.Delim('}')
+ if isTopLevel {
+ in.Consumed()
+ }
+}
diff --git a/pkg/services/control/service_grpc.pb.go b/pkg/services/control/service_grpc.pb.go
index 987e08c5..045662cc 100644
--- a/pkg/services/control/service_grpc.pb.go
+++ b/pkg/services/control/service_grpc.pb.go
@@ -41,6 +41,7 @@ const (
ControlService_SealWriteCache_FullMethodName = "/control.ControlService/SealWriteCache"
ControlService_DetachShards_FullMethodName = "/control.ControlService/DetachShards"
ControlService_StartShardRebuild_FullMethodName = "/control.ControlService/StartShardRebuild"
+ ControlService_ListShardsForObject_FullMethodName = "/control.ControlService/ListShardsForObject"
)
// ControlServiceClient is the client API for ControlService service.
@@ -95,6 +96,8 @@ type ControlServiceClient interface {
DetachShards(ctx context.Context, in *DetachShardsRequest, opts ...grpc.CallOption) (*DetachShardsResponse, error)
// StartShardRebuild starts shard rebuild process.
StartShardRebuild(ctx context.Context, in *StartShardRebuildRequest, opts ...grpc.CallOption) (*StartShardRebuildResponse, error)
+ // ListShardsForObject returns shard info where object is stored.
+ ListShardsForObject(ctx context.Context, in *ListShardsForObjectRequest, opts ...grpc.CallOption) (*ListShardsForObjectResponse, error)
}
type controlServiceClient struct {
@@ -303,6 +306,15 @@ func (c *controlServiceClient) StartShardRebuild(ctx context.Context, in *StartS
return out, nil
}
+func (c *controlServiceClient) ListShardsForObject(ctx context.Context, in *ListShardsForObjectRequest, opts ...grpc.CallOption) (*ListShardsForObjectResponse, error) {
+ out := new(ListShardsForObjectResponse)
+ err := c.cc.Invoke(ctx, ControlService_ListShardsForObject_FullMethodName, in, out, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
// ControlServiceServer is the server API for ControlService service.
// All implementations should embed UnimplementedControlServiceServer
// for forward compatibility
@@ -355,6 +367,8 @@ type ControlServiceServer interface {
DetachShards(context.Context, *DetachShardsRequest) (*DetachShardsResponse, error)
// StartShardRebuild starts shard rebuild process.
StartShardRebuild(context.Context, *StartShardRebuildRequest) (*StartShardRebuildResponse, error)
+ // ListShardsForObject returns shard info where object is stored.
+ ListShardsForObject(context.Context, *ListShardsForObjectRequest) (*ListShardsForObjectResponse, error)
}
// UnimplementedControlServiceServer should be embedded to have forward compatible implementations.
@@ -427,6 +441,9 @@ func (UnimplementedControlServiceServer) DetachShards(context.Context, *DetachSh
func (UnimplementedControlServiceServer) StartShardRebuild(context.Context, *StartShardRebuildRequest) (*StartShardRebuildResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method StartShardRebuild not implemented")
}
+func (UnimplementedControlServiceServer) ListShardsForObject(context.Context, *ListShardsForObjectRequest) (*ListShardsForObjectResponse, error) {
+ return nil, status.Errorf(codes.Unimplemented, "method ListShardsForObject not implemented")
+}
// UnsafeControlServiceServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ControlServiceServer will
@@ -835,6 +852,24 @@ func _ControlService_StartShardRebuild_Handler(srv interface{}, ctx context.Cont
return interceptor(ctx, in, info, handler)
}
+func _ControlService_ListShardsForObject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(ListShardsForObjectRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(ControlServiceServer).ListShardsForObject(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: ControlService_ListShardsForObject_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(ControlServiceServer).ListShardsForObject(ctx, req.(*ListShardsForObjectRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
// ControlService_ServiceDesc is the grpc.ServiceDesc for ControlService service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@@ -930,6 +965,10 @@ var ControlService_ServiceDesc = grpc.ServiceDesc{
MethodName: "StartShardRebuild",
Handler: _ControlService_StartShardRebuild_Handler,
},
+ {
+ MethodName: "ListShardsForObject",
+ Handler: _ControlService_ListShardsForObject_Handler,
+ },
},
Streams: []grpc.StreamDesc{},
Metadata: "pkg/services/control/service.proto",