frostfs-s3-gw/pkg/service/tree/tree_client_grpc.go

312 lines
7.9 KiB
Go
Raw Normal View History

package tree
import (
"context"
"errors"
"fmt"
"io"
"strings"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/accessbox"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/services/tree"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"google.golang.org/grpc"
)
type GetNodeByPathResponseInfoWrapper struct {
response *tree.GetNodeByPathResponse_Info
}
func (n GetNodeByPathResponseInfoWrapper) GetNodeID() uint64 {
return n.response.GetNodeId()
}
func (n GetNodeByPathResponseInfoWrapper) GetParentID() uint64 {
return n.response.GetParentId()
}
func (n GetNodeByPathResponseInfoWrapper) GetTimestamp() uint64 {
return n.response.GetTimestamp()
}
func (n GetNodeByPathResponseInfoWrapper) GetMeta() []Meta {
res := make([]Meta, len(n.response.Meta))
for i, value := range n.response.Meta {
res[i] = value
}
return res
}
type GetSubTreeResponseBodyWrapper struct {
response *tree.GetSubTreeResponse_Body
}
func (n GetSubTreeResponseBodyWrapper) GetNodeID() uint64 {
return n.response.GetNodeId()
}
func (n GetSubTreeResponseBodyWrapper) GetParentID() uint64 {
return n.response.GetParentId()
}
func (n GetSubTreeResponseBodyWrapper) GetTimestamp() uint64 {
return n.response.GetTimestamp()
}
func (n GetSubTreeResponseBodyWrapper) GetMeta() []Meta {
res := make([]Meta, len(n.response.Meta))
for i, value := range n.response.Meta {
res[i] = value
}
return res
}
type ServiceClientGRPC struct {
key *keys.PrivateKey
conn *grpc.ClientConn
service tree.TreeServiceClient
}
func NewTreeServiceClientGRPC(ctx context.Context, addr string, key *keys.PrivateKey, grpcOpts ...grpc.DialOption) (*ServiceClientGRPC, error) {
conn, err := grpc.Dial(addr, grpcOpts...)
if err != nil {
return nil, fmt.Errorf("did not connect: %v", err)
}
c := tree.NewTreeServiceClient(conn)
if _, err = c.Healthcheck(ctx, &tree.HealthcheckRequest{}); err != nil {
return nil, fmt.Errorf("healthcheck: %w", err)
}
return &ServiceClientGRPC{
key: key,
conn: conn,
service: c,
}, nil
}
func (c *ServiceClientGRPC) GetNodes(ctx context.Context, p *GetNodesParams) ([]NodeResponse, error) {
request := &tree.GetNodeByPathRequest{
Body: &tree.GetNodeByPathRequest_Body{
ContainerId: p.BktInfo.CID[:],
TreeId: p.TreeID,
Path: p.Path,
Attributes: p.Meta,
PathAttribute: FileNameKey,
LatestOnly: p.LatestOnly,
AllAttributes: p.AllAttrs,
BearerToken: getBearer(ctx, p.BktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return nil, err
}
resp, err := c.service.GetNodeByPath(ctx, request)
if err != nil {
return nil, handleError("failed to get node by path", err)
}
res := make([]NodeResponse, len(resp.GetBody().GetNodes()))
for i, info := range resp.GetBody().GetNodes() {
res[i] = GetNodeByPathResponseInfoWrapper{info}
}
return res, nil
}
func (c *ServiceClientGRPC) GetSubTree(ctx context.Context, bktInfo *data.BucketInfo, treeID string, rootID uint64, depth uint32) ([]NodeResponse, error) {
request := &tree.GetSubTreeRequest{
Body: &tree.GetSubTreeRequest_Body{
ContainerId: bktInfo.CID[:],
TreeId: treeID,
RootId: rootID,
Depth: depth,
BearerToken: getBearer(ctx, bktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return nil, err
}
cli, err := c.service.GetSubTree(ctx, request)
if err != nil {
return nil, handleError("failed to get sub tree client", err)
}
var subtree []NodeResponse
for {
resp, err := cli.Recv()
if err == io.EOF {
break
} else if err != nil {
return nil, handleError("failed to get sub tree", err)
}
subtree = append(subtree, GetSubTreeResponseBodyWrapper{resp.Body})
}
return subtree, nil
}
func (c *ServiceClientGRPC) AddNode(ctx context.Context, bktInfo *data.BucketInfo, treeID string, parent uint64, meta map[string]string) (uint64, error) {
request := &tree.AddRequest{
Body: &tree.AddRequest_Body{
ContainerId: bktInfo.CID[:],
TreeId: treeID,
ParentId: parent,
Meta: metaToKV(meta),
BearerToken: getBearer(ctx, bktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return 0, err
}
resp, err := c.service.Add(ctx, request)
if err != nil {
return 0, handleError("failed to add node", err)
}
return resp.GetBody().GetNodeId(), nil
}
func (c *ServiceClientGRPC) AddNodeByPath(ctx context.Context, bktInfo *data.BucketInfo, treeID string, path []string, meta map[string]string) (uint64, error) {
request := &tree.AddByPathRequest{
Body: &tree.AddByPathRequest_Body{
ContainerId: bktInfo.CID[:],
TreeId: treeID,
Path: path,
Meta: metaToKV(meta),
PathAttribute: FileNameKey,
BearerToken: getBearer(ctx, bktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return 0, err
}
resp, err := c.service.AddByPath(ctx, request)
if err != nil {
return 0, handleError("failed to add node by path", err)
}
body := resp.GetBody()
if body == nil {
return 0, errors.New("nil body in tree service response")
} else if len(body.Nodes) == 0 {
return 0, errors.New("empty list of added nodes in tree service response")
}
// The first node is the leaf that we add, according to tree service docs.
return body.Nodes[0], nil
}
func (c *ServiceClientGRPC) MoveNode(ctx context.Context, bktInfo *data.BucketInfo, treeID string, nodeID, parentID uint64, meta map[string]string) error {
request := &tree.MoveRequest{
Body: &tree.MoveRequest_Body{
ContainerId: bktInfo.CID[:],
TreeId: treeID,
NodeId: nodeID,
ParentId: parentID,
Meta: metaToKV(meta),
BearerToken: getBearer(ctx, bktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return err
}
if _, err := c.service.Move(ctx, request); err != nil {
return handleError("failed to move node", err)
}
return nil
}
func (c *ServiceClientGRPC) RemoveNode(ctx context.Context, bktInfo *data.BucketInfo, treeID string, nodeID uint64) error {
request := &tree.RemoveRequest{
Body: &tree.RemoveRequest_Body{
ContainerId: bktInfo.CID[:],
TreeId: treeID,
NodeId: nodeID,
BearerToken: getBearer(ctx, bktInfo),
},
}
if err := c.signRequest(request.Body, func(key, sign []byte) {
request.Signature = &tree.Signature{
Key: key,
Sign: sign,
}
}); err != nil {
return err
}
if _, err := c.service.Remove(ctx, request); err != nil {
return handleError("failed to remove node", err)
}
return nil
}
func metaToKV(meta map[string]string) []*tree.KeyValue {
result := make([]*tree.KeyValue, 0, len(meta))
for key, value := range meta {
result = append(result, &tree.KeyValue{Key: key, Value: []byte(value)})
}
return result
}
func getBearer(ctx context.Context, bktInfo *data.BucketInfo) []byte {
if bd, ok := ctx.Value(api.BoxData).(*accessbox.Box); ok && bd != nil && bd.Gate != nil {
if bd.Gate.BearerToken != nil {
if bktInfo.Owner.Equals(bearer.ResolveIssuer(*bd.Gate.BearerToken)) {
return bd.Gate.BearerToken.Marshal()
}
}
}
return nil
}
func handleError(msg string, err error) error {
if strings.Contains(err.Error(), "not found") {
return fmt.Errorf("%w: %s", ErrNodeNotFound, err.Error())
} else if strings.Contains(err.Error(), "is denied by") {
return fmt.Errorf("%w: %s", ErrNodeAccessDenied, err.Error())
}
return fmt.Errorf("%s: %w", msg, err)
}