launchplan

Documentation

Index

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

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

func IsWorkflowTerminated

func IsWorkflowTerminated(p core.WorkflowExecution_Phase) bool

IsWorkflowTerminated returns a true if the Workflow Phase is in a Terminal Phase, else returns a false

Types

type AdminConfig

type AdminConfig struct {
	// TPS indicates the maximum transactions per second to kozmo 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 kozmo 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."`
}

AdminConfig provides a "admin-launcher" section in core Kozmopropeller configuration and can be used to configure the rate at which Kozmopropeller can query for status of workflows in kozmoadmin or create new executions

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"   // In case of bad specification, invalid arguments, etc
)

type Executor

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

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

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

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

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

type KozmoAdmin

type KozmoAdmin interface {
	Executor
	Reader
}

func NewAdminLaunchPlanExecutor

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

func NewFailFastLaunchPlanExecutor

func NewFailFastLaunchPlanExecutor() KozmoAdmin

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
	// If a node in recovery mode launched this execution, propagate recovery mode to the child execution.
	RecoveryExecution *core.WorkflowExecutionIdentifier
	// SecurityContext contains information from the parent execution about the security context.
	SecurityContext core.SecurityContext
	// MaxParallelism
	MaxParallelism uint32
	// RawOutputDataConfig
	RawOutputDataConfig  *admin.RawOutputDataConfig
	Annotations          map[string]string
	Labels               map[string]string
	Interruptible        *bool
	OverwriteCache       bool
	EnvironmentVariables map[string]string
}

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

type Reader

type Reader interface {
	// GetLaunchPlan gets the definition of a launch plan. This is primarily used to ensure all the TypedInterfaces match up before actually executing.
	GetLaunchPlan(ctx context.Context, launchPlanRef *core.Identifier) (*admin.LaunchPlan, error)
}