uploader

Documentation

Index

• func GCPBlockDataObjectName(computationResult *execution.ComputationResult) string
• func WriteComputationResultsTo(computationResult *execution.ComputationResult, writer io.Writer) error
• type AsyncUploader
  • func NewAsyncUploader(uploader Uploader, retryInitialTimeout time.Duration, maxRetryNumber uint64, ...) *AsyncUploader
  • func (a *AsyncUploader) SetOnCompleteCallback(onComplete OnCompleteFunc)
  • func (a *AsyncUploader) Upload(computationResult *execution.ComputationResult) error
  • func (a *AsyncUploader) UploadTask(ctx context.Context, computationResult *execution.ComputationResult)
  • func (a *AsyncUploader) UploadWorker(ctx irrecoverable.SignalerContext, ready component.ReadyFunc)
• type BadgerRetryableUploaderWrapper
  • func NewBadgerRetryableUploaderWrapper(uploader *AsyncUploader, blocks storage.Blocks, commits storage.Commits, ...) *BadgerRetryableUploaderWrapper
  • func (b *BadgerRetryableUploaderWrapper) RetryUpload() error
  • func (b *BadgerRetryableUploaderWrapper) Upload(computationResult *execution.ComputationResult) error
• type BlockData
  • func ComputationResultToBlockData(computationResult *execution.ComputationResult) *BlockData
• type FileUploader
  • func NewFileUploader(dir string) *FileUploader
  • func (f *FileUploader) Upload(computationResult *execution.ComputationResult) error
• type GCPBucketUploader
  • func NewGCPBucketUploader(ctx context.Context, bucketName string, log zerolog.Logger) (*GCPBucketUploader, error)
  • func (u *GCPBucketUploader) Upload(computationResult *execution.ComputationResult) error
• type Manager
  • func NewManager(tracer module.Tracer) *Manager
  • func (m *Manager) AddUploader(uploader Uploader)
  • func (m *Manager) Enabled() bool
  • func (m *Manager) RetryUploads() (err error)
  • func (m *Manager) SetEnabled(enabled bool)
  • func (m *Manager) Upload(ctx context.Context, result *execution.ComputationResult) error
• type OnCompleteFunc
• type RetryableUploaderWrapper
• type S3Uploader
  • func NewS3Uploader(ctx context.Context, client *s3.Client, bucket string, log zerolog.Logger) *S3Uploader
  • func (u *S3Uploader) Upload(result *execution.ComputationResult) error
• type Uploader

Functions

func GCPBlockDataObjectName

func GCPBlockDataObjectName(computationResult *execution.ComputationResult) string

func WriteComputationResultsTo

func WriteComputationResultsTo(computationResult *execution.ComputationResult, writer io.Writer) error

Types

type AsyncUploader

type AsyncUploader struct {
	component.Component
	// contains filtered or unexported fields
}

AsyncUploader wraps up another Uploader instance and make its upload asynchronous

func NewAsyncUploader

func NewAsyncUploader(uploader Uploader,
	retryInitialTimeout time.Duration,
	maxRetryNumber uint64,
	log zerolog.Logger,
	metrics module.ExecutionMetrics) *AsyncUploader

func (*AsyncUploader) SetOnCompleteCallback

func (a *AsyncUploader) SetOnCompleteCallback(onComplete OnCompleteFunc)

func (*AsyncUploader) Upload

func (a *AsyncUploader) Upload(computationResult *execution.ComputationResult) error

Upload adds the computation result to a queue to be processed asynchronously by workers, ensuring that multiple uploads can be run in parallel. No errors expected during normal operation.

func (*AsyncUploader) UploadTask

func (a *AsyncUploader) UploadTask(ctx context.Context, computationResult *execution.ComputationResult)

UploadTask implements retrying for uploading computation results. When the upload is complete, the callback will be called with the result (for example, to record that the upload was successful) and any error. No errors expected during normal operation.

func (*AsyncUploader) UploadWorker

func (a *AsyncUploader) UploadWorker(ctx irrecoverable.SignalerContext, ready component.ReadyFunc)

UploadWorker implements a component worker which asynchronously uploads computation results from the execution node (after a block is executed) to storage such as a GCP bucket or S3 bucket.

type BadgerRetryableUploaderWrapper

type BadgerRetryableUploaderWrapper struct {
	component.Component
	// contains filtered or unexported fields
}

BadgerRetryableUploaderWrapper is the BadgerDB based implementation to RetryableUploaderWrapper

func NewBadgerRetryableUploaderWrapper

func NewBadgerRetryableUploaderWrapper(
	uploader *AsyncUploader,
	blocks storage.Blocks,
	commits storage.Commits,
	collections storage.Collections,
	events storage.Events,
	results storage.ExecutionResults,
	transactionResults storage.TransactionResults,
	uploadStatusStore storage.ComputationResultUploadStatus,
	execDataDownloader execution_data.Downloader,
	metrics module.ExecutionMetrics) *BadgerRetryableUploaderWrapper

func (*BadgerRetryableUploaderWrapper) RetryUpload

func (b *BadgerRetryableUploaderWrapper) RetryUpload() error

func (*BadgerRetryableUploaderWrapper) Upload

func (b *BadgerRetryableUploaderWrapper) Upload(computationResult *execution.ComputationResult) error

type BlockData

type BlockData struct {
	Block                *flow.Block
	Collections          []*entity.CompleteCollection
	TxResults            []*flow.TransactionResult
	Events               []*flow.Event
	TrieUpdates          []*ledger.TrieUpdate
	FinalStateCommitment flow.StateCommitment
}

func ComputationResultToBlockData

func ComputationResultToBlockData(computationResult *execution.ComputationResult) *BlockData

type FileUploader

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

func NewFileUploader

func NewFileUploader(dir string) *FileUploader

func (*FileUploader) Upload

func (f *FileUploader) Upload(computationResult *execution.ComputationResult) error

type GCPBucketUploader

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

func NewGCPBucketUploader

func NewGCPBucketUploader(ctx context.Context, bucketName string, log zerolog.Logger) (*GCPBucketUploader, error)

func (*GCPBucketUploader) Upload

func (u *GCPBucketUploader) Upload(computationResult *execution.ComputationResult) error

Upload uploads the computation result to the configured GCP bucket. All errors returned from this function can be considered benign.

type Manager

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

Manager encapsulates the logic for uploading computation results to cloud storage.

func NewManager

func NewManager(tracer module.Tracer) *Manager

NewManager creates a new uploader manager

func (*Manager) AddUploader

func (m *Manager) AddUploader(uploader Uploader)

AddUploader adds an uploader to the manager

func (*Manager) Enabled

func (m *Manager) Enabled() bool

Enabled returns whether the manager is enabled

func (*Manager) RetryUploads

func (m *Manager) RetryUploads() (err error)

RetryUploads retries uploads for all uploaders that implement RetryableUploaderWrapper Any errors returned by the uploaders may be considered benign

func (*Manager) SetEnabled

func (m *Manager) SetEnabled(enabled bool)

SetEnabled enables or disables the manager

func (*Manager) Upload

func (m *Manager) Upload(
	ctx context.Context,
	result *execution.ComputationResult,
) error

Upload uploads the given computation result with all uploaders Any errors returned by the uploaders may be considered benign

type OnCompleteFunc

type OnCompleteFunc func(*execution.ComputationResult, error)

OnCompleteFunc is the type of function being called at upload completion.

type RetryableUploaderWrapper

type RetryableUploaderWrapper interface {
	Uploader
	RetryUpload() error
}

RetryableUploaderWrapper defines the interface for uploader that is retryable

type S3Uploader

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

S3Uploader is a S3 implementation of the uploader interface.

func NewS3Uploader

func NewS3Uploader(ctx context.Context, client *s3.Client, bucket string, log zerolog.Logger) *S3Uploader

NewS3Uploader returns a new S3 uploader instance.

func (*S3Uploader) Upload

func (u *S3Uploader) Upload(result *execution.ComputationResult) error

Upload uploads the given computation result to the configured S3 bucket.

type Uploader

type Uploader interface {
	Upload(computationResult *execution.ComputationResult) error
}