storj/uplink/piecestore/client.go
Ivan Fraixedes bf97ef06fc
storagenode: Add new endpoint to receive satellite requests for… (#3590)
* pkg/pg: Add new service function storage node

  Add a new service function to the storage node piece store for deleting
  pieces when satellites request them.

* storagenode/piecestore: Add endpoint to delete piece

  Add a new endpoint to receive from trusted satellites to delete a piece.

* private/testplanet: Fix storagenode mock

  Add to the storagenode mock the new endpoint method.

* proto.lock: Update it with the last protbuff changes

* storagenode/piecestore: Reuse test piece upload

  Extract the repeated logic from several tests functions for uploading a
  test piece to a test helper function.

* uplink/piecestore: Implement client side method

  Implement the client side method of the new piecestore RPC function.

* storagenode/piecestore: Add test DeletePiece endpoint

  Implement a test for the DeletePiece new endpoint method.
2019-11-26 18:47:19 +01:00

116 lines
3.0 KiB
Go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package piecestore
import (
"context"
"io"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/storj/pkg/identity"
"storj.io/storj/pkg/pb"
"storj.io/storj/pkg/rpc"
"storj.io/storj/pkg/storj"
"storj.io/storj/private/memory"
)
// Error is the default error class for piecestore client.
var Error = errs.Class("piecestore")
// Config defines piecestore client parameters for upload and download.
type Config struct {
UploadBufferSize int64
DownloadBufferSize int64
InitialStep int64
MaximumStep int64
}
// DefaultConfig are the default params used for upload and download.
var DefaultConfig = Config{
UploadBufferSize: 256 * memory.KiB.Int64(),
DownloadBufferSize: 256 * memory.KiB.Int64(),
InitialStep: 64 * memory.KiB.Int64(),
MaximumStep: 1 * memory.MiB.Int64(),
}
// Client implements uploading, downloading and deleting content from a piecestore.
type Client struct {
log *zap.Logger
client rpc.PiecestoreClient
conn *rpc.Conn
config Config
}
// Dial dials the target piecestore endpoint.
func Dial(ctx context.Context, dialer rpc.Dialer, target *pb.Node, log *zap.Logger, config Config) (*Client, error) {
conn, err := dialer.DialNode(ctx, target)
if err != nil {
return nil, Error.Wrap(err)
}
return &Client{
log: log,
client: conn.PiecestoreClient(),
conn: conn,
config: config,
}, nil
}
// Delete uses delete order limit to delete a piece on piece store.
func (client *Client) Delete(ctx context.Context, limit *pb.OrderLimit, privateKey storj.PiecePrivateKey) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.Delete(ctx, &pb.PieceDeleteRequest{
Limit: limit,
})
return Error.Wrap(err)
}
// DeletePiece deletes a piece.
func (client *Client) DeletePiece(ctx context.Context, id storj.PieceID) (err error) {
defer mon.Task()(&ctx, id.String())(&err)
_, err = client.client.DeletePiece(ctx, &pb.PieceDeletePieceRequest{
PieceId: id,
})
return Error.Wrap(err)
}
// Retain uses a bloom filter to tell the piece store which pieces to keep.
func (client *Client) Retain(ctx context.Context, req *pb.RetainRequest) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = client.client.Retain(ctx, req)
return Error.Wrap(err)
}
// Close closes the underlying connection.
func (client *Client) Close() error {
return client.conn.Close()
}
// GetPeerIdentity gets the connection's peer identity
func (client *Client) GetPeerIdentity() (*identity.PeerIdentity, error) {
return client.conn.PeerIdentity()
}
// next allocation step find the next trusted step.
func (client *Client) nextAllocationStep(previous int64) int64 {
// TODO: ensure that this is frame idependent
next := previous * 3 / 2
if next > client.config.MaximumStep {
next = client.config.MaximumStep
}
return next
}
// ignoreEOF is an utility func for ignoring EOF error, when it's not important.
func ignoreEOF(err error) error {
if err == io.EOF {
return nil
}
return err
}