catalog

Documentation

Index

• Constants
• func IsNotFound(err error) bool
• func NewWriterProcessor(catalogClient Client) workqueue.Processor
• type AsyncClient
• type AsyncClientImpl
  • func NewAsyncClient(client Client, cfg Config, scope promutils.Scope) (AsyncClientImpl, error)
  • func (c AsyncClientImpl) Download(ctx context.Context, requests ...DownloadRequest) (outputFuture DownloadFuture, err error)
  • func (c AsyncClientImpl) Start(ctx context.Context) error
  • func (c AsyncClientImpl) Upload(ctx context.Context, requests ...UploadRequest) (putFuture UploadFuture, err error)
• type Client
• type Config
  • func GetConfig() *Config
  • func (cfg Config) GetPFlagSet(prefix string) *pflag.FlagSet
• type DownloadFuture
• type DownloadRequest
• type DownloadResponse
• type Future
• type Key
  • func (k Key) String() string
• type Metadata
• type ReaderProcessor
  • func NewReaderProcessor(catalogClient Client) ReaderProcessor
  • func (p ReaderProcessor) Process(ctx context.Context, workItem workqueue.WorkItem) (workqueue.WorkStatus, error)
• type ReaderWorkItem
  • func NewReaderWorkItem(key Key, outputsWriter io.OutputWriter) *ReaderWorkItem
  • func (item ReaderWorkItem) IsCached() bool
• type ReadyHandler
• type ResponseStatus
• type UploadFuture
• type UploadRequest
• type WriterWorkItem
  • func NewWriterWorkItem(key Key, data io.OutputReader, metadata Metadata) *WriterWorkItem

Constants

const (
	ErrResponseNotReady errors.ErrorCode = "RESPONSE_NOT_READY"
	ErrSystemError      errors.ErrorCode = "SYSTEM_ERROR"
)

Functions

func IsNotFound

func IsNotFound(err error) bool

func NewWriterProcessor

func NewWriterProcessor(catalogClient Client) workqueue.Processor

Types

type AsyncClient

type AsyncClient interface {
	// Returns if an entry exists for the given task and input. It returns the data as a LiteralMap
	Download(ctx context.Context, requests ...DownloadRequest) (outputFuture DownloadFuture, err error)

	// Adds a new entry to catalog for the given task execution context and the generated output
	Upload(ctx context.Context, requests ...UploadRequest) (putFuture UploadFuture, err error)
}

An interface that helps async interaction with catalog service

type AsyncClientImpl

type AsyncClientImpl struct {
	Reader workqueue.IndexedWorkQueue
	Writer workqueue.IndexedWorkQueue
}

An async-client for catalog that can queue download and upload requests on workqueues.

func NewAsyncClient

func NewAsyncClient(client Client, cfg Config, scope promutils.Scope) (AsyncClientImpl, error)

func (AsyncClientImpl) Download

func (c AsyncClientImpl) Download(ctx context.Context, requests ...DownloadRequest) (outputFuture DownloadFuture, err error)

func (AsyncClientImpl) Start

func (c AsyncClientImpl) Start(ctx context.Context) error

func (AsyncClientImpl) Upload

func (c AsyncClientImpl) Upload(ctx context.Context, requests ...UploadRequest) (putFuture UploadFuture, err error)

type Client

type Client interface {
	Get(ctx context.Context, key Key) (io.OutputReader, error)
	Put(ctx context.Context, key Key, reader io.OutputReader, metadata Metadata) error
}

Default Catalog client that allows memoization and indexing of intermediate data in Flyte

type Config

type Config struct {
	ReaderWorkqueueConfig workqueue.Config
	WriterWorkqueueConfig workqueue.Config
}

func GetConfig

func GetConfig() *Config

func (Config) GetPFlagSet

func (cfg Config) GetPFlagSet(prefix string) *pflag.FlagSet

GetPFlagSet will return strongly types pflags for all fields in Config and its nested types. The format of the flags is json-name.json-sub-name... etc.

type DownloadFuture

type DownloadFuture interface {
	Future

	// Gets the actual response from the future. This will return an error if the future isn't ready yet.
	GetResponse() (DownloadResponse, error)
}

Catalog download future to represent async process of downloading catalog artifacts.

type DownloadRequest

type DownloadRequest struct {
	Key    Key
	Target io.OutputWriter
}

Catalog Download Request to represent async operation download request.

type DownloadResponse

type DownloadResponse interface {
	// Gets a bit set representing which items from the request were cached.
	GetCachedResults() *bitarray.BitSet

	// Gets the total size of the cached result.
	GetResultsSize() int

	// A convenience method to retrieve the number of cached items.
	GetCachedCount() int
}

Catalog download response.

type Future

type Future interface {
	// Gets the response status for the future. If the future represents multiple operations, the status will only be
	// ready if all of them are.
	GetResponseStatus() ResponseStatus

	// Sets a callback handler to be called when the future status changes to ready.
	OnReady(handler ReadyHandler)

	GetResponseError() error
}

A generic Future interface to represent async operations results

type Key

type Key struct {
	Identifier     core.Identifier
	CacheVersion   string
	TypedInterface core.TypedInterface
	InputReader    io.InputReader
}

An identifier for a catalog object.

func (Key) String

func (k Key) String() string

type Metadata

type Metadata struct {
	WorkflowExecutionIdentifier *core.WorkflowExecutionIdentifier
	NodeExecutionIdentifier     *core.NodeExecutionIdentifier
	TaskExecutionIdentifier     *core.TaskExecutionIdentifier
}

Metadata to be associated with the catalog object

type ReaderProcessor

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

func NewReaderProcessor

func NewReaderProcessor(catalogClient Client) ReaderProcessor

func (ReaderProcessor) Process

func (p ReaderProcessor) Process(ctx context.Context, workItem workqueue.WorkItem) (workqueue.WorkStatus, error)

type ReaderWorkItem

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

func NewReaderWorkItem

func NewReaderWorkItem(key Key, outputsWriter io.OutputWriter) *ReaderWorkItem

func (ReaderWorkItem) IsCached

func (item ReaderWorkItem) IsCached() bool

type ReadyHandler

type ReadyHandler func(ctx context.Context, future Future)

type ResponseStatus

type ResponseStatus uint8

const (
	ResponseStatusNotReady ResponseStatus = iota
	ResponseStatusReady
)

type UploadFuture

type UploadFuture interface {
	Future
}

Catalog Upload future to represent async process of uploading catalog artifacts.

type UploadRequest

type UploadRequest struct {
	Key              Key
	ArtifactData     io.OutputReader
	ArtifactMetadata Metadata
}

type WriterWorkItem

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

func NewWriterWorkItem

func NewWriterWorkItem(key Key, data io.OutputReader, metadata Metadata) *WriterWorkItem