coordinator

package
v2.15.4 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: GPL-3.0 Imports: 65 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrMissingTLSConfig = fmt.Errorf("TLS enabled but no certificates provided")
)

Errors

Functions

func NewRuntimeDispatcher 

func NewRuntimeDispatcher(registry serviceregistry.ServiceRegistry, peerConfig config.Peer) (dispatch.Dispatcher, error)

NewRuntimeDispatcher creates a coordinator-backed dispatcher for runtime DAG execution.

func NewRuntimeProfileResolver added in v2.15.4 

func NewRuntimeProfileResolver(
	client RuntimeProfileClient,
	owner serviceregistry.HostInfo,
	run RuntimeProfileRun,
	fallback profilepkg.RuntimeResolver,
) profilepkg.RuntimeResolver

func NewSecretReferenceResolver 

func NewSecretReferenceResolver(client SecretReferenceClient, workspace string, owner serviceregistry.HostInfo, run SecretReferenceRun) providers.ReferenceResolver

func NewStateStoreClient 

func NewStateStoreClient(client StateClient) dagrun.StateStore

NewStateStoreClient adapts coordinator state RPCs to the persistent state store interface.

func NewSubWorkflowRunnerFactory added in v2.13.0 

func NewSubWorkflowRunnerFactory(cfg SubWorkflowRunnerConfig) func(context.Context) (runtimeexec.SubWorkflowRunner, error)

NewSubWorkflowRunnerFactory creates recursive child workflow runners.

func StreamTypeToExtension 

func StreamTypeToExtension(streamType coordinatorv1.LogStreamType) string

StreamTypeToExtension returns the file extension for a given stream type.

Types

type AgentSessionCleanupClient added in v2.14.0 

type AgentSessionCleanupClient interface {
	ClaimAgentSessionCleanup(ctx context.Context, req *coordinatorv1.ClaimAgentSessionCleanupRequest) (*coordinatorv1.ClaimAgentSessionCleanupResponse, error)
	CompleteAgentSessionCleanupTo(ctx context.Context, owner serviceregistry.HostInfo, req *coordinatorv1.CompleteAgentSessionCleanupRequest) (*coordinatorv1.CompleteAgentSessionCleanupResponse, error)
}

AgentSessionCleanupClient exposes coordinator-backed provider cleanup RPCs.

type Client 

type Client interface {
	dispatch.Dispatcher

	// Poll retrieves a task from the coordinator.
	Poll(ctx context.Context, policy backoff.RetryPolicy, req *coordinatorv1.PollRequest) (*coordinatorv1.Task, error)

	// GetWorkers retrieves the list of workers from the coordinator
	GetWorkers(ctx context.Context) ([]*coordinatorv1.WorkerInfo, error)

	// Heartbeat sends a heartbeat to the coordinator and returns the response
	// which may include cancellation directives
	Heartbeat(ctx context.Context, req *coordinatorv1.HeartbeatRequest) (*coordinatorv1.HeartbeatResponse, error)

	// AckTaskClaim confirms a claimed task with its owner coordinator.
	AckTaskClaimTo(ctx context.Context, owner serviceregistry.HostInfo, req *coordinatorv1.AckTaskClaimRequest) (*coordinatorv1.AckTaskClaimResponse, error)

	// RunHeartbeat refreshes leases for tasks owned by a specific coordinator.
	RunHeartbeatTo(ctx context.Context, owner serviceregistry.HostInfo, req *coordinatorv1.RunHeartbeatRequest) (*coordinatorv1.RunHeartbeatResponse, error)

	// ReportStatus sends a worker status update to the coordinator.
	ReportStatus(ctx context.Context, req *coordinatorv1.ReportStatusRequest) (*coordinatorv1.ReportStatusResponse, error)

	// ReportStatusTo sends a status update to a specific owner coordinator.
	ReportStatusTo(ctx context.Context, owner serviceregistry.HostInfo, req *coordinatorv1.ReportStatusRequest) (*coordinatorv1.ReportStatusResponse, error)

	// StreamLogs returns a log streaming client for sending logs to the coordinator
	StreamLogs(ctx context.Context) (coordinatorv1.CoordinatorService_StreamLogsClient, error)

	// StreamLogsTo opens a log stream to a specific owner coordinator.
	StreamLogsTo(ctx context.Context, owner serviceregistry.HostInfo) (coordinatorv1.CoordinatorService_StreamLogsClient, error)

	// StreamArtifacts returns an artifact streaming client for sending artifacts to the coordinator.
	StreamArtifacts(ctx context.Context) (coordinatorv1.CoordinatorService_StreamArtifactsClient, error)

	// StreamArtifactsTo opens an artifact stream to a specific owner coordinator.
	StreamArtifactsTo(ctx context.Context, owner serviceregistry.HostInfo) (coordinatorv1.CoordinatorService_StreamArtifactsClient, error)

	// RequestCancel requests cancellation of a DAG run through the coordinator.
	// Used by worker sub-DAG cancellation.
	RequestCancel(ctx context.Context, dagName, dagRunID string, rootRef *ir.DAGRunRef) error

	// GetDAG retrieves a DAG definition (raw YAML) from the coordinator's DAG store.
	// Used as a fallback when a worker's local DAG store misses a definition.
	GetDAG(ctx context.Context, name string) (string, error)

	// Metrics returns the metrics for the coordinator client
	Metrics() Metrics
}

Client abstracts handling communication with the coordinator service using service registry and gRPC.

func New 

func New(registry serviceregistry.ServiceRegistry, config *Config) Client

New creates a new coordinator client with the given configuration

type Config 

type Config struct {
	// WorkspaceBundleDir is required when dispatching DAGs with file dependencies.
	WorkspaceBundleDir string

	// TLS configuration
	Insecure      bool   // Use insecure connection (default: true)
	CertFile      string // Client certificate
	KeyFile       string // Client key
	CAFile        string // CA certificate
	SkipTLSVerify bool   // Skip server certificate verification

	// Timeouts
	DialTimeout      time.Duration // Connection timeout (default: 10s)
	RequestTimeout   time.Duration // Per-request timeout (default: 5m)
	HeartbeatTimeout time.Duration // Worker heartbeat timeout (default: 10s)

	// Retry configuration
	MaxRetries    int           // Max dispatch retries (default: 3)
	RetryInterval time.Duration // Base retry interval (default: 1s)
}

Config holds configuration for the coordinator client

func ConfigFromPeer added in v2.14.0 

func ConfigFromPeer(peer appconfig.Peer) *Config

ConfigFromPeer maps application peer settings to coordinator client settings.

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig returns a Config with default values

func (*Config) Validate 

func (c *Config) Validate() error

Validate checks if the configuration is valid

type Handler 

type Handler struct {
	coordinatorv1.UnimplementedCoordinatorServiceServer
	// contains filtered or unexported fields
}

func NewHandler 

func NewHandler(cfg HandlerConfig) *Handler

NewHandler creates a new Handler with the given configuration.

func (*Handler) AckTaskClaim 

func (h *Handler) AckTaskClaim(ctx context.Context, req *coordinatorv1.AckTaskClaimRequest) (*coordinatorv1.AckTaskClaimResponse, error)

AckTaskClaim confirms that a worker accepted a claimed task and creates the initial active lease for that distributed attempt.

func (*Handler) ClaimAgentSessionCleanup added in v2.14.0 

func (h *Handler) ClaimAgentSessionCleanup(
	ctx context.Context,
	req *coordinatorv1.ClaimAgentSessionCleanupRequest,
) (*coordinatorv1.ClaimAgentSessionCleanupResponse, error)

ClaimAgentSessionCleanup reserves deferred provider cleanup for its owning worker.

func (*Handler) Close 

func (h *Handler) Close(ctx context.Context)

Close cleans up all resources held by the handler. This should be called during coordinator shutdown.

func (*Handler) CompleteAgentSessionCleanup added in v2.14.0 

func (h *Handler) CompleteAgentSessionCleanup(
	ctx context.Context,
	req *coordinatorv1.CompleteAgentSessionCleanupRequest,
) (*coordinatorv1.CompleteAgentSessionCleanupResponse, error)

CompleteAgentSessionCleanup completes or releases a provider cleanup claim.

func (*Handler) DeleteState 

func (h *Handler) DeleteState(ctx context.Context, req *coordinatorv1.DeleteStateRequest) (*coordinatorv1.DeleteStateResponse, error)

DeleteState removes a state entry through the coordinator state store.

func (*Handler) Dispatch 

func (h *Handler) Dispatch(ctx context.Context, req *coordinatorv1.DispatchRequest) (*coordinatorv1.DispatchResponse, error)

Dispatch tries to send a task to a waiting poller It fails if no pollers are available or no workers match the selector

func (*Handler) GetDAG 

func (h *Handler) GetDAG(ctx context.Context, req *coordinatorv1.GetDAGRequest) (*coordinatorv1.GetDAGResponse, error)

GetDAG retrieves the raw specification of a DAG by name. Workers use this to obtain DAG definitions that may not be available locally.

func (*Handler) GetDAGRunStatus 

func (h *Handler) GetDAGRunStatus(ctx context.Context, req *coordinatorv1.GetDAGRunStatusRequest) (*coordinatorv1.GetDAGRunStatusResponse, error)

GetDAGRunStatus retrieves the status of a DAG run. Parent DAGs use this to poll remote sub-DAG status through the coordinator.

func (*Handler) GetState 

func (h *Handler) GetState(ctx context.Context, req *coordinatorv1.GetStateRequest) (*coordinatorv1.GetStateResponse, error)

GetState returns the current state entry for a reference, if it exists.

func (*Handler) GetWorkers 

func (h *Handler) GetWorkers(_ context.Context, _ *coordinatorv1.GetWorkersRequest) (*coordinatorv1.GetWorkersResponse, error)

GetWorkers returns the list of currently connected workers

func (*Handler) GetWorkspaceBundle 

func (h *Handler) GetWorkspaceBundle(req *coordinatorv1.GetWorkspaceBundleRequest, stream coordinatorv1.CoordinatorService_GetWorkspaceBundleServer) error

func (*Handler) HasWorkspaceBundle 

func (h *Handler) HasWorkspaceBundle(ctx context.Context, req *coordinatorv1.HasWorkspaceBundleRequest) (*coordinatorv1.HasWorkspaceBundleResponse, error)

func (*Handler) Heartbeat 

func (h *Handler) Heartbeat(ctx context.Context, req *coordinatorv1.HeartbeatRequest) (*coordinatorv1.HeartbeatResponse, error)

Heartbeat receives periodic status updates from workers.

func (*Handler) ListState 

func (h *Handler) ListState(ctx context.Context, req *coordinatorv1.ListStateRequest) (*coordinatorv1.ListStateResponse, error)

ListState returns state entries matching the requested scope and key prefix.

func (*Handler) Poll 

func (h *Handler) Poll(ctx context.Context, req *coordinatorv1.PollRequest) (*coordinatorv1.PollResponse, error)

Poll implements long polling - workers wait until a task is available

func (*Handler) PutState 

func (h *Handler) PutState(ctx context.Context, req *coordinatorv1.PutStateRequest) (*coordinatorv1.PutStateResponse, error)

PutState creates or updates a state entry through the coordinator state store.

func (*Handler) PutWorkspaceBundle 

func (h *Handler) PutWorkspaceBundle(stream coordinatorv1.CoordinatorService_PutWorkspaceBundleServer) error

func (*Handler) ReportStatus 

func (h *Handler) ReportStatus(ctx context.Context, req *coordinatorv1.ReportStatusRequest) (*coordinatorv1.ReportStatusResponse, error)

ReportStatus receives status updates from workers and persists them.

func (*Handler) RequestCancel 

func (h *Handler) RequestCancel(ctx context.Context, req *coordinatorv1.RequestCancelRequest) (*coordinatorv1.RequestCancelResponse, error)

RequestCancel handles requests to cancel a DAG run. Parent workers use this for sub-DAG cancellation through the coordinator.

func (*Handler) ResolveRuntimeProfile added in v2.15.4 

func (h *Handler) ResolveRuntimeProfile(
	ctx context.Context,
	req *coordinatorv1.ResolveRuntimeProfileRequest,
) (*coordinatorv1.ResolveRuntimeProfileResponse, error)

func (*Handler) ResolveSecretReference 

func (h *Handler) ResolveSecretReference(ctx context.Context, req *coordinatorv1.ResolveSecretReferenceRequest) (*coordinatorv1.ResolveSecretReferenceResponse, error)

func (*Handler) RunHeartbeat 

func (h *Handler) RunHeartbeat(ctx context.Context, req *coordinatorv1.RunHeartbeatRequest) (*coordinatorv1.RunHeartbeatResponse, error)

RunHeartbeat refreshes leases for tasks owned by this coordinator and returns cancellation directives for those exact tasks.

func (*Handler) StartZombieDetector 

func (h *Handler) StartZombieDetector(ctx context.Context, interval time.Duration)

StartZombieDetector starts a background goroutine that periodically checks for zombie runs. It detects workers that have stopped sending heartbeats and marks their running tasks as failed. The interval parameter controls how often the detector runs (recommended: 45 seconds). Call WaitZombieDetector after canceling the context to ensure clean shutdown. This method is safe to call multiple times; subsequent calls are no-ops.

func (*Handler) StreamArtifacts 

func (h *Handler) StreamArtifacts(stream coordinatorv1.CoordinatorService_StreamArtifactsServer) error

StreamArtifacts receives artifact streams from workers and writes them to local filesystem.

func (*Handler) StreamLogs 

func (h *Handler) StreamLogs(stream coordinatorv1.CoordinatorService_StreamLogsServer) error

StreamLogs receives log streams from workers and writes them to local filesystem.

func (*Handler) WaitZombieDetector 

func (h *Handler) WaitZombieDetector()

WaitZombieDetector waits for the zombie detector goroutine to finish. This should be called after the context passed to StartZombieDetector is canceled.

type HandlerConfig 

type HandlerConfig struct {
	// DAGRunRepository provides application access to persisted DAG-run statuses.
	// Required for worker status reporting.
	DAGRunRepository *persis.DAGRunRepository

	// LogDir is the directory for streamed worker log storage.
	// Required for worker log streaming.
	LogDir string

	// ArtifactDir is the directory for streamed worker artifact storage.
	// Required for worker artifact streaming.
	ArtifactDir string

	// StateStore is the persistent DAG state store used by state RPCs.
	StateStore dagrun.StateStore

	// WorkspaceBundleDir stores immutable task workspace bundles by digest.
	WorkspaceBundleDir string

	// Owner identifies this coordinator instance for shared task ownership.
	Owner dispatch.CoordinatorEndpoint

	// DispatchTaskStore is the shared store for distributed pending tasks.
	DispatchTaskStore dispatch.DispatchTaskStore

	// DispatchAdmissionStore reserves and binds distributed queue admission.
	DispatchAdmissionStore dispatch.DispatchAdmissionStore

	// WorkerHeartbeatStore is the shared store for worker presence.
	WorkerHeartbeatStore dispatch.WorkerHeartbeatStore

	// DAGRunLeaseStore is the shared store for active distributed attempt leases.
	DAGRunLeaseStore dispatch.DAGRunLeaseStore

	// ActiveDistributedRunStore is the shared store for the coordinator-owned
	// active distributed attempt index used by zombie detection.
	ActiveDistributedRunStore dispatch.ActiveDistributedRunStore

	// DAGRepository serves DAG definitions for the GetDAG RPC.
	// Optional - when nil, GetDAG returns Unimplemented.
	DAGRepository *persis.DAGRepository

	// SecretStore resolves Dagu-managed secret registry refs for workers.
	// Optional - when nil, ResolveSecretReference returns FailedPrecondition.
	SecretStore secretpkg.Store

	// ProfileStore resolves runtime profiles for workers.
	// Optional - when nil, ResolveRuntimeProfile returns FailedPrecondition.
	ProfileStore profilepkg.Store

	// AgentSessionCleanupQueue stores provider cleanup claimed by owning workers.
	AgentSessionCleanupQueue *agentsession.CleanupQueue

	// StaleHeartbeatThreshold is the duration after which a worker's heartbeat
	// is considered stale. Defaults to 30 seconds if not set.
	StaleHeartbeatThreshold time.Duration

	// StaleLeaseThreshold is the duration after which a distributed run's
	// lease is considered stale (worker stopped pushing status). Defaults to 90 seconds.
	StaleLeaseThreshold time.Duration

	// EventService persists coordinator-originated event envelopes.
	EventService *eventstore.Service

	// EventSourceInstance identifies this coordinator instance in event envelopes.
	EventSourceInstance string
}

HandlerConfig holds configuration for creating a Handler.

type Metrics 

type Metrics struct {
	FailCount        int   // Total number of failures
	IsConnected      bool  // Whether the client is currently connected
	ConsecutiveFails int   // Number of consecutive failures
	LastError        error // Last error encountered
}

Metrics defines the metrics for the coordinator client

type RuntimeProfileClient added in v2.15.4 

type RuntimeProfileClient interface {
	ResolveRuntimeProfile(context.Context, serviceregistry.HostInfo, profilepkg.RuntimeRequest, RuntimeProfileRun) (*profilepkg.RuntimeResolved, error)
}

type RuntimeProfileRun added in v2.15.4 

type RuntimeProfileRun struct {
	WorkerID   string
	AttemptKey string
	AttemptID  string
	DAGName    string
}

type SecretReferenceClient 

type SecretReferenceClient interface {
	ResolveSecretReference(ctx context.Context, owner serviceregistry.HostInfo, ref secretref.Ref, workspace string, checkOnly bool, run SecretReferenceRun) (string, error)
}

type SecretReferenceRun 

type SecretReferenceRun struct {
	WorkerID   string
	AttemptKey string
	AttemptID  string
	DAGName    string
}

type Service 

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

func NewService 

func NewService(
	server *grpc.Server,
	handler *Handler,
	grpcListener net.Listener,
	grpcHealthServer *health.Server,
	httpHealthServer *healthcheck.Server,
	registry serviceregistry.ServiceRegistry,
	cfg *config.Config,
	instanceID string,
	configuredHost string,
) *Service

func (*Service) DisableHealthServer 

func (srv *Service) DisableHealthServer()

DisableHealthServer disables the dedicated HTTP health check server.

func (*Service) Start 

func (srv *Service) Start(ctx context.Context) (err error)

func (*Service) Stop 

func (srv *Service) Stop(ctx context.Context) error

type StateClient 

type StateClient interface {
	// GetState retrieves one state entry by reference.
	GetState(ctx context.Context, req *coordinatorv1.GetStateRequest) (*coordinatorv1.GetStateResponse, error)
	// PutState creates or updates one state entry.
	PutState(ctx context.Context, req *coordinatorv1.PutStateRequest) (*coordinatorv1.PutStateResponse, error)
	// DeleteState removes one state entry by reference.
	DeleteState(ctx context.Context, req *coordinatorv1.DeleteStateRequest) (*coordinatorv1.DeleteStateResponse, error)
	// ListState lists state entries under a scope, namespace, and key prefix.
	ListState(ctx context.Context, req *coordinatorv1.ListStateRequest) (*coordinatorv1.ListStateResponse, error)
}

StateClient exposes coordinator-backed persistent DAG state RPCs.

type StaticRegistry 

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

StaticRegistry is a simple service registry that returns a fixed list of coordinator addresses. This is useful for worker deployments where the coordinator addresses are known and specified via CLI flags or environment variables.

func NewStaticRegistry 

func NewStaticRegistry(addresses []string) (*StaticRegistry, error)

NewStaticRegistry creates a new StaticRegistry from a list of address strings. Each address should be in the format "host[:port]" or "[ipv6]:port".

func (*StaticRegistry) GetServiceMembers 

func (r *StaticRegistry) GetServiceMembers(_ context.Context, name serviceregistry.ServiceName) ([]serviceregistry.HostInfo, error)

GetServiceMembers returns the list of coordinator hosts. Only ServiceNameCoordinator is supported; other services return an empty list.

func (*StaticRegistry) Register 

func (r *StaticRegistry) Register(_ context.Context, _ serviceregistry.ServiceName, _ serviceregistry.HostInfo) error

Register is a no-op for StaticRegistry since we don't support registration.

func (*StaticRegistry) Unregister 

func (r *StaticRegistry) Unregister(_ context.Context)

Unregister is a no-op for StaticRegistry since we don't support registration.

func (*StaticRegistry) UpdateStatus 

func (r *StaticRegistry) UpdateStatus(_ context.Context, _ serviceregistry.ServiceName, _ serviceregistry.ServiceStatus) error

UpdateStatus is a no-op for StaticRegistry since we don't support status updates.

type SubWorkflowRunnerConfig added in v2.13.0 

type SubWorkflowRunnerConfig struct {
	// Dispatcher is caller-owned and remains live after a child runner is cleaned up.
	Dispatcher        dispatch.Dispatcher
	DAGRunMgr         runtime.Manager
	DAGRepository     *persis.DAGRepository
	DAGRunRepository  *persis.DAGRunRepository
	RunStateStore     runstate.Store
	QueueStore        queue.QueueStore
	StateStore        dagrun.StateStore
	SecretStore       secret.Store
	SecretResolver    func(*ir.DAG) providers.ReferenceResolver
	ProfileStore      profile.Store
	ProfileResolver   func(*ir.DAG) profile.RuntimeResolver
	ServiceRegistry   serviceregistry.ServiceRegistry
	PeerConfig        config.Peer
	DefaultExecMode   config.ExecutionMode
	StatusPusher      runtime.StatusPusher
	LogWriterFactory  runctx.LogWriterFactory
	ArtifactFinalizer runtime.ArtifactFinalizer
	RemoteDAGLoader   rtagent.RemoteDAGLoader
	WorkerID          string
	DAGRunLogDir      string
	DAGRunArtifactDir string
}

SubWorkflowRunnerConfig contains dependencies for child workflow execution.

Source Files

View all Source files

Directories

Path Synopsis
Package subflow adapts Dagu child workflow execution to the runtime executor's child workflow interface.
 Package subflow adapts Dagu child workflow execution to the runtime executor's child workflow interface.

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.