gracefulexit

Documentation

Index

• Variables
• type Chore
  • func NewChore(log *zap.Logger, db DB, overlay overlay.DB, metaLoop *metainfo.Loop, ...) *Chore
  • func (chore *Chore) Close() error
  • func (chore *Chore) Run(ctx context.Context) (err error)
• type Config
• type DB
• type Endpoint
  • func NewEndpoint(log *zap.Logger, signer signing.Signer, db DB, overlaydb overlay.DB, ...) *Endpoint
  • func (endpoint *Endpoint) DRPC() pb.DRPCSatelliteGracefulExitServer
  • func (endpoint *Endpoint) Process(stream pb.SatelliteGracefulExit_ProcessServer) (err error)
• type PathCollector
  • func NewPathCollector(db DB, nodeIDs storj.NodeIDList, log *zap.Logger, batchSize int) *PathCollector
  • func (collector *PathCollector) Flush(ctx context.Context) (err error)
  • func (collector *PathCollector) InlineSegment(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)
  • func (collector *PathCollector) Object(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)
  • func (collector *PathCollector) RemoteSegment(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)
• type Progress
• type TransferQueueItem

Variables

var (
	// Error is the default error class for graceful exit package.
	Error = errs.Class("gracefulexit")

	// ErrNodeNotFound is returned if a graceful exit entry for a  node does not exist in database
	ErrNodeNotFound = errs.Class("graceful exit node not found")

	// ErrAboveOptimalThreshold is returned if a graceful exit entry for a node has more pieces than required.
	ErrAboveOptimalThreshold = errs.Class("pointer has more pieces than required")
)

var (
	// ErrInvalidArgument is an error class for invalid argument errors used to check which rpc code to use.
	ErrInvalidArgument = errs.Class("graceful exit")
)

Types

type Chore

type Chore struct {
	Loop sync2.Cycle
	// contains filtered or unexported fields
}

Chore populates the graceful exit transfer queue.

architecture: Chore

func NewChore

func NewChore(log *zap.Logger, db DB, overlay overlay.DB, metaLoop *metainfo.Loop, config Config) *Chore

NewChore instantiates Chore.

func (*Chore) Close

func (chore *Chore) Close() error

Close closes chore.

func (*Chore) Run

func (chore *Chore) Run(ctx context.Context) (err error)

Run starts the chore.

type Config

type Config struct {
	Enabled bool `help:"whether or not graceful exit is enabled on the satellite side." releaseDefault:"false" devDefault:"true"`

	ChoreBatchSize int           `help:"size of the buffer used to batch inserts into the transfer queue." default:"500"`
	ChoreInterval  time.Duration `help:"how often to run the transfer queue chore." releaseDefault:"30s" devDefault:"10s"`

	EndpointBatchSize int `help:"size of the buffer used to batch transfer queue reads and sends to the storage node." default:"100"`

	MaxFailuresPerPiece          int           `help:"maximum number of transfer failures per piece." default:"3"`
	OverallMaxFailuresPercentage int           `help:"maximum percentage of transfer failures per node." default:"10"`
	MaxInactiveTimeFrame         time.Duration `help:"maximum inactive time frame of transfer activities per node." default:"500h"`
	RecvTimeout                  time.Duration `help:"the minimum duration for receiving a stream from a storage node before timing out" default:"10m"`
	MaxOrderLimitSendCount       int           `help:"maximum number of order limits a satellite sends to a node before marking piece transfer failed" default:"5"`
}

Config for the chore

type DB

type DB interface {
	// IncrementProgress increments transfer stats for a node.
	IncrementProgress(ctx context.Context, nodeID storj.NodeID, bytes int64, successfulTransfers int64, failedTransfers int64) error
	// GetProgress gets a graceful exit progress entry.
	GetProgress(ctx context.Context, nodeID storj.NodeID) (*Progress, error)

	// Enqueue batch inserts graceful exit transfer queue entries it does not exist.
	Enqueue(ctx context.Context, items []TransferQueueItem) error
	// UpdateTransferQueueItem creates a graceful exit transfer queue entry.
	UpdateTransferQueueItem(ctx context.Context, item TransferQueueItem) error
	// DeleteTransferQueueItem deletes a graceful exit transfer queue entry.
	DeleteTransferQueueItem(ctx context.Context, nodeID storj.NodeID, path []byte, pieceNum int32) error
	// DeleteTransferQueueItem deletes a graceful exit transfer queue entries by nodeID.
	DeleteTransferQueueItems(ctx context.Context, nodeID storj.NodeID) error
	// DeleteFinishedTransferQueueItem deletes finiahed graceful exit transfer queue entries.
	DeleteFinishedTransferQueueItems(ctx context.Context, nodeID storj.NodeID) error
	// GetTransferQueueItem gets a graceful exit transfer queue entry.
	GetTransferQueueItem(ctx context.Context, nodeID storj.NodeID, path []byte, pieceNum int32) (*TransferQueueItem, error)
	// GetIncomplete gets incomplete graceful exit transfer queue entries ordered by durability ratio and queued date ascending.
	GetIncomplete(ctx context.Context, nodeID storj.NodeID, limit int, offset int64) ([]*TransferQueueItem, error)
	// GetIncompleteNotFailed gets incomplete graceful exit transfer queue entries in the database ordered by durability ratio and queued date ascending.
	GetIncompleteNotFailed(ctx context.Context, nodeID storj.NodeID, limit int, offset int64) ([]*TransferQueueItem, error)
	// GetIncompleteNotFailed gets incomplete graceful exit transfer queue entries that have failed <= maxFailures times, ordered by durability ratio and queued date ascending.
	GetIncompleteFailed(ctx context.Context, nodeID storj.NodeID, maxFailures int, limit int, offset int64) ([]*TransferQueueItem, error)
	// IncrementOrderLimitSendCount increments the number of times a node has been sent an order limit for transferring.
	IncrementOrderLimitSendCount(ctx context.Context, nodeID storj.NodeID, path []byte, pieceNum int32) error
}

DB implements CRUD operations for graceful exit service

architecture: Database

type Endpoint

type Endpoint struct {
	// contains filtered or unexported fields
}

Endpoint for handling the transfer of pieces for Graceful Exit.

func NewEndpoint

func NewEndpoint(log *zap.Logger, signer signing.Signer, db DB, overlaydb overlay.DB, overlay *overlay.Service, metainfo *metainfo.Service, orders *orders.Service,
	peerIdentities overlay.PeerIdentities, config Config) *Endpoint

NewEndpoint creates a new graceful exit endpoint.

func (*Endpoint) DRPC

func (endpoint *Endpoint) DRPC() pb.DRPCSatelliteGracefulExitServer

DRPC returns a DRPC form of the endpoint.

func (*Endpoint) Process

func (endpoint *Endpoint) Process(stream pb.SatelliteGracefulExit_ProcessServer) (err error)

Process is called by storage nodes to receive pieces to transfer to new nodes and get exit status.

type PathCollector

type PathCollector struct {
	// contains filtered or unexported fields
}

PathCollector uses the metainfo loop to add paths to node reservoirs

architecture: Observer

func NewPathCollector

func NewPathCollector(db DB, nodeIDs storj.NodeIDList, log *zap.Logger, batchSize int) *PathCollector

NewPathCollector instantiates a path collector.

func (*PathCollector) Flush

func (collector *PathCollector) Flush(ctx context.Context) (err error)

Flush persists the current buffer items to the database.

func (*PathCollector) InlineSegment

func (collector *PathCollector) InlineSegment(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)

InlineSegment returns nil because we're only auditing for storage nodes for now

func (*PathCollector) Object

func (collector *PathCollector) Object(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)

Object returns nil because the audit service does not interact with objects

func (*PathCollector) RemoteSegment

func (collector *PathCollector) RemoteSegment(ctx context.Context, path metainfo.ScopedPath, pointer *pb.Pointer) (err error)

RemoteSegment takes a remote segment found in metainfo and creates a graceful exit transfer queue item if it doesn't exist already

type Progress

type Progress struct {
	NodeID            storj.NodeID
	BytesTransferred  int64
	PiecesTransferred int64
	PiecesFailed      int64
	UpdatedAt         time.Time
}

Progress represents the persisted graceful exit progress record.

type TransferQueueItem

type TransferQueueItem struct {
	NodeID              storj.NodeID
	Path                []byte
	PieceNum            int32
	RootPieceID         storj.PieceID
	DurabilityRatio     float64
	QueuedAt            time.Time
	RequestedAt         *time.Time
	LastFailedAt        *time.Time
	LastFailedCode      *int
	FailedCount         *int
	FinishedAt          *time.Time
	OrderLimitSendCount int
}

TransferQueueItem represents the persisted graceful exit queue record.