package services import ( "context" "errors" "fmt" "io" "strings" "sync/atomic" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer" "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" grpcService "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/services/tree" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/pkg/service/tree" "github.com/nspcc-dev/neo-go/pkg/crypto/keys" "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) type GetNodeByPathResponseInfoWrapper struct { response *grpcService.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() []tree.Meta { res := make([]tree.Meta, len(n.response.Meta)) for i, value := range n.response.Meta { res[i] = value } return res } type GetSubTreeResponseBodyWrapper struct { response *grpcService.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() []tree.Meta { res := make([]tree.Meta, len(n.response.Meta)) for i, value := range n.response.Meta { res[i] = value } return res } type ServiceClientGRPC struct { key *keys.PrivateKey log *zap.Logger clients []treeClient startIndex int32 } type treeClient struct { address string conn *grpc.ClientConn service grpcService.TreeServiceClient } func (c *ServiceClientGRPC) Endpoints() []string { res := make([]string, len(c.clients)) for i, client := range c.clients { res[i] = client.address } return res } func NewTreeServiceClientGRPC(ctx context.Context, endpoints []string, key *keys.PrivateKey, log *zap.Logger, grpcOpts ...grpc.DialOption) (*ServiceClientGRPC, error) { res := &ServiceClientGRPC{ key: key, log: log, } for _, addr := range endpoints { conn, err := grpc.Dial(addr, grpcOpts...) if err != nil { log.Warn("dial node tree service", zap.String("address", addr), zap.Error(err)) continue } c := grpcService.NewTreeServiceClient(conn) if _, err = c.Healthcheck(ctx, &grpcService.HealthcheckRequest{}); err != nil { log.Warn("healthcheck tree service", zap.String("address", addr), zap.Error(err)) continue } res.clients = append(res.clients, treeClient{ address: addr, conn: conn, service: c, }) } if len(res.clients) == 0 { return nil, errors.New("no healthy tree grpc client") } return res, nil } func (c *ServiceClientGRPC) GetNodes(ctx context.Context, p *tree.GetNodesParams) ([]tree.NodeResponse, error) { request := &grpcService.GetNodeByPathRequest{ Body: &grpcService.GetNodeByPathRequest_Body{ ContainerId: p.BktInfo.CID[:], TreeId: p.TreeID, Path: p.Path, Attributes: p.Meta, PathAttribute: tree.FileNameKey, LatestOnly: p.LatestOnly, AllAttributes: p.AllAttrs, BearerToken: getBearer(ctx, p.BktInfo), }, } if err := c.signRequest(request.Body, func(key, sign []byte) { request.Signature = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return nil, err } var resp *grpcService.GetNodeByPathResponse if err := c.requestWithRetry(func(client treeClient) (inErr error) { resp, inErr = client.service.GetNodeByPath(ctx, request) return handleError("failed to get node by path", inErr) }); err != nil { return nil, err } res := make([]tree.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) ([]tree.NodeResponse, error) { request := &grpcService.GetSubTreeRequest{ Body: &grpcService.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 = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return nil, err } var cli grpcService.TreeService_GetSubTreeClient if err := c.requestWithRetry(func(client treeClient) (inErr error) { cli, inErr = client.service.GetSubTree(ctx, request) return handleError("failed to get sub tree client", inErr) }); err != nil { return nil, err } var subtree []tree.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 := &grpcService.AddRequest{ Body: &grpcService.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 = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return 0, err } var resp *grpcService.AddResponse if err := c.requestWithRetry(func(client treeClient) (inErr error) { resp, inErr = client.service.Add(ctx, request) return handleError("failed to add node", inErr) }); err != nil { return 0, 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 := &grpcService.AddByPathRequest{ Body: &grpcService.AddByPathRequest_Body{ ContainerId: bktInfo.CID[:], TreeId: treeID, Path: path, Meta: metaToKV(meta), PathAttribute: tree.FileNameKey, BearerToken: getBearer(ctx, bktInfo), }, } if err := c.signRequest(request.Body, func(key, sign []byte) { request.Signature = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return 0, err } var resp *grpcService.AddByPathResponse if err := c.requestWithRetry(func(client treeClient) (inErr error) { resp, inErr = client.service.AddByPath(ctx, request) return handleError("failed to add node by path", inErr) }); err != nil { return 0, 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 := &grpcService.MoveRequest{ Body: &grpcService.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 = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return err } return c.requestWithRetry(func(client treeClient) error { if _, err := client.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 := &grpcService.RemoveRequest{ Body: &grpcService.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 = &grpcService.Signature{ Key: key, Sign: sign, } }); err != nil { return err } return c.requestWithRetry(func(client treeClient) error { if _, err := client.service.Remove(ctx, request); err != nil { return handleError("failed to remove node", err) } return nil }) } func (c *ServiceClientGRPC) requestWithRetry(fn func(client treeClient) error) (err error) { start := int(atomic.LoadInt32(&c.startIndex)) for i := start; i < start+len(c.clients); i++ { index := i % len(c.clients) err = fn(c.clients[index]) if !shouldTryAgain(err) { atomic.StoreInt32(&c.startIndex, int32(index)) return err } c.log.Debug("tree request error", zap.String("address", c.clients[index].address), zap.Error(err)) } return err } func shouldTryAgain(err error) bool { if err == nil { return false } code := status.Code(unwrapErr(err)) if code == codes.Unavailable || code == codes.Unimplemented { return true } errText := err.Error() if strings.Contains(errText, "not found") || strings.Contains(errText, "shard is in read-only mode") || strings.Contains(errText, "shard is in degraded mode") { return true } return false } func unwrapErr(err error) error { for e := errors.Unwrap(err); e != nil; e = errors.Unwrap(err) { err = e } return err } func metaToKV(meta map[string]string) []*grpcService.KeyValue { result := make([]*grpcService.KeyValue, 0, len(meta)) for key, value := range meta { result = append(result, &grpcService.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 bd.Gate.BearerToken.Impersonate() || bktInfo.Owner.Equals(bearer.ResolveIssuer(*bd.Gate.BearerToken)) { return bd.Gate.BearerToken.Marshal() } } } return nil } func handleError(msg string, err error) error { if err == nil { return nil } if strings.Contains(err.Error(), "not found") { return fmt.Errorf("%w: %s", tree.ErrNodeNotFound, err.Error()) } else if strings.Contains(err.Error(), "is denied by") { return fmt.Errorf("%w: %s", tree.ErrNodeAccessDenied, err.Error()) } return fmt.Errorf("%s: %w", msg, err) }