s3coordinator

Documentation

Index

• type CoordinatorS3
  • func NewS3(bucket string, cfgs ...*aws.Config) (*CoordinatorS3, error)
  • func NewS3Recipe(bucket string) (*CoordinatorS3, error)
  • func (c *CoordinatorS3) AssignOperationTablePart(operationID string, workerIndex int) (*model.OperationTablePart, error)
  • func (c *CoordinatorS3) ClearAssignedTablesParts(ctx context.Context, operationID string, workerIndex int) (int64, error)
  • func (c *CoordinatorS3) CreateOperationTablesParts(operationID string, tables []*model.OperationTablePart) error
  • func (c *CoordinatorS3) CreateOperationWorkers(operationID string, workersCount int) error
  • func (c *CoordinatorS3) FinishOperation(operationID string, shardIndex int, taskErr error) error
  • func (c *CoordinatorS3) GetOperationProgress(operationID string) (*model.AggregatedProgress, error)
  • func (c *CoordinatorS3) GetOperationTablesParts(operationID string) ([]*model.OperationTablePart, error)
  • func (c *CoordinatorS3) GetOperationWorkers(operationID string) ([]*model.OperationWorker, error)
  • func (c *CoordinatorS3) GetOperationWorkersCount(operationID string, completed bool) (int, error)
  • func (c *CoordinatorS3) GetTransferState(transferID string) (map[string]*coordinator.TransferStateData, error)
  • func (c *CoordinatorS3) RemoveTransferState(transferID string, keys []string) error
  • func (c *CoordinatorS3) SetTransferState(transferID string, state map[string]*coordinator.TransferStateData) error
  • func (c *CoordinatorS3) UpdateOperationTablesParts(operationID string, tables []*model.OperationTablePart) error

Types

type CoordinatorS3

type CoordinatorS3 struct {
	*coordinator.CoordinatorNoOp
	// contains filtered or unexported fields
}

func NewS3

func NewS3(bucket string, cfgs ...*aws.Config) (*CoordinatorS3, error)

NewS3 creates a new CoordinatorS3 with AWS SDK v1.

func NewS3Recipe

func NewS3Recipe(bucket string) (*CoordinatorS3, error)

func (*CoordinatorS3) AssignOperationTablePart

func (c *CoordinatorS3) AssignOperationTablePart(operationID string, workerIndex int) (*model.OperationTablePart, error)

AssignOperationTablePart assigns a table part to a worker.

func (*CoordinatorS3) ClearAssignedTablesParts

func (c *CoordinatorS3) ClearAssignedTablesParts(ctx context.Context, operationID string, workerIndex int) (int64, error)

ClearAssignedTablesParts clears the table parts assigned to a worker.

func (*CoordinatorS3) CreateOperationTablesParts

func (c *CoordinatorS3) CreateOperationTablesParts(operationID string, tables []*model.OperationTablePart) error

CreateOperationTablesParts creates table parts for an operation and stores them in S3.

func (*CoordinatorS3) CreateOperationWorkers

func (c *CoordinatorS3) CreateOperationWorkers(operationID string, workersCount int) error

CreateOperationWorkers creates worker files with initial progress for the operation.

func (*CoordinatorS3) FinishOperation

func (c *CoordinatorS3) FinishOperation(operationID string, shardIndex int, taskErr error) error

func (*CoordinatorS3) GetOperationProgress

func (c *CoordinatorS3) GetOperationProgress(operationID string) (*model.AggregatedProgress, error)

GetOperationProgress do nothing

func (*CoordinatorS3) GetOperationTablesParts

func (c *CoordinatorS3) GetOperationTablesParts(operationID string) ([]*model.OperationTablePart, error)

GetOperationTablesParts fetches table parts for a given operation.

func (*CoordinatorS3) GetOperationWorkers

func (c *CoordinatorS3) GetOperationWorkers(operationID string) ([]*model.OperationWorker, error)

GetOperationWorkers fetches all workers for the given operationID.

func (*CoordinatorS3) GetOperationWorkersCount

func (c *CoordinatorS3) GetOperationWorkersCount(operationID string, completed bool) (int, error)

GetOperationWorkersCount returns the count of workers either completed or not completed.

func (*CoordinatorS3) GetTransferState

func (c *CoordinatorS3) GetTransferState(transferID string) (map[string]*coordinator.TransferStateData, error)

GetTransferState fetches all state objects with a given transferID (prefix).

func (*CoordinatorS3) RemoveTransferState

func (c *CoordinatorS3) RemoveTransferState(transferID string, keys []string) error

RemoveTransferState removes the specified state keys for a given transferID from S3.

func (*CoordinatorS3) SetTransferState

func (c *CoordinatorS3) SetTransferState(transferID string, state map[string]*coordinator.TransferStateData) error

SetTransferState stores the given transfer state into S3 as JSON objects.

func (*CoordinatorS3) UpdateOperationTablesParts

func (c *CoordinatorS3) UpdateOperationTablesParts(operationID string, tables []*model.OperationTablePart) error

UpdateOperationTablesParts updates the status of table parts in S3.