storj/uplink/metainfo/client.go

1235 lines
37 KiB
Go
Raw Normal View History

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package metainfo
import (
"bytes"
"context"
"time"
"github.com/skyrings/skyring-common/tools/uuid"
"github.com/zeebo/errs"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"gopkg.in/spacemonkeygo/monkit.v2"
"storj.io/storj/pkg/auth/grpcauth"
"storj.io/storj/pkg/pb"
"storj.io/storj/pkg/storj"
"storj.io/storj/pkg/transport"
"storj.io/storj/storage"
)
var (
mon = monkit.Package()
// Error is the errs class of standard metainfo errors
Error = errs.Class("metainfo error")
)
2019-06-25 16:36:23 +01:00
// Client creates a grpcClient
type Client struct {
client pb.MetainfoClient
2019-06-25 16:36:23 +01:00
conn *grpc.ClientConn
}
// ListItem is a single item in a listing
type ListItem struct {
Path storj.Path
Pointer *pb.Pointer
IsPrefix bool
}
2019-06-25 16:36:23 +01:00
// New used as a public function
func New(client pb.MetainfoClient) *Client {
return &Client{
client: client,
}
}
2019-06-25 16:36:23 +01:00
// Dial dials to metainfo endpoint with the specified api key.
func Dial(ctx context.Context, tc transport.Client, address string, apikey string) (*Client, error) {
conn, err := tc.DialAddress(
ctx,
address,
grpc.WithPerRPCCredentials(grpcauth.NewAPIKeyCredentials(apikey)),
)
if err != nil {
return nil, Error.Wrap(err)
}
2019-06-25 16:36:23 +01:00
return &Client{
client: pb.NewMetainfoClient(conn),
conn: conn,
}, nil
}
// Close closes the dialed connection.
func (client *Client) Close() error {
if client.conn != nil {
return Error.Wrap(client.conn.Close())
}
return nil
}
// CreateSegment requests the order limits for creating a new segment
func (client *Client) CreateSegment(ctx context.Context, bucket string, path storj.Path, segmentIndex int64, redundancy *pb.RedundancyScheme, maxEncryptedSegmentSize int64, expiration time.Time) (limits []*pb.AddressedOrderLimit, rootPieceID storj.PieceID, piecePrivateKey storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.CreateSegmentOld(ctx, &pb.SegmentWriteRequestOld{
Bucket: []byte(bucket),
Path: []byte(path),
Segment: segmentIndex,
Redundancy: redundancy,
MaxEncryptedSegmentSize: maxEncryptedSegmentSize,
Expiration: expiration,
})
if err != nil {
return nil, rootPieceID, piecePrivateKey, Error.Wrap(err)
}
return response.GetAddressedLimits(), response.RootPieceId, response.PrivateKey, nil
}
// CommitSegment requests to store the pointer for the segment
func (client *Client) CommitSegment(ctx context.Context, bucket string, path storj.Path, segmentIndex int64, pointer *pb.Pointer, originalLimits []*pb.OrderLimit) (savedPointer *pb.Pointer, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.CommitSegmentOld(ctx, &pb.SegmentCommitRequestOld{
Bucket: []byte(bucket),
Path: []byte(path),
Segment: segmentIndex,
Pointer: pointer,
OriginalLimits: originalLimits,
})
if err != nil {
return nil, Error.Wrap(err)
}
return response.GetPointer(), nil
}
// SegmentInfo requests the pointer of a segment
2019-06-25 16:36:23 +01:00
func (client *Client) SegmentInfo(ctx context.Context, bucket string, path storj.Path, segmentIndex int64) (pointer *pb.Pointer, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.SegmentInfoOld(ctx, &pb.SegmentInfoRequestOld{
Bucket: []byte(bucket),
Path: []byte(path),
Segment: segmentIndex,
})
if err != nil {
if status.Code(err) == codes.NotFound {
return nil, storage.ErrKeyNotFound.Wrap(err)
}
return nil, Error.Wrap(err)
}
return response.GetPointer(), nil
}
// ReadSegment requests the order limits for reading a segment
func (client *Client) ReadSegment(ctx context.Context, bucket string, path storj.Path, segmentIndex int64) (pointer *pb.Pointer, limits []*pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.DownloadSegmentOld(ctx, &pb.SegmentDownloadRequestOld{
Bucket: []byte(bucket),
Path: []byte(path),
Segment: segmentIndex,
})
if err != nil {
if status.Code(err) == codes.NotFound {
return nil, nil, piecePrivateKey, storage.ErrKeyNotFound.Wrap(err)
}
return nil, nil, piecePrivateKey, Error.Wrap(err)
}
return response.GetPointer(), sortLimits(response.GetAddressedLimits(), response.GetPointer()), response.PrivateKey, nil
}
// sortLimits sorts order limits and fill missing ones with nil values
func sortLimits(limits []*pb.AddressedOrderLimit, pointer *pb.Pointer) []*pb.AddressedOrderLimit {
sorted := make([]*pb.AddressedOrderLimit, pointer.GetRemote().GetRedundancy().GetTotal())
for _, piece := range pointer.GetRemote().GetRemotePieces() {
sorted[piece.GetPieceNum()] = getLimitByStorageNodeID(limits, piece.NodeId)
}
return sorted
}
func getLimitByStorageNodeID(limits []*pb.AddressedOrderLimit, storageNodeID storj.NodeID) *pb.AddressedOrderLimit {
for _, limit := range limits {
if limit.GetLimit().StorageNodeId == storageNodeID {
return limit
}
}
return nil
}
// DeleteSegment requests the order limits for deleting a segment
func (client *Client) DeleteSegment(ctx context.Context, bucket string, path storj.Path, segmentIndex int64) (limits []*pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.DeleteSegmentOld(ctx, &pb.SegmentDeleteRequestOld{
Bucket: []byte(bucket),
Path: []byte(path),
Segment: segmentIndex,
})
if err != nil {
if status.Code(err) == codes.NotFound {
return nil, piecePrivateKey, storage.ErrKeyNotFound.Wrap(err)
}
return nil, piecePrivateKey, Error.Wrap(err)
}
return response.GetAddressedLimits(), response.PrivateKey, nil
}
// ListSegments lists the available segments
2019-06-25 16:36:23 +01:00
func (client *Client) ListSegments(ctx context.Context, bucket string, prefix, startAfter, endBefore storj.Path, recursive bool, limit int32, metaFlags uint32) (items []ListItem, more bool, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.ListSegmentsOld(ctx, &pb.ListSegmentsRequestOld{
Bucket: []byte(bucket),
Prefix: []byte(prefix),
StartAfter: []byte(startAfter),
EndBefore: []byte(endBefore),
Recursive: recursive,
Limit: limit,
MetaFlags: metaFlags,
})
if err != nil {
return nil, false, Error.Wrap(err)
}
list := response.GetItems()
items = make([]ListItem, len(list))
for i, item := range list {
items[i] = ListItem{
Path: storj.Path(item.GetPath()),
Pointer: item.GetPointer(),
IsPrefix: item.IsPrefix,
}
}
return items, response.GetMore(), nil
}
// SetAttribution tries to set the attribution information on the bucket.
2019-06-25 16:36:23 +01:00
func (client *Client) SetAttribution(ctx context.Context, bucket string, partnerID uuid.UUID) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.SetAttributionOld(ctx, &pb.SetAttributionRequestOld{
PartnerId: partnerID[:], // TODO: implement storj.UUID that can be sent using pb
BucketName: []byte(bucket),
})
return Error.Wrap(err)
}
lib/uplink: encryption context (#2349) * lib/uplink: encryption context Change-Id: I5c23dca3286a46b713b30c4997e9ae6e630b2280 * lib/uplink: bucket operation examples Change-Id: Ia0f6e69f365dcff0cf11c731f51b30842bce053b * lib/uplink: encryption key sharing test cases Change-Id: I3a172d565f33f4e591402cdcb9460664a7cc7fbe * fix encrypted path prefix restriction issue Change-Id: I8f3921f9d52aaf4b84039de608b8cbbc88769554 * implement panics in libuplink encryption code todo on cipher suite selection as well as an api concern Change-Id: Ifa39eb3cc4b3443f7d96f9304df9b2ac4ec4085d * implement GetProjectInfo api call to get salt Change-Id: Ic5f6b3be9ea35df48c1aa214ab5d355fb328e2cf * some fixes and accessors for encryption store Change-Id: I3bb61f6712a037900e2a96e72ad4029ec1d3f718 * general fixes to builds/tests/etc Change-Id: I9930fa96acb3b221d9a001f8e274af5729cc8a47 * java bindings changes Change-Id: Ia2bd4c9c69739c8d3154d79616cff1f36fb403b6 * get libuplink examples passing Change-Id: I828f09a144160e0a5dd932324f78491ae2ec8a07 * fix proto.lock file Change-Id: I2fbbf4d0976a7d0473c2645e6dcb21aaa3be7651 * fix proto.lock again Change-Id: I92702cf49e1a340eef6379c2be4f7c4a268112a9 * fix golint issues Change-Id: I631ff9f43307a58e3b25a58cbb4a4cc2495f5eb6 * more linting fixes Change-Id: I51f8f30b367b5bca14c94b15417b9a4c9e7aa0ce * bug fixed by structs bump Change-Id: Ibb03c691fce7606c35c08721b3ef0781ab48a38a * retrigger Change-Id: Ieee0470b6a2d07168a1578552e8e7f271ae93a13 * retrigger Change-Id: I753d63853171e6a436c104ce176048892eb974c5 * semantic merge conflict Change-Id: I9419448496de90340569047a6a16a1b858a7978a * update total to match prod defaults Change-Id: I693d55c1ebb28b5803ee1d26e9e198decf82308b * retrigger Change-Id: I28b74d5d6202f61aa3866fe407d423f6a0a14b9e * retrigger Change-Id: I6fd054885c715f602e2cef623fd464c42e88742c * retrigger Change-Id: I6a01bae88c72406d4ed5a8f13bf8a2b3c650bd2d
2019-06-27 18:36:51 +01:00
// GetProjectInfo gets the ProjectInfo for the api key associated with the metainfo client.
func (client *Client) GetProjectInfo(ctx context.Context) (resp *pb.ProjectInfoResponse, err error) {
defer mon.Task()(&ctx)(&err)
return client.client.ProjectInfo(ctx, &pb.ProjectInfoRequest{})
}
2019-08-06 15:56:23 +01:00
// CreateBucketParams parameters for CreateBucket method
type CreateBucketParams struct {
Name []byte
PathCipher storj.CipherSuite
PartnerID []byte
DefaultSegmentsSize int64
DefaultRedundancyScheme storj.RedundancyScheme
DefaultEncryptionParameters storj.EncryptionParameters
}
func (params *CreateBucketParams) toRequest() *pb.BucketCreateRequest {
defaultRS := params.DefaultRedundancyScheme
defaultEP := params.DefaultEncryptionParameters
return &pb.BucketCreateRequest{
Name: params.Name,
PathCipher: pb.CipherSuite(params.PathCipher),
PartnerId: params.PartnerID,
DefaultSegmentSize: params.DefaultSegmentsSize,
DefaultRedundancyScheme: &pb.RedundancyScheme{
Type: pb.RedundancyScheme_SchemeType(defaultRS.Algorithm),
MinReq: int32(defaultRS.RequiredShares),
Total: int32(defaultRS.TotalShares),
RepairThreshold: int32(defaultRS.RepairShares),
SuccessThreshold: int32(defaultRS.OptimalShares),
ErasureShareSize: defaultRS.ShareSize,
},
DefaultEncryptionParameters: &pb.EncryptionParameters{
CipherSuite: pb.CipherSuite(defaultEP.CipherSuite),
BlockSize: int64(defaultEP.BlockSize),
},
}
}
// BatchItem returns single item for batch request
func (params *CreateBucketParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_BucketCreate{
BucketCreate: params.toRequest(),
},
}
}
// TODO potential names *Response/*Out/*Result
// CreateBucketResponse response for CreateBucket request
type CreateBucketResponse struct {
Bucket storj.Bucket
}
func newCreateBucketResponse(response *pb.BucketCreateResponse) (CreateBucketResponse, error) {
bucket, err := convertProtoToBucket(response.Bucket)
if err != nil {
return CreateBucketResponse{}, err
}
return CreateBucketResponse{
Bucket: bucket,
}, nil
}
// CreateBucket creates a new bucket
2019-08-06 15:56:23 +01:00
func (client *Client) CreateBucket(ctx context.Context, params CreateBucketParams) (respBucket storj.Bucket, err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
response, err := client.client.CreateBucket(ctx, params.toRequest())
if err != nil {
2019-08-06 15:56:23 +01:00
return storj.Bucket{}, Error.Wrap(err)
}
2019-08-06 15:56:23 +01:00
respBucket, err = convertProtoToBucket(response.Bucket)
if err != nil {
return storj.Bucket{}, Error.Wrap(err)
}
2019-08-06 15:56:23 +01:00
return respBucket, nil
}
2019-08-06 15:56:23 +01:00
// GetBucketParams parmaters for GetBucketParams method
type GetBucketParams struct {
Name []byte
}
func (params *GetBucketParams) toRequest() *pb.BucketGetRequest {
return &pb.BucketGetRequest{Name: params.Name}
}
// BatchItem returns single item for batch request
func (params *GetBucketParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_BucketGet{
BucketGet: params.toRequest(),
},
}
}
// GetBucketResponse response for GetBucket request
type GetBucketResponse struct {
Bucket storj.Bucket
}
func newGetBucketResponse(response *pb.BucketGetResponse) (GetBucketResponse, error) {
bucket, err := convertProtoToBucket(response.Bucket)
if err != nil {
2019-08-06 15:56:23 +01:00
return GetBucketResponse{}, err
}
2019-08-06 15:56:23 +01:00
return GetBucketResponse{
Bucket: bucket,
}, nil
}
// GetBucket returns a bucket
2019-08-06 15:56:23 +01:00
func (client *Client) GetBucket(ctx context.Context, params GetBucketParams) (respBucket storj.Bucket, err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
resp, err := client.client.GetBucket(ctx, params.toRequest())
if err != nil {
if status.Code(err) == codes.NotFound {
return storj.Bucket{}, storj.ErrBucketNotFound.Wrap(err)
}
return storj.Bucket{}, Error.Wrap(err)
}
respBucket, err = convertProtoToBucket(resp.Bucket)
if err != nil {
2019-08-06 15:56:23 +01:00
return storj.Bucket{}, Error.Wrap(err)
}
return respBucket, nil
}
2019-08-06 15:56:23 +01:00
// DeleteBucketParams parmaters for DeleteBucket method
type DeleteBucketParams struct {
Name []byte
}
func (params *DeleteBucketParams) toRequest() *pb.BucketDeleteRequest {
return &pb.BucketDeleteRequest{Name: params.Name}
}
// BatchItem returns single item for batch request
func (params *DeleteBucketParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_BucketDelete{
BucketDelete: params.toRequest(),
},
}
}
// DeleteBucket deletes a bucket
2019-08-06 15:56:23 +01:00
func (client *Client) DeleteBucket(ctx context.Context, params DeleteBucketParams) (err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
_, err = client.client.DeleteBucket(ctx, params.toRequest())
if err != nil {
if status.Code(err) == codes.NotFound {
return storj.ErrBucketNotFound.Wrap(err)
}
return Error.Wrap(err)
}
return nil
}
2019-08-06 15:56:23 +01:00
// ListBucketsParams parmaters for ListBucketsParams method
type ListBucketsParams struct {
ListOpts storj.BucketListOptions
}
func (params *ListBucketsParams) toRequest() *pb.BucketListRequest {
return &pb.BucketListRequest{
Cursor: []byte(params.ListOpts.Cursor),
Limit: int32(params.ListOpts.Limit),
Direction: int32(params.ListOpts.Direction),
}
}
// BatchItem returns single item for batch request
func (params *ListBucketsParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_BucketList{
BucketList: params.toRequest(),
},
}
}
// ListBucketsResponse response for ListBucket request
type ListBucketsResponse struct {
BucketList storj.BucketList
}
func newListBucketsResponse(response *pb.BucketListResponse) ListBucketsResponse {
bucketList := storj.BucketList{
More: response.More,
}
bucketList.Items = make([]storj.Bucket, len(response.Items))
for i, item := range response.GetItems() {
bucketList.Items[i] = storj.Bucket{
Name: string(item.Name),
Created: item.CreatedAt,
}
}
return ListBucketsResponse{
BucketList: bucketList,
}
}
// ListBuckets lists buckets
2019-08-06 15:56:23 +01:00
func (client *Client) ListBuckets(ctx context.Context, params ListBucketsParams) (_ storj.BucketList, err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
resp, err := client.client.ListBuckets(ctx, params.toRequest())
if err != nil {
return storj.BucketList{}, Error.Wrap(err)
}
resultBucketList := storj.BucketList{
More: resp.GetMore(),
}
resultBucketList.Items = make([]storj.Bucket, len(resp.GetItems()))
for i, item := range resp.GetItems() {
resultBucketList.Items[i] = storj.Bucket{
Name: string(item.GetName()),
Created: item.GetCreatedAt(),
}
}
return resultBucketList, nil
}
func convertProtoToBucket(pbBucket *pb.Bucket) (bucket storj.Bucket, err error) {
defaultRS := pbBucket.GetDefaultRedundancyScheme()
defaultEP := pbBucket.GetDefaultEncryptionParameters()
var partnerID uuid.UUID
err = partnerID.UnmarshalJSON(pbBucket.GetPartnerId())
if err != nil && !partnerID.IsZero() {
return bucket, errs.New("Invalid uuid")
}
return storj.Bucket{
Name: string(pbBucket.GetName()),
PartnerID: partnerID,
PathCipher: storj.CipherSuite(pbBucket.GetPathCipher()),
Created: pbBucket.GetCreatedAt(),
DefaultSegmentsSize: pbBucket.GetDefaultSegmentSize(),
DefaultRedundancyScheme: storj.RedundancyScheme{
Algorithm: storj.RedundancyAlgorithm(defaultRS.GetType()),
ShareSize: defaultRS.GetErasureShareSize(),
RequiredShares: int16(defaultRS.GetMinReq()),
RepairShares: int16(defaultRS.GetRepairThreshold()),
OptimalShares: int16(defaultRS.GetSuccessThreshold()),
TotalShares: int16(defaultRS.GetTotal()),
},
DefaultEncryptionParameters: storj.EncryptionParameters{
CipherSuite: storj.CipherSuite(defaultEP.CipherSuite),
BlockSize: int32(defaultEP.BlockSize),
},
}, nil
}
2019-07-16 11:39:23 +01:00
2019-08-06 15:56:23 +01:00
// SetBucketAttributionParams parameters for SetBucketAttribution method
type SetBucketAttributionParams struct {
Bucket string
PartnerID uuid.UUID
}
func (params *SetBucketAttributionParams) toRequest() *pb.BucketSetAttributionRequest {
return &pb.BucketSetAttributionRequest{
Name: []byte(params.Bucket),
PartnerId: params.PartnerID[:],
}
}
// BatchItem returns single item for batch request
func (params *SetBucketAttributionParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_BucketSetAttribution{
BucketSetAttribution: params.toRequest(),
},
}
}
// SetBucketAttribution tries to set the attribution information on the bucket.
func (client *Client) SetBucketAttribution(ctx context.Context, params SetBucketAttributionParams) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.SetBucketAttribution(ctx, params.toRequest())
return Error.Wrap(err)
}
// BeginObjectParams parmaters for BeginObject method
type BeginObjectParams struct {
Bucket []byte
EncryptedPath []byte
Version int32
Redundancy storj.RedundancyScheme
EncryptionParameters storj.EncryptionParameters
ExpiresAt time.Time
}
2019-08-06 15:56:23 +01:00
func (params *BeginObjectParams) toRequest() *pb.ObjectBeginRequest {
return &pb.ObjectBeginRequest{
Bucket: params.Bucket,
EncryptedPath: params.EncryptedPath,
Version: params.Version,
ExpiresAt: params.ExpiresAt,
2019-07-16 11:39:23 +01:00
RedundancyScheme: &pb.RedundancyScheme{
Type: pb.RedundancyScheme_SchemeType(params.Redundancy.Algorithm),
ErasureShareSize: params.Redundancy.ShareSize,
MinReq: int32(params.Redundancy.RequiredShares),
RepairThreshold: int32(params.Redundancy.RepairShares),
SuccessThreshold: int32(params.Redundancy.OptimalShares),
Total: int32(params.Redundancy.TotalShares),
2019-07-16 11:39:23 +01:00
},
EncryptionParameters: &pb.EncryptionParameters{
CipherSuite: pb.CipherSuite(params.EncryptionParameters.CipherSuite),
BlockSize: int64(params.EncryptionParameters.BlockSize),
2019-07-16 11:39:23 +01:00
},
2019-08-06 15:56:23 +01:00
}
}
// BatchItem returns single item for batch request
func (params *BeginObjectParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectBegin{
ObjectBegin: params.toRequest(),
},
}
}
// BeginObjectResponse response for BeginObject request
type BeginObjectResponse struct {
StreamID storj.StreamID
}
func newBeginObjectResponse(response *pb.ObjectBeginResponse) BeginObjectResponse {
return BeginObjectResponse{
StreamID: response.StreamId,
}
}
// BeginObject begins object creation
func (client *Client) BeginObject(ctx context.Context, params BeginObjectParams) (_ storj.StreamID, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.BeginObject(ctx, params.toRequest())
2019-07-16 11:39:23 +01:00
if err != nil {
return nil, Error.Wrap(err)
}
return response.StreamId, nil
}
// CommitObjectParams parmaters for CommitObject method
type CommitObjectParams struct {
StreamID storj.StreamID
EncryptedMetadataNonce storj.Nonce
EncryptedMetadata []byte
}
2019-08-06 15:56:23 +01:00
func (params *CommitObjectParams) toRequest() *pb.ObjectCommitRequest {
return &pb.ObjectCommitRequest{
StreamId: params.StreamID,
EncryptedMetadataNonce: params.EncryptedMetadataNonce,
EncryptedMetadata: params.EncryptedMetadata,
}
}
// BatchItem returns single item for batch request
func (params *CommitObjectParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectCommit{
ObjectCommit: params.toRequest(),
},
}
}
2019-07-16 11:39:23 +01:00
// CommitObject commits created object
func (client *Client) CommitObject(ctx context.Context, params CommitObjectParams) (err error) {
2019-07-16 11:39:23 +01:00
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
_, err = client.client.CommitObject(ctx, params.toRequest())
2019-07-16 11:39:23 +01:00
return Error.Wrap(err)
}
// GetObjectParams parameters for GetObject method
type GetObjectParams struct {
Bucket []byte
EncryptedPath []byte
Version int32
}
2019-08-06 15:56:23 +01:00
func (params *GetObjectParams) toRequest() *pb.ObjectGetRequest {
return &pb.ObjectGetRequest{
Bucket: params.Bucket,
EncryptedPath: params.EncryptedPath,
Version: params.Version,
2019-08-06 15:56:23 +01:00
}
}
2019-08-06 15:56:23 +01:00
// BatchItem returns single item for batch request
func (params *GetObjectParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectGet{
ObjectGet: params.toRequest(),
},
}
2019-08-06 15:56:23 +01:00
}
2019-08-06 15:56:23 +01:00
// GetObjectResponse response for GetObject request
type GetObjectResponse struct {
Info storj.ObjectInfo
}
func newGetObjectResponse(response *pb.ObjectGetResponse) GetObjectResponse {
object := storj.ObjectInfo{
Bucket: string(response.Object.Bucket),
Path: storj.Path(response.Object.EncryptedPath),
StreamID: response.Object.StreamId,
Created: response.Object.CreatedAt,
Modified: response.Object.CreatedAt,
Expires: response.Object.ExpiresAt,
Metadata: response.Object.EncryptedMetadata,
Stream: storj.Stream{
Size: response.Object.TotalSize,
EncryptionParameters: storj.EncryptionParameters{
CipherSuite: storj.CipherSuite(response.Object.EncryptionParameters.CipherSuite),
BlockSize: int32(response.Object.EncryptionParameters.BlockSize),
},
},
}
pbRS := response.Object.RedundancyScheme
if pbRS != nil {
object.Stream.RedundancyScheme = storj.RedundancyScheme{
Algorithm: storj.RedundancyAlgorithm(pbRS.Type),
ShareSize: pbRS.ErasureShareSize,
RequiredShares: int16(pbRS.MinReq),
RepairShares: int16(pbRS.RepairThreshold),
OptimalShares: int16(pbRS.SuccessThreshold),
TotalShares: int16(pbRS.Total),
}
}
2019-08-06 15:56:23 +01:00
return GetObjectResponse{
Info: object,
}
}
// GetObject gets single object
func (client *Client) GetObject(ctx context.Context, params GetObjectParams) (_ storj.ObjectInfo, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.GetObject(ctx, params.toRequest())
2019-08-06 15:56:23 +01:00
if err != nil {
if status.Code(err) == codes.NotFound {
return storj.ObjectInfo{}, storj.ErrObjectNotFound.Wrap(err)
}
return storj.ObjectInfo{}, Error.Wrap(err)
}
getResponse := newGetObjectResponse(response)
return getResponse.Info, nil
}
// BeginDeleteObjectParams parameters for BeginDeleteObject method
type BeginDeleteObjectParams struct {
Bucket []byte
EncryptedPath []byte
Version int32
}
2019-08-06 15:56:23 +01:00
func (params *BeginDeleteObjectParams) toRequest() *pb.ObjectBeginDeleteRequest {
return &pb.ObjectBeginDeleteRequest{
Bucket: params.Bucket,
EncryptedPath: params.EncryptedPath,
Version: params.Version,
}
}
// BatchItem returns single item for batch request
func (params *BeginDeleteObjectParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectBeginDelete{
ObjectBeginDelete: params.toRequest(),
},
}
}
// BeginDeleteObjectResponse response for BeginDeleteObject request
type BeginDeleteObjectResponse struct {
StreamID storj.StreamID
}
func newBeginDeleteObjectResponse(response *pb.ObjectBeginDeleteResponse) BeginDeleteObjectResponse {
return BeginDeleteObjectResponse{
StreamID: response.StreamId,
}
}
2019-07-16 11:39:23 +01:00
// BeginDeleteObject begins object deletion process
func (client *Client) BeginDeleteObject(ctx context.Context, params BeginDeleteObjectParams) (_ storj.StreamID, err error) {
2019-07-16 11:39:23 +01:00
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
response, err := client.client.BeginDeleteObject(ctx, params.toRequest())
2019-07-16 11:39:23 +01:00
if err != nil {
if status.Code(err) == codes.NotFound {
return storj.StreamID{}, storj.ErrObjectNotFound.Wrap(err)
}
2019-07-16 11:39:23 +01:00
return storj.StreamID{}, Error.Wrap(err)
}
return response.StreamId, nil
}
2019-08-06 15:56:23 +01:00
// FinishDeleteObjectParams parameters for FinishDeleteObject method
type FinishDeleteObjectParams struct {
StreamID storj.StreamID
}
func (params *FinishDeleteObjectParams) toRequest() *pb.ObjectFinishDeleteRequest {
return &pb.ObjectFinishDeleteRequest{
StreamId: params.StreamID,
}
}
// BatchItem returns single item for batch request
func (params *FinishDeleteObjectParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectFinishDelete{
ObjectFinishDelete: params.toRequest(),
},
}
}
2019-07-16 11:39:23 +01:00
// FinishDeleteObject finishes object deletion process
2019-08-06 15:56:23 +01:00
func (client *Client) FinishDeleteObject(ctx context.Context, params FinishDeleteObjectParams) (err error) {
2019-07-16 11:39:23 +01:00
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
_, err = client.client.FinishDeleteObject(ctx, params.toRequest())
2019-07-16 11:39:23 +01:00
return Error.Wrap(err)
}
// ListObjectsParams parameters for ListObjects method
type ListObjectsParams struct {
Bucket []byte
EncryptedPrefix []byte
EncryptedCursor []byte
Limit int32
IncludeMetadata bool
Recursive bool
}
2019-08-06 15:56:23 +01:00
func (params *ListObjectsParams) toRequest() *pb.ObjectListRequest {
return &pb.ObjectListRequest{
Bucket: params.Bucket,
EncryptedPrefix: params.EncryptedPrefix,
EncryptedCursor: params.EncryptedCursor,
Limit: params.Limit,
ObjectIncludes: &pb.ObjectListItemIncludes{
Metadata: params.IncludeMetadata,
},
Recursive: params.Recursive,
2019-07-16 11:39:23 +01:00
}
2019-08-06 15:56:23 +01:00
}
// BatchItem returns single item for batch request
func (params *ListObjectsParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_ObjectList{
ObjectList: params.toRequest(),
},
}
}
// ListObjectsResponse response for ListObjects request
type ListObjectsResponse struct {
Items []storj.ObjectListItem
More bool
}
2019-07-16 11:39:23 +01:00
2019-08-06 15:56:23 +01:00
func newListObjectsResponse(response *pb.ObjectListResponse, encryptedPrefix []byte, recursive bool) ListObjectsResponse {
2019-07-16 11:39:23 +01:00
objects := make([]storj.ObjectListItem, len(response.Items))
for i, object := range response.Items {
encryptedPath := object.EncryptedPath
isPrefix := false
2019-08-06 15:56:23 +01:00
if !recursive && len(encryptedPath) != 0 && encryptedPath[len(encryptedPath)-1] == '/' && !bytes.Equal(encryptedPath, encryptedPrefix) {
isPrefix = true
}
2019-07-16 11:39:23 +01:00
objects[i] = storj.ObjectListItem{
EncryptedPath: object.EncryptedPath,
Version: object.Version,
Status: int32(object.Status),
StatusAt: object.StatusAt,
CreatedAt: object.CreatedAt,
ExpiresAt: object.ExpiresAt,
EncryptedMetadataNonce: object.EncryptedMetadataNonce,
EncryptedMetadata: object.EncryptedMetadata,
IsPrefix: isPrefix,
2019-07-16 11:39:23 +01:00
}
}
2019-08-06 15:56:23 +01:00
return ListObjectsResponse{
Items: objects,
More: response.More,
}
}
// ListObjects lists objects according to specific parameters
func (client *Client) ListObjects(ctx context.Context, params ListObjectsParams) (_ []storj.ObjectListItem, more bool, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.ListObjects(ctx, params.toRequest())
if err != nil {
return []storj.ObjectListItem{}, false, Error.Wrap(err)
}
listResponse := newListObjectsResponse(response, params.EncryptedPrefix, params.Recursive)
return listResponse.Items, listResponse.More, Error.Wrap(err)
2019-07-16 11:39:23 +01:00
}
// BeginSegmentParams parameters for BeginSegment method
type BeginSegmentParams struct {
StreamID storj.StreamID
Position storj.SegmentPosition
MaxOderLimit int64
}
2019-08-06 15:56:23 +01:00
func (params *BeginSegmentParams) toRequest() *pb.SegmentBeginRequest {
return &pb.SegmentBeginRequest{
StreamId: params.StreamID,
Position: &pb.SegmentPosition{
PartNumber: params.Position.PartNumber,
Index: params.Position.Index,
},
MaxOrderLimit: params.MaxOderLimit,
2019-08-06 15:56:23 +01:00
}
}
// BatchItem returns single item for batch request
func (params *BeginSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentBegin{
SegmentBegin: params.toRequest(),
},
}
}
// BeginSegmentResponse response for BeginSegment request
type BeginSegmentResponse struct {
SegmentID storj.SegmentID
Limits []*pb.AddressedOrderLimit
PiecePrivateKey storj.PiecePrivateKey
}
func newBeginSegmentResponse(response *pb.SegmentBeginResponse) BeginSegmentResponse {
return BeginSegmentResponse{
SegmentID: response.SegmentId,
Limits: response.AddressedLimits,
PiecePrivateKey: response.PrivateKey,
}
}
// BeginSegment begins segment upload
func (client *Client) BeginSegment(ctx context.Context, params BeginSegmentParams) (_ storj.SegmentID, limits []*pb.AddressedOrderLimit, piecePrivateKey storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.BeginSegment(ctx, params.toRequest())
if err != nil {
return storj.SegmentID{}, nil, storj.PiecePrivateKey{}, Error.Wrap(err)
}
return response.SegmentId, response.AddressedLimits, response.PrivateKey, nil
}
// CommitSegmentParams parameters for CommitSegment method
type CommitSegmentParams struct {
SegmentID storj.SegmentID
Encryption storj.SegmentEncryption
SizeEncryptedData int64
UploadResult []*pb.SegmentPieceUploadResult
}
2019-08-06 15:56:23 +01:00
func (params *CommitSegmentParams) toRequest() *pb.SegmentCommitRequest {
return &pb.SegmentCommitRequest{
SegmentId: params.SegmentID,
EncryptedKeyNonce: params.Encryption.EncryptedKeyNonce,
EncryptedKey: params.Encryption.EncryptedKey,
SizeEncryptedData: params.SizeEncryptedData,
UploadResult: params.UploadResult,
}
2019-08-06 15:56:23 +01:00
}
2019-08-06 15:56:23 +01:00
// BatchItem returns single item for batch request
func (params *CommitSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentCommit{
SegmentCommit: params.toRequest(),
},
}
}
// CommitSegmentNew commits segment after upload
func (client *Client) CommitSegmentNew(ctx context.Context, params CommitSegmentParams) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.CommitSegment(ctx, params.toRequest())
return Error.Wrap(err)
}
// MakeInlineSegmentParams parameters for MakeInlineSegment method
type MakeInlineSegmentParams struct {
StreamID storj.StreamID
Position storj.SegmentPosition
Encryption storj.SegmentEncryption
EncryptedInlineData []byte
}
2019-08-06 15:56:23 +01:00
func (params *MakeInlineSegmentParams) toRequest() *pb.SegmentMakeInlineRequest {
return &pb.SegmentMakeInlineRequest{
StreamId: params.StreamID,
Position: &pb.SegmentPosition{
PartNumber: params.Position.PartNumber,
Index: params.Position.Index,
},
EncryptedKeyNonce: params.Encryption.EncryptedKeyNonce,
EncryptedKey: params.Encryption.EncryptedKey,
EncryptedInlineData: params.EncryptedInlineData,
}
2019-08-06 15:56:23 +01:00
}
2019-08-06 15:56:23 +01:00
// BatchItem returns single item for batch request
func (params *MakeInlineSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentMakeInline{
SegmentMakeInline: params.toRequest(),
},
}
}
// MakeInlineSegment commits segment after upload
func (client *Client) MakeInlineSegment(ctx context.Context, params MakeInlineSegmentParams) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.MakeInlineSegment(ctx, params.toRequest())
return Error.Wrap(err)
}
// BeginDeleteSegmentParams parameters for BeginDeleteSegment method
type BeginDeleteSegmentParams struct {
StreamID storj.StreamID
Position storj.SegmentPosition
}
2019-08-06 15:56:23 +01:00
func (params *BeginDeleteSegmentParams) toRequest() *pb.SegmentBeginDeleteRequest {
return &pb.SegmentBeginDeleteRequest{
StreamId: params.StreamID,
Position: &pb.SegmentPosition{
PartNumber: params.Position.PartNumber,
Index: params.Position.Index,
},
2019-08-06 15:56:23 +01:00
}
}
// BatchItem returns single item for batch request
func (params *BeginDeleteSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentBeginDelete{
SegmentBeginDelete: params.toRequest(),
},
}
}
// BeginDeleteSegmentResponse response for BeginDeleteSegment request
type BeginDeleteSegmentResponse struct {
SegmentID storj.SegmentID
Limits []*pb.AddressedOrderLimit
PiecePrivateKey storj.PiecePrivateKey
}
func newBeginDeleteSegmentResponse(response *pb.SegmentBeginDeleteResponse) BeginDeleteSegmentResponse {
return BeginDeleteSegmentResponse{
SegmentID: response.SegmentId,
Limits: response.AddressedLimits,
PiecePrivateKey: response.PrivateKey,
}
}
// BeginDeleteSegment begins segment upload process
func (client *Client) BeginDeleteSegment(ctx context.Context, params BeginDeleteSegmentParams) (_ storj.SegmentID, limits []*pb.AddressedOrderLimit, _ storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.BeginDeleteSegment(ctx, params.toRequest())
if err != nil {
return storj.SegmentID{}, nil, storj.PiecePrivateKey{}, Error.Wrap(err)
}
return response.SegmentId, response.AddressedLimits, response.PrivateKey, nil
}
// FinishDeleteSegmentParams parameters for FinishDeleteSegment method
type FinishDeleteSegmentParams struct {
SegmentID storj.SegmentID
DeleteResults []*pb.SegmentPieceDeleteResult
}
2019-08-06 15:56:23 +01:00
func (params *FinishDeleteSegmentParams) toRequest() *pb.SegmentFinishDeleteRequest {
return &pb.SegmentFinishDeleteRequest{
SegmentId: params.SegmentID,
Results: params.DeleteResults,
}
}
// BatchItem returns single item for batch request
func (params *FinishDeleteSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentFinishDelete{
SegmentFinishDelete: params.toRequest(),
},
}
}
// FinishDeleteSegment finishes segment upload process
func (client *Client) FinishDeleteSegment(ctx context.Context, params FinishDeleteSegmentParams) (err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
_, err = client.client.FinishDeleteSegment(ctx, params.toRequest())
return Error.Wrap(err)
}
// DownloadSegmentParams parameters for DownloadSegment method
type DownloadSegmentParams struct {
StreamID storj.StreamID
Position storj.SegmentPosition
}
2019-08-06 15:56:23 +01:00
func (params *DownloadSegmentParams) toRequest() *pb.SegmentDownloadRequest {
return &pb.SegmentDownloadRequest{
StreamId: params.StreamID,
CursorPosition: &pb.SegmentPosition{
PartNumber: params.Position.PartNumber,
Index: params.Position.Index,
},
}
2019-08-06 15:56:23 +01:00
}
// BatchItem returns single item for batch request
func (params *DownloadSegmentParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentDownload{
SegmentDownload: params.toRequest(),
},
}
}
// DownloadSegmentResponse response for DownloadSegment request
type DownloadSegmentResponse struct {
Info storj.SegmentDownloadInfo
2019-08-06 15:56:23 +01:00
Limits []*pb.AddressedOrderLimit
}
func newDownloadSegmentResponse(response *pb.SegmentDownloadResponse) DownloadSegmentResponse {
info := storj.SegmentDownloadInfo{
SegmentID: response.SegmentId,
Size: response.SegmentSize,
EncryptedInlineData: response.EncryptedInlineData,
PiecePrivateKey: response.PrivateKey,
SegmentEncryption: storj.SegmentEncryption{
EncryptedKeyNonce: response.EncryptedKeyNonce,
EncryptedKey: response.EncryptedKey,
},
}
if response.Next != nil {
info.Next = storj.SegmentPosition{
PartNumber: response.Next.PartNumber,
Index: response.Next.Index,
}
}
for i := range response.AddressedLimits {
if response.AddressedLimits[i].Limit == nil {
response.AddressedLimits[i] = nil
}
}
2019-08-06 15:56:23 +01:00
return DownloadSegmentResponse{
Info: info,
Limits: response.AddressedLimits,
}
}
// DownloadSegment gets info for downloading remote segment or data from inline segment
func (client *Client) DownloadSegment(ctx context.Context, params DownloadSegmentParams) (_ storj.SegmentDownloadInfo, _ []*pb.AddressedOrderLimit, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.DownloadSegment(ctx, params.toRequest())
if err != nil {
return storj.SegmentDownloadInfo{}, nil, Error.Wrap(err)
}
2019-08-06 15:56:23 +01:00
downloadResponse := newDownloadSegmentResponse(response)
return downloadResponse.Info, downloadResponse.Limits, nil
}
// ListSegmentsParams parameters for ListSegment method
type ListSegmentsParams struct {
StreamID storj.StreamID
CursorPosition storj.SegmentPosition
Limit int32
}
2019-08-06 15:56:23 +01:00
// ListSegmentsResponse response for ListSegments request
type ListSegmentsResponse struct {
Items []storj.SegmentListItem
More bool
}
2019-08-06 15:56:23 +01:00
func (params *ListSegmentsParams) toRequest() *pb.SegmentListRequest {
return &pb.SegmentListRequest{
StreamId: params.StreamID,
CursorPosition: &pb.SegmentPosition{
PartNumber: params.CursorPosition.PartNumber,
Index: params.CursorPosition.Index,
},
Limit: params.Limit,
}
2019-08-06 15:56:23 +01:00
}
// BatchItem returns single item for batch request
func (params *ListSegmentsParams) BatchItem() *pb.BatchRequestItem {
return &pb.BatchRequestItem{
Request: &pb.BatchRequestItem_SegmentList{
SegmentList: params.toRequest(),
},
}
}
2019-08-06 15:56:23 +01:00
func newListSegmentsResponse(response *pb.SegmentListResponse) ListSegmentsResponse {
items := make([]storj.SegmentListItem, len(response.Items))
for i, responseItem := range response.Items {
items[i] = storj.SegmentListItem{
Position: storj.SegmentPosition{
PartNumber: responseItem.Position.PartNumber,
Index: responseItem.Position.Index,
},
}
}
2019-08-06 15:56:23 +01:00
return ListSegmentsResponse{
Items: items,
More: response.More,
}
}
2019-08-06 15:56:23 +01:00
// ListSegmentsNew lists object segments
func (client *Client) ListSegmentsNew(ctx context.Context, params ListSegmentsParams) (_ []storj.SegmentListItem, more bool, err error) {
defer mon.Task()(&ctx)(&err)
response, err := client.client.ListSegments(ctx, params.toRequest())
if err != nil {
return []storj.SegmentListItem{}, false, Error.Wrap(err)
}
listResponse := newListSegmentsResponse(response)
return listResponse.Items, listResponse.More, Error.Wrap(err)
}
2019-08-06 15:56:23 +01:00
// Batch sends multiple requests in one batch
func (client *Client) Batch(ctx context.Context, requests ...BatchItem) (resp []BatchResponse, err error) {
defer mon.Task()(&ctx)(&err)
2019-08-06 15:56:23 +01:00
batchItems := make([]*pb.BatchRequestItem, len(requests))
for i, request := range requests {
batchItems[i] = request.BatchItem()
}
response, err := client.client.Batch(ctx, &pb.BatchRequest{
Requests: batchItems,
})
2019-08-06 15:56:23 +01:00
if err != nil {
return []BatchResponse{}, err
}
2019-08-06 15:56:23 +01:00
resp = make([]BatchResponse, len(response.Responses))
for i, response := range response.Responses {
resp[i] = BatchResponse{
pbRequest: batchItems[i].Request,
pbResponse: response.Response,
}
}
return resp, nil
}