launchplan

Documentation

Index

• func IsAlreadyExists(err error) bool
• func IsNotFound(err error) bool
• func IsUserError(err error) bool
• type AdminConfig
  • func GetAdminConfig() *AdminConfig
  • func (cfg AdminConfig) GetPFlagSet(prefix string) *pflag.FlagSet
• type ErrorCode
• type Executor
  • func NewAdminLaunchPlanExecutor(_ context.Context, client service.AdminServiceClient, syncPeriod time.Duration, ...) (Executor, error)
  • func NewFailFastLaunchPlanExecutor() Executor
• type LaunchContext

Functions

func IsAlreadyExists

func IsAlreadyExists(err error) bool

Checks if the error is of type RemoteError and the ErrorCode is of type RemoteErrorAlreadyExists

func IsNotFound

func IsNotFound(err error) bool

Checks if the error is of type RemoteError and the ErrorCode is of type RemoteErrorNotFound

func IsUserError

func IsUserError(err error) bool

Checks if the error is of type RemoteError and the ErrorCode is of type RemoteErrorUser

Types

type AdminConfig

type AdminConfig struct {
	// TPS indicates the maximum transactions per second to flyte admin from this client.
	// If it's zero, the created client will use DefaultTPS: 5
	TPS int64 `json:"tps" pflag:",The maximum number of transactions per second to flyte admin from this client."`

	// Maximum burst for throttle.
	// If it's zero, the created client will use DefaultBurst: 10.
	Burst int `json:"burst" pflag:",Maximum burst for throttle"`

	MaxCacheSize int `json:"cacheSize" pflag:",Maximum cache in terms of number of items stored."`

	Workers int `json:"workers" pflag:",Number of parallel workers to work on the queue."`
}

func GetAdminConfig

func GetAdminConfig() *AdminConfig

func (AdminConfig) GetPFlagSet

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

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

type ErrorCode

type ErrorCode = errors2.ErrorCode

const (
	RemoteErrorAlreadyExists ErrorCode = "AlreadyExists"
	RemoteErrorNotFound      ErrorCode = "NotFound"
	RemoteErrorSystem        ErrorCode = "SystemError" // timeouts, network error etc
	RemoteErrorUser          ErrorCode = "UserError"   // Incase of bad specification, invalid arguments, etc
)

type Executor

type Executor interface {
	// Start an execution of a launchplan
	Launch(ctx context.Context, launchCtx LaunchContext, executionID *core.WorkflowExecutionIdentifier, launchPlanRef *core.Identifier, inputs *core.LiteralMap) error

	// Retrieve status of a LaunchPlan execution
	GetStatus(ctx context.Context, executionID *core.WorkflowExecutionIdentifier) (*admin.ExecutionClosure, error)

	// Kill a remote execution
	Kill(ctx context.Context, executionID *core.WorkflowExecutionIdentifier, reason string) error

	// Initializes Executor.
	Initialize(ctx context.Context) error
}

Interface to be implemented by the remote system that can allow workflow launching capabilities

func NewAdminLaunchPlanExecutor

func NewAdminLaunchPlanExecutor(_ context.Context, client service.AdminServiceClient,
	syncPeriod time.Duration, cfg *AdminConfig, scope promutils.Scope) (Executor, error)

func NewFailFastLaunchPlanExecutor

func NewFailFastLaunchPlanExecutor() Executor

type LaunchContext

type LaunchContext struct {
	// Nesting level of the current workflow (parent)
	NestingLevel uint32
	// Principal of the current workflow, so that billing can be tied correctly
	Principal string
	// If a node launched the execution, this specifies which node execution
	ParentNodeExecution *core.NodeExecutionIdentifier
}

A simple context that is used to start an execution of a LaunchPlan. It encapsulates enough parent information to tie the executions