executor

package
v1.2.11 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 28, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyRegistered = errors.New("executor already registered")
View Source 
var ErrLocalFuncAlreadyRegistered = errors.New("local function already registered")
View Source 
var ErrNotImplemented = errors.New("executor not implemented")

Functions

func ClassifyFailure 

func ClassifyFailure(err error) (retryable bool, retryAfter time.Duration, classified bool)

func PermanentFailure 

func PermanentFailure(err error) error

func RetryableFailure 

func RetryableFailure(err error, retryAfter time.Duration) error

Types

type AsyncExecutor 

type AsyncExecutor interface {
	Executor
	Poll(ctx context.Context, task ExecuteTask, externalTaskID string) (ExecuteResult, error)
	Cancel(ctx context.Context, task ExecuteTask, externalTaskID string) error
}

AsyncExecutor is optional and only needed for queue/worker style execution.

type ContainerExecutor 

type ContainerExecutor struct{}

func (*ContainerExecutor) Cancel 

func (e *ContainerExecutor) Cancel(context.Context, Request, string) error

func (*ContainerExecutor) Execute 

func (e *ContainerExecutor) Execute(ctx context.Context, task Request) (Result, error)

ContainerExecutor skeleton: submit container job, then poll.

func (*ContainerExecutor) Poll 

func (e *ContainerExecutor) Poll(context.Context, Request, string) (Result, error)

func (*ContainerExecutor) Type 

func (e *ContainerExecutor) Type() Type

type Dispatcher 

type Dispatcher interface {
	Dispatch(task ExecuteTask) (Executor, error)
}

Dispatcher resolves a task to a concrete executor implementation.

type ExecuteResult 

type ExecuteResult struct {
	Status          Status         `json:"status,omitempty"`
	Output          any            `json:"output,omitempty"`
	Variables       map[string]any `json:"variables,omitempty"`
	DeleteVariables []string       `json:"delete_variables,omitempty"`
	Error           string         `json:"error,omitempty"`
	Metadata        map[string]any `json:"metadata,omitempty"`
	Logs            []string       `json:"logs,omitempty"`
	ExternalTaskID  string         `json:"external_task_id,omitempty"`
	RetryAfter      time.Duration  `json:"retry_after,omitempty"`
	FinishedAt      time.Time      `json:"finished_at,omitempty,omitzero"`
}

func (ExecuteResult) NormalizedStatus 

func (r ExecuteResult) NormalizedStatus() Status

type ExecuteTask 

type ExecuteTask struct {
	DispatchID      string         `json:"dispatch_id,omitempty"`
	RunID           string         `json:"run_id"`
	NodeID          string         `json:"node_id"`
	ExecutorType    string         `json:"executor_type"`
	ExecutorRef     string         `json:"executor_ref,omitempty"`
	CredentialScope string         `json:"credential_scope,omitempty"`
	Attempt         int            `json:"attempt,omitempty"`
	MaxAttempts     int            `json:"max_attempts,omitempty"`
	Input           any            `json:"input,omitempty"`
	Params          map[string]any `json:"params,omitempty"`
	Context         map[string]any `json:"context,omitempty"`
	Timeout         time.Duration  `json:"timeout,omitempty"`
	Deadline        time.Time      `json:"deadline,omitempty,omitzero"`
	Async           bool           `json:"async,omitempty"`
	PollInterval    time.Duration  `json:"poll_interval,omitempty"`
	HeartbeatFreq   time.Duration  `json:"heartbeat_freq,omitempty"`
}

func (ExecuteTask) Validate 

func (t ExecuteTask) Validate() error

type ExecutionFailure 

type ExecutionFailure struct {
	Err        error
	Retry      bool
	RetryAfter time.Duration
}

ExecutionFailure lets executors distinguish permanent failures from transient failures without making the scheduler inspect error strings.

func (*ExecutionFailure) Error 

func (e *ExecutionFailure) Error() string

func (*ExecutionFailure) Unwrap 

func (e *ExecutionFailure) Unwrap() error

type Executor 

type Executor interface {
	Type() Type
	Execute(ctx context.Context, task ExecuteTask) (ExecuteResult, error)
}

type HTTPExecutor 

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

func NewHTTPExecutor 

func NewHTTPExecutor(client *http.Client) *HTTPExecutor

func (*HTTPExecutor) Execute 

func (e *HTTPExecutor) Execute(ctx context.Context, req Request) (Result, error)

func (*HTTPExecutor) Type 

func (e *HTTPExecutor) Type() Type

type InMemoryQueueBroker 

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

func NewInMemoryQueueBroker 

func NewInMemoryQueueBroker() *InMemoryQueueBroker

func (*InMemoryQueueBroker) Ack 

func (b *InMemoryQueueBroker) Ack(_ context.Context, ticket string, result ExecuteResult) error

func (*InMemoryQueueBroker) Cancel 

func (b *InMemoryQueueBroker) Cancel(_ context.Context, ticket string) error

func (*InMemoryQueueBroker) Dequeue 

func (b *InMemoryQueueBroker) Dequeue(ctx context.Context, queue string, wait time.Duration) (*QueuedTask, error)

func (*InMemoryQueueBroker) Enqueue 

func (b *InMemoryQueueBroker) Enqueue(ctx context.Context, queue string, task ExecuteTask) (string, error)

func (*InMemoryQueueBroker) Poll 

func (b *InMemoryQueueBroker) Poll(_ context.Context, ticket string) (ExecuteResult, error)

type LocalExecutor 

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

func NewLocalExecutor 

func NewLocalExecutor() *LocalExecutor

func (*LocalExecutor) Execute 

func (e *LocalExecutor) Execute(ctx context.Context, req Request) (Result, error)

func (*LocalExecutor) Register 

func (e *LocalExecutor) Register(name string, fn LocalFunc) error

func (*LocalExecutor) RegisterOrReplace 

func (e *LocalExecutor) RegisterOrReplace(name string, fn LocalFunc) error

func (*LocalExecutor) Type 

func (e *LocalExecutor) Type() Type

type LocalFunc 

type LocalFunc func(ctx context.Context, req Request) (Result, error)

type NodeExecutor 

type NodeExecutor struct{}

func (*NodeExecutor) Execute 

func (e *NodeExecutor) Execute(context.Context, Request) (Result, error)

func (*NodeExecutor) Type 

func (e *NodeExecutor) Type() Type

type PythonExecutor 

type PythonExecutor struct{}

func (*PythonExecutor) Execute 

func (e *PythonExecutor) Execute(context.Context, Request) (Result, error)

func (*PythonExecutor) Type 

func (e *PythonExecutor) Type() Type

type QueueBroker 

type QueueBroker interface {
	Enqueue(ctx context.Context, queue string, task ExecuteTask) (string, error)
	Dequeue(ctx context.Context, queue string, wait time.Duration) (*QueuedTask, error)
	Ack(ctx context.Context, ticket string, result ExecuteResult) error
	Poll(ctx context.Context, ticket string) (ExecuteResult, error)
	Cancel(ctx context.Context, ticket string) error
}

type QueueExecutor 

type QueueExecutor struct {
	Broker QueueBroker
}

func NewQueueExecutor 

func NewQueueExecutor(broker QueueBroker) *QueueExecutor

func (*QueueExecutor) Cancel 

func (e *QueueExecutor) Cancel(ctx context.Context, task ExecuteTask, externalTaskID string) error

func (*QueueExecutor) Execute 

func (e *QueueExecutor) Execute(ctx context.Context, task ExecuteTask) (ExecuteResult, error)

func (*QueueExecutor) Poll 

func (e *QueueExecutor) Poll(ctx context.Context, task ExecuteTask, externalTaskID string) (ExecuteResult, error)

func (*QueueExecutor) Type 

func (e *QueueExecutor) Type() Type

type QueuedTask 

type QueuedTask struct {
	Ticket string      `json:"ticket"`
	Queue  string      `json:"queue"`
	Task   ExecuteTask `json:"task"`
}

type Registry 

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

func NewRegistry 

func NewRegistry() *Registry

func (*Registry) Get 

func (r *Registry) Get(t Type) (Executor, bool)

func (*Registry) MustRegister 

func (r *Registry) MustRegister(exec Executor)

func (*Registry) Register 

func (r *Registry) Register(exec Executor) error

func (*Registry) RegisterAll 

func (r *Registry) RegisterAll(executors ...Executor) error

RegisterAll validates the complete batch before changing the registry. This prevents a partially installed executor set when one item conflicts.

func (*Registry) RegisterOrReplace 

func (r *Registry) RegisterOrReplace(exec Executor) error

RegisterOrReplace makes replacement an explicit operation. Prefer Register during normal composition so accidental type collisions fail fast.

func (*Registry) Types 

func (r *Registry) Types() []Type

type RegistryDispatcher 

type RegistryDispatcher struct {
	Registry *Registry
}

func NewRegistryDispatcher 

func NewRegistryDispatcher(reg *Registry) *RegistryDispatcher

func (*RegistryDispatcher) Dispatch 

func (d *RegistryDispatcher) Dispatch(task ExecuteTask) (Executor, error)

type RemoteExecutor 

type RemoteExecutor struct{}

func (*RemoteExecutor) Cancel 

func (e *RemoteExecutor) Cancel(context.Context, Request, string) error

func (*RemoteExecutor) Execute 

func (e *RemoteExecutor) Execute(ctx context.Context, task Request) (Result, error)

RemoteExecutor skeleton: submit -> accepted, then Poll on externalTaskID.

func (*RemoteExecutor) Poll 

func (e *RemoteExecutor) Poll(context.Context, Request, string) (Result, error)

func (*RemoteExecutor) Type 

func (e *RemoteExecutor) Type() Type

type Request 

type Request = ExecuteTask

Backward-compatible aliases for existing code during migration.

type Result 

type Result = ExecuteResult

type ScriptExecutor 

type ScriptExecutor struct{}

ScriptExecutor executes lightweight inline scripts for python/node runtimes. It is intentionally simple and can be replaced by a sandboxed implementation later.

func NewScriptExecutor 

func NewScriptExecutor() *ScriptExecutor

func (*ScriptExecutor) Execute 

func (e *ScriptExecutor) Execute(ctx context.Context, task ExecuteTask) (ExecuteResult, error)

func (*ScriptExecutor) Type 

func (e *ScriptExecutor) Type() Type

type Status 

type Status string
const (
	StatusSucceeded Status = "succeeded"
	StatusFailed    Status = "failed"
	StatusRetryable Status = "retryable"
	StatusAccepted  Status = "accepted" // async task accepted by backend worker
	StatusRunning   Status = "running"  // async task still running when polled
)

type Type 

type Type string
const (
	TypeHTTP      Type = "http"
	TypeLocalGo   Type = "local_go"
	TypeScript    Type = "script"
	TypePython    Type = "python"
	TypeNodeJS    Type = "node"
	TypeQueue     Type = "queue"
	TypeRemote    Type = "remote"
	TypeContainer Type = "container"
	TypeUnit      Type = "unit"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.