webapi

Documentation

Index

• Constants
• func CreateRemotePlugin(pluginEntry webapi.PluginEntry) core.PluginEntry
• type CacheItem
  • func (c CacheItem) IsTerminal() bool
• type Client
• type CorePlugin
  • func (c CorePlugin) Abort(ctx context.Context, tCtx core.TaskExecutionContext) error
  • func (c CorePlugin) Finalize(ctx context.Context, tCtx core.TaskExecutionContext) error
  • func (c CorePlugin) GetID() string
  • func (c CorePlugin) GetProperties() core.PluginProperties
  • func (c CorePlugin) Handle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error)
• type Metrics
• type Phase
  • func PhaseString(s string) (Phase, error)
  • func PhaseValues() []Phase
  • func ToPluginPhase(s core.Phase) (Phase, error)
  • func (i Phase) IsAPhase() bool
  • func (i Phase) IsTerminal() bool
  • func (i Phase) String() string
• type ResourceCache
  • func NewResourceCache(ctx context.Context, name string, client Client, cfg webapi.CachingConfig, ...) (ResourceCache, error)
  • func (q *ResourceCache) SyncResource(ctx context.Context, batch cache.Batch) (updatedBatch []cache.ItemSyncResponse, err error)
• type State

Constants

const (
	BadReturnCodeError stdErrors.ErrorCode = "RETURNED_UNKNOWN"
)

Functions

func CreateRemotePlugin

func CreateRemotePlugin(pluginEntry webapi.PluginEntry) core.PluginEntry

Types

type CacheItem

type CacheItem struct {
	State
	Resource webapi.Resource
}

A wrapper for each item in the cache.

func (CacheItem) IsTerminal

func (c CacheItem) IsTerminal() bool

type Client

type Client interface {
	// Get multiple resources that match all the keys. If the plugin hits any failure, it should stop and return
	// the failure. This batch will not be processed further.
	Get(ctx context.Context, tCtx webapi.GetContext) (latest webapi.Resource, err error)

	// Status checks the status of a given resource and translates it to a Flyte-understandable PhaseInfo. This API
	// should avoid making any network calls and should run very efficiently.
	Status(ctx context.Context, tCtx webapi.StatusContext) (phase core.PhaseInfo, err error)
}

Client interface needed for resource cache to fetch latest updates for resources.

type CorePlugin

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

func (CorePlugin) Abort

func (c CorePlugin) Abort(ctx context.Context, tCtx core.TaskExecutionContext) error

func (CorePlugin) Finalize

func (c CorePlugin) Finalize(ctx context.Context, tCtx core.TaskExecutionContext) error

func (CorePlugin) GetID

func (c CorePlugin) GetID() string

func (CorePlugin) GetProperties

func (c CorePlugin) GetProperties() core.PluginProperties

func (CorePlugin) Handle

func (c CorePlugin) Handle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error)

type Metrics

type Metrics struct {
	Scope                   promutils.Scope
	ResourceReleased        labeled.Counter
	ResourceReleaseFailed   labeled.Counter
	AllocationGranted       labeled.Counter
	AllocationNotGranted    labeled.Counter
	ResourceWaitTime        prometheus.Summary
	SucceededUnmarshalState labeled.StopWatch
	FailedUnmarshalState    labeled.Counter
}

type Phase

type Phase int

Phase represents current phase of the execution

const (
	// PhaseNotStarted the default phase.
	PhaseNotStarted Phase = iota

	// PhaseAllocationTokenAcquired once all required tokens have been acquired. The task is ready to be executed
	// remotely.
	PhaseAllocationTokenAcquired

	// PhaseResourcesCreated indicates the task has been created remotely.
	PhaseResourcesCreated

	// The resource has successfully been executed remotely.
	PhaseSucceeded

	// The resource has failed to be executed.
	PhaseUserFailure

	// The resource has failed to be executed due to a system error.
	PhaseSystemFailure
)

func PhaseString

func PhaseString(s string) (Phase, error)

PhaseString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.

func PhaseValues

func PhaseValues() []Phase

PhaseValues returns all values of the enum

func ToPluginPhase

func ToPluginPhase(s core.Phase) (Phase, error)

ToPluginPhase translates the more granular task phase into the webapi plugin phase.

func (Phase) IsAPhase

func (i Phase) IsAPhase() bool

IsAPhase returns "true" if the value is listed in the enum definition. "false" otherwise

func (Phase) IsTerminal

func (i Phase) IsTerminal() bool

func (Phase) String

func (i Phase) String() string

type ResourceCache

type ResourceCache struct {
	// AutoRefresh
	cache.AutoRefresh
	// contains filtered or unexported fields
}

A generic AutoRefresh cache that uses a client to fetch items' status.

func NewResourceCache

func NewResourceCache(ctx context.Context, name string, client Client, cfg webapi.CachingConfig,
	rateCfg webapi.RateLimiterConfig,
	scope promutils.Scope) (ResourceCache, error)

func (*ResourceCache) SyncResource

func (q *ResourceCache) SyncResource(ctx context.Context, batch cache.Batch) (
	updatedBatch []cache.ItemSyncResponse, err error)

This basically grab an updated status from Client and store it in the cache All other handling should be in the synchronous loop.

type State

type State struct {
	// Phase current phase of the resource.
	Phase Phase `json:"phase,omitempty"`

	// PhaseVersion is the version of the phase. This is used to detect if the phase has changed since the last time
	PhaseVersion uint32

	// ResourceMeta contain metadata about resource this task created. This can be a complex structure or a simple type
	// (e.g. a string). It should contain enough information for the plugin to interact (retrieve, check status, delete)
	// with the resource through the remote service.
	ResourceMeta webapi.ResourceMeta `json:"resourceMeta,omitempty"`

	// This number keeps track of the number of failures within the sync function. Without this, what happens in
	// the sync function is entirely opaque. Note that this field is completely orthogonal to Flyte system/node/task
	// level retries, just errors from hitting API, inside the sync loop
	SyncFailureCount int `json:"syncFailureCount,omitempty"`

	// The time the execution first requests for an allocation token
	AllocationTokenRequestStartTime time.Time `json:"allocationTokenRequestStartTime,omitempty"`

	// ErrorMessage generated during cache synchronization.
	ErrorMessage string `json:"error_message,omitempty"`
}

State is the persisted State of the resource.