agent

Documentation

Index

• func RegisterAgentPlugin()
• func SetConfig(cfg *Config) error
• type Agent
• type ClientSet
• type Config
  • func GetConfig() *Config
• type Deployment
• type Plugin
  • func (p Plugin) Create(ctx context.Context, taskCtx webapi.TaskExecutionContextReader) (webapi.ResourceMeta, webapi.Resource, error)
  • func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
  • func (p Plugin) ExecuteTaskSync(ctx context.Context, client service.SyncAgentServiceClient, ...) (webapi.ResourceMeta, webapi.Resource, error)
  • func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest webapi.Resource, err error)
  • func (p Plugin) GetConfig() webapi.PluginConfig
  • func (p Plugin) ResourceRequirements(_ context.Context, _ webapi.TaskExecutionContextReader) (namespace core.ResourceNamespace, constraints core.ResourceConstraintsSpec, ...)
  • func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase core.PhaseInfo, err error)
• type Registry
• type ResourceMetaWrapper
• type ResourceWrapper

Functions

func RegisterAgentPlugin

func RegisterAgentPlugin()

func SetConfig

func SetConfig(cfg *Config) error

Types

type Agent

type Agent struct {
	// IsSync indicates whether this agent is a sync agent. Sync agents are expected to return their
	// results synchronously when called by propeller. Given that sync agents can affect the performance
	// of the system, it's important to enforce strict timeout policies.
	// An Async agent, on the other hand, is required to be able to identify jobs by an
	// identifier and query for job statuses as jobs progress.
	IsSync bool
	// AgentDeployment is the agent deployment where this agent is running.
	AgentDeployment *Deployment
}

type ClientSet

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

ClientSet contains the clients exposed to communicate with various agent services.

type Config

type Config struct {
	// WebAPI defines config for the base WebAPI plugin
	WebAPI webapi.PluginConfig `json:"webApi" pflag:",Defines config for the base WebAPI plugin."`

	// ResourceConstraints defines resource constraints on how many executions to be created per project/overall at any given time
	ResourceConstraints core.ResourceConstraintsSpec `` /* 141-byte string literal not displayed */

	// The default agent if there does not exist a more specific matching against task types
	DefaultAgent Deployment `json:"defaultAgent" pflag:",The default agent."`

	// The agents used to match against specific task types. {agentDeploymentID: AgentDeployment}
	AgentDeployments map[string]*Deployment `json:"agents" pflag:",The agents."`

	// Maps task types to their agents. {TaskType: agentDeploymentID}
	AgentForTaskTypes map[string]string `json:"agentForTaskTypes" pflag:"-,"`

	// SupportedTaskTypes is a list of task types that are supported by this plugin.
	SupportedTaskTypes []string `json:"supportedTaskTypes" pflag:"-,Defines a list of task types that are supported by this plugin."`
}

Config is config for 'agent' plugin

func GetConfig

func GetConfig() *Config

type Deployment

type Deployment struct {
	// Endpoint points to an agent gRPC endpoint
	Endpoint string `json:"endpoint"`

	// Insecure indicates whether the communication with the gRPC service is insecure
	Insecure bool `json:"insecure"`

	// DefaultServiceConfig sets default gRPC service config; check https://github.com/grpc/grpc/blob/master/doc/service_config.md for more details
	DefaultServiceConfig string `json:"defaultServiceConfig"`

	// Timeouts defines various RPC timeout values for different plugin operations: CreateTask, GetTask, DeleteTask; if not configured, defaults to DefaultTimeout
	Timeouts map[string]config.Duration `json:"timeouts"`

	// DefaultTimeout gives the default RPC timeout if a more specific one is not defined in Timeouts; if neither DefaultTimeout nor Timeouts is defined for an operation, RPC timeout will not be enforced
	DefaultTimeout config.Duration `json:"defaultTimeout"`
}

type Plugin

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

func (Plugin) Create

func (p Plugin) Create(ctx context.Context, taskCtx webapi.TaskExecutionContextReader) (webapi.ResourceMeta,
	webapi.Resource, error)

func (Plugin) Delete

func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error

func (Plugin) ExecuteTaskSync

func (p Plugin) ExecuteTaskSync(
	ctx context.Context,
	client service.SyncAgentServiceClient,
	header *admin.CreateRequestHeader,
	inputs *flyteIdl.LiteralMap,
) (webapi.ResourceMeta, webapi.Resource, error)

func (Plugin) Get

func (p Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest webapi.Resource, err error)

func (Plugin) GetConfig

func (p Plugin) GetConfig() webapi.PluginConfig

func (Plugin) ResourceRequirements

func (p Plugin) ResourceRequirements(_ context.Context, _ webapi.TaskExecutionContextReader) (
	namespace core.ResourceNamespace, constraints core.ResourceConstraintsSpec, err error)

func (Plugin) Status

func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase core.PhaseInfo, err error)

type Registry

type Registry map[string]map[int32]*Agent // map[taskTypeName][taskTypeVersion] => Agent

type ResourceMetaWrapper

type ResourceMetaWrapper struct {
	OutputPrefix      string
	AgentResourceMeta []byte
	TaskCategory      admin.TaskCategory
}

type ResourceWrapper

type ResourceWrapper struct {
	Phase flyteIdl.TaskExecution_Phase
	// Deprecated: Please Use Phase instead.
	State    admin.State
	Outputs  *flyteIdl.LiteralMap
	Message  string
	LogLinks []*flyteIdl.TaskLog
}