agent

Documentation

Index

• Constants
• func RegisterAgentPlugin(agentService *core.AgentService)
• 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
  • func (r ResourceWrapper) IsTerminal() bool

Constants

const ID = "agent-service"

Functions

func RegisterAgentPlugin

func RegisterAgentPlugin(agentService *core.AgentService)

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."`

	// PollInterval is the interval at which the plugin should poll the agent for metadata updates
	PollInterval config.Duration `json:"pollInterval" pflag:",The interval at which the plugin should poll the agent for metadata updates."`
}

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
	CustomInfo *structpb.Struct
	AgentError *admin.AgentError
}

func (ResourceWrapper) IsTerminal

func (r ResourceWrapper) IsTerminal() bool

IsTerminal is used to avoid making network calls to the agent service if the resource is already in a terminal state.