worker

package
v1.0.0-rc.2 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrPlatformConfig = errors.New("platform-url configuration error")

ErrPlatformConfig marks errors that stem from a misconfigured platform-url (empty, unparsable, unsupported scheme, or a scheme that disagrees with what the origin serves). These are deterministic — retrying will not help — so callers should fail fast rather than loop. Transient reachability failures are NOT wrapped with this sentinel.

Functions

func NewAPIStorage 

func NewAPIStorage(client *WorkerAPIClient) storage.Storage

NewAPIStorage wraps a platform API client so it satisfies storage.Storage. It lets a remote worker's registry resolve sources, sinks and workflows from the platform API when no local database is available.

Types

type SyncContext 

type SyncContext struct {
	SourceMap map[string]storage.Source
	SinkMap   map[string]storage.Sink
	WorkerID  string
}

type Worker 

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

Worker syncs the state of workflows from storage to the registry.

func NewWorker 

func NewWorker(storage WorkerStorage, registry *registry.Registry) *Worker

NewWorker creates a new worker.

func (*Worker) Deregister 

func (w *Worker) Deregister(ctx context.Context)

Deregister removes the worker entry from storage.

func (*Worker) GetMetrics 

func (w *Worker) GetMetrics() (cpu, mem float64)

GetMetrics returns the worker's current resource usage metrics.

func (*Worker) IsDraining 

func (w *Worker) IsDraining() bool

IsDraining reports whether a graceful shutdown has been requested.

func (*Worker) ReleaseAllLeases 

func (w *Worker) ReleaseAllLeases(ctx context.Context)

ReleaseAllLeases releases all workflow leases held by this worker.

func (*Worker) RequestShutdown 

func (w *Worker) RequestShutdown(id string)

RequestShutdown asks this worker to begin a graceful shutdown if the given id matches its own GUID. It is safe to call concurrently and is a no-op for a different identity.

func (*Worker) SelfRegister 

func (w *Worker) SelfRegister(ctx context.Context) error

SelfRegister registers the worker in the storage if it doesn't already exist.

func (*Worker) SetAdmissionThresholds 

func (w *Worker) SetAdmissionThresholds(cpu, mem float64)

SetAdmissionThresholds overrides this worker's load-shedding thresholds. Zero or negative leaves the process-wide setting in place.

The process-wide values come from the environment and are read once at startup, which is the right shape for an operator setting and the wrong one for anything that needs to differ per worker — a worker sized for bursty CDC wants different headroom from one polling a few APIs, and a test exercising lease failover wants no shedding at all. Passing 1 or above disables that dimension, the same escape hatch the environment variables offer.

func (*Worker) SetLeaseTTL 

func (w *Worker) SetLeaseTTL(ttlSeconds int)

SetLeaseTTL allows configuring the lease TTL in seconds (default 30).

func (*Worker) SetMetrics 

func (w *Worker) SetMetrics(cpu, mem float64)

SetMetrics updates the worker's current resource usage metrics.

func (*Worker) SetRegistrationInfo 

func (w *Worker) SetRegistrationInfo(name, host string, port int, description string)

SetRegistrationInfo sets the information used for self-registration.

func (*Worker) SetShutdownFunc 

func (w *Worker) SetShutdownFunc(fn context.CancelFunc)

SetShutdownFunc registers a callback used to stop the host process after the worker has gracefully drained. It is typically the application's context cancel function so a dedicated worker process exits cleanly.

func (*Worker) SetStorage 

func (w *Worker) SetStorage(s storage.Storage)

SetStorage updates the worker's storage backend.

func (*Worker) SetSyncInterval 

func (w *Worker) SetSyncInterval(d time.Duration)

SetSyncInterval sets how often the worker reconciles workflows from storage.

func (*Worker) SetWorkerCacheTTL 

func (w *Worker) SetWorkerCacheTTL(d time.Duration)

SetWorkerCacheTTL sets the TTL for the worker sharding cache.

func (*Worker) SetWorkerConfig 

func (w *Worker) SetWorkerConfig(workerID, totalWorkers int, workerGUID string, workerToken string)

SetWorkerConfig sets the worker sharding configuration and optional GUID and Token.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) (err error)

func (*Worker) SyncWorkflow 

func (w *Worker) SyncWorkflow(ctx context.Context, wf storage.Workflow, sctx SyncContext)

func (*Worker) TriggerShutdown 

func (w *Worker) TriggerShutdown()

TriggerShutdown invokes the registered shutdown callback, if any, to stop the host process. It is called after the worker has drained.

type WorkerAPIClient 

type WorkerAPIClient struct {
	BaseURL    string
	Token      string
	HTTPClient *http.Client
}

WorkerAPIClient handles communication with the Hermod platform API.

func NewWorkerAPIClient 

func NewWorkerAPIClient(baseURL string, token string) *WorkerAPIClient

func (*WorkerAPIClient) AcquireWorkflowLease 

func (c *WorkerAPIClient) AcquireWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)

Lease APIs for the platform-backed client. The platform does not yet expose dedicated lease endpoints; for an API-backed remote worker, assignment is already enforced by matching wf.WorkerID against the worker's GUID (see isWorkflowAssigned). In that model there is no separate lease backend that can be "lost", so acquire/renew report the lease as owned. Returning false here would make leaseRenewalLoop interpret every tick as "lease lost" and repeatedly tear down healthy engines (closing CDC sources), which is the worst possible default. Treating "no lease backend" as "always owned" keeps engines stable while remaining safe because real ownership is enforced by the explicit worker assignment.

func (*WorkerAPIClient) CreateLog 

func (c *WorkerAPIClient) CreateLog(ctx context.Context, log storage.Log) error

func (*WorkerAPIClient) CreateLogs 

func (c *WorkerAPIClient) CreateLogs(ctx context.Context, logs []storage.Log) error

CreateLogs ships a batch of workflow logs to the platform.

A remote worker has no database, so its engines' logs exist only in its own process output unless they travel here — which is exactly the deployment where reading a worker's console is hardest. This used to be a no-op, and every workflow log a remote worker produced was accepted and discarded.

A platform that predates the batch route answers 404. Workers and platforms are upgraded separately, so fall back to the per-log endpoint rather than going silent for the duration of a rollout.

func (*WorkerAPIClient) CreateWorker 

func (c *WorkerAPIClient) CreateWorker(ctx context.Context, w storage.Worker) error

func (*WorkerAPIClient) DeleteWorker 

func (c *WorkerAPIClient) DeleteWorker(ctx context.Context, id string) error

func (*WorkerAPIClient) GetSink 

func (c *WorkerAPIClient) GetSink(ctx context.Context, id string) (storage.Sink, error)

func (*WorkerAPIClient) GetSource 

func (c *WorkerAPIClient) GetSource(ctx context.Context, id string) (storage.Source, error)

func (*WorkerAPIClient) GetWorker 

func (c *WorkerAPIClient) GetWorker(ctx context.Context, id string) (storage.Worker, error)

func (*WorkerAPIClient) GetWorkflow 

func (c *WorkerAPIClient) GetWorkflow(ctx context.Context, id string) (storage.Workflow, error)

func (*WorkerAPIClient) ListSinks 

func (c *WorkerAPIClient) ListSinks(ctx context.Context, filter storage.CommonFilter) ([]storage.Sink, int, error)

func (*WorkerAPIClient) ListSources 

func (c *WorkerAPIClient) ListSources(ctx context.Context, filter storage.CommonFilter) ([]storage.Source, int, error)

func (*WorkerAPIClient) ListWorkers 

func (c *WorkerAPIClient) ListWorkers(ctx context.Context, filter storage.CommonFilter) ([]storage.Worker, int, error)

func (*WorkerAPIClient) ListWorkflows 

func (c *WorkerAPIClient) ListWorkflows(ctx context.Context, filter storage.CommonFilter) ([]storage.Workflow, int, error)

func (*WorkerAPIClient) Ping 

func (c *WorkerAPIClient) Ping(ctx context.Context) error

Ping performs a one-shot connectivity check against the platform API's /healthz endpoint. It returns an actionable error for the common misconfigurations — most notably a platform-url whose scheme does not match what the origin actually serves — so the worker can fail fast at startup instead of silently failing every sync cycle (see sync.go).

func (*WorkerAPIClient) ReleaseWorkflowLease 

func (c *WorkerAPIClient) ReleaseWorkflowLease(ctx context.Context, workflowID, ownerID string) error

func (*WorkerAPIClient) RenewWorkflowLease 

func (c *WorkerAPIClient) RenewWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)

func (*WorkerAPIClient) UpdateSink 

func (c *WorkerAPIClient) UpdateSink(ctx context.Context, snk storage.Sink) error

func (*WorkerAPIClient) UpdateSource 

func (c *WorkerAPIClient) UpdateSource(ctx context.Context, src storage.Source) error

func (*WorkerAPIClient) UpdateWorkerHeartbeat 

func (c *WorkerAPIClient) UpdateWorkerHeartbeat(ctx context.Context, id string, cpu, mem float64) error

func (*WorkerAPIClient) UpdateWorkflow 

func (c *WorkerAPIClient) UpdateWorkflow(ctx context.Context, wf storage.Workflow) error

func (*WorkerAPIClient) UpdateWorkflowStats 

func (c *WorkerAPIClient) UpdateWorkflowStats(ctx context.Context, id string, processed, errors, lag uint64) error

func (*WorkerAPIClient) UpdateWorkflowStatus 

func (c *WorkerAPIClient) UpdateWorkflowStatus(ctx context.Context, id string, status string) error

type WorkerStorage 

type WorkerStorage interface {
	GetWorker(ctx context.Context, id string) (storage.Worker, error)
	CreateWorker(ctx context.Context, worker storage.Worker) error
	ListWorkflows(ctx context.Context, filter storage.CommonFilter) ([]storage.Workflow, int, error)
	GetWorkflow(ctx context.Context, id string) (storage.Workflow, error)
	UpdateWorkflow(ctx context.Context, wf storage.Workflow) error
	UpdateWorkflowStatus(ctx context.Context, id string, status string) error
	UpdateWorkflowStats(ctx context.Context, id string, processed, errors, lag uint64) error
	GetSource(ctx context.Context, id string) (storage.Source, error)
	GetSink(ctx context.Context, id string) (storage.Sink, error)
	ListSources(ctx context.Context, filter storage.CommonFilter) ([]storage.Source, int, error)
	ListSinks(ctx context.Context, filter storage.CommonFilter) ([]storage.Sink, int, error)
	ListWorkers(ctx context.Context, filter storage.CommonFilter) ([]storage.Worker, int, error)
	UpdateSource(ctx context.Context, src storage.Source) error
	UpdateSink(ctx context.Context, snk storage.Sink) error
	UpdateWorkerHeartbeat(ctx context.Context, id string, cpu, mem float64) error
	DeleteWorker(ctx context.Context, id string) error
	CreateLog(ctx context.Context, log storage.Log) error
	AcquireWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)
	RenewWorkflowLease(ctx context.Context, workflowID, ownerID string, ttlSeconds int) (bool, error)
	ReleaseWorkflowLease(ctx context.Context, workflowID, ownerID string) error
}

WorkerStorage interface subset needed by worker.

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.