Documentation
¶
Index ¶
- Constants
- func NewPrometheusProvider() (metric.MeterProvider, error)
- type EventConsumer
- type EventHandler
- type QueueType
- type UnitQueueManager
- func (u *UnitQueueManager) CancelOperation(operationID uuid.UUID) bool
- func (u *UnitQueueManager) ErrorChannel() <-chan error
- func (u *UnitQueueManager) QueueFunctionEvent(ctx context.Context, event api.FunctionWorkerEventRequest, ...)
- func (u *UnitQueueManager) QueueStatusEvent(unitID string, result *api.ActionResult, sendFn func(*api.ActionResult) error)
- func (u *UnitQueueManager) RegisterCancelFunc(operationID uuid.UUID, cancel context.CancelFunc)
- func (u *UnitQueueManager) Start(ctx context.Context)
- func (u *UnitQueueManager) Stop()
- func (u *UnitQueueManager) UnregisterCancelFunc(operationID uuid.UUID)
- type WatcherManager
- func (m *WatcherManager) CancelAll()
- func (m *WatcherManager) IsWatcherActive(unitID uuid.UUID, watchType api.ActionType) bool
- func (m *WatcherManager) Stats() (activeWatchers int, runningWorkers int, idleWorkers int, pendingTasks int)
- func (m *WatcherManager) StopAndWait()
- func (m *WatcherManager) SubmitWatcher(ctx context.Context, unitID uuid.UUID, watchType api.ActionType, ...)
- func (m *WatcherManager) SubmitWatcherAndWait(ctx context.Context, unitID uuid.UUID, watchType api.ActionType, ...)
- type Worker
- func (b *Worker) IsServing() bool
- func (b *Worker) LastEventHandledAt() time.Time
- func (b *Worker) Start(ctx context.Context) error
- func (b *Worker) WaitForPendingOperations()
- func (b *Worker) WithFunctionExecutor(functionExecutor executor.FunctionExecutor) *Worker
- func (b *Worker) WithMetricsMeter(meter metric.Meter) *Worker
- type WorkerFrontdoorClient
Constants ¶
const ( ErrorTypeUnmarshalWorkerEvent = "failed_to_unmarshal_worker_event" ErrorTypeUnmarshalFunctionEvent = "failed_to_unmarshal_function_event" ErrorTypeFunctionCommand = "failed_to_execute_function_command" )
Variables ¶
This section is empty.
Functions ¶
func NewPrometheusProvider ¶
func NewPrometheusProvider() (metric.MeterProvider, error)
Types ¶
type EventConsumer ¶ added in v0.1.95
type EventConsumer struct {
// contains filtered or unexported fields
}
EventConsumer reads the ConfigHub event log over HTTP long-poll and invokes a handler for each delivered fact.
It is a pure consumer, deliberately separate from the bridge/function worker connection (lib.Worker). It authenticates as a worker (bot) identity — which keys its server-held delivery cursor and authorizes what it can observe — but it holds no lease and never dequeues operations. Nothing it does contends with a Target's directed work: it only reads the append-only log by cursor.
The subscription's Name is the cursor name; it goes in the request path. The filter travels in the body each poll — the server stores only the cursor position, so a restarted consumer resumes from its bookmark without keeping any local state.
func NewEventConsumer ¶ added in v0.1.95
func NewEventConsumer(serverURL, workerID, workerSecret string, subscription api.EventSubscription, handler EventHandler) *EventConsumer
NewEventConsumer builds a consumer for the given worker identity and one subscription (its Name is the cursor name). handler is invoked once per delivered event-log fact. Run several consumers for several cursors.
func (*EventConsumer) Run ¶ added in v0.1.95
func (ec *EventConsumer) Run(ctx context.Context) error
Run authenticates, then long-polls the event log until ctx is canceled. It survives transient network/5xx failures with a brief backoff and refreshes its JWT on 401, mirroring the queued_operation poll's resilience.
type EventHandler ¶ added in v0.1.95
type EventHandler = func(ctx context.Context, entry api.EventLogEntry)
EventHandler receives one event-log fact from the event log. It is used by EventConsumer, the standalone event-log reader — event delivery is not part of the bridge/function worker connection.
type UnitQueueManager ¶
type UnitQueueManager struct {
// contains filtered or unexported fields
}
UnitQueueManager manages queues for different units to ensure serialized operations
func NewUnitQueueManager ¶
func NewUnitQueueManager() *UnitQueueManager
NewUnitQueueManager creates a new UnitQueueManager instance
func (*UnitQueueManager) CancelOperation ¶
func (u *UnitQueueManager) CancelOperation(operationID uuid.UUID) bool
CancelOperation cancels a running operation by its ID Returns true if operation was found and cancelled, false otherwise
func (*UnitQueueManager) ErrorChannel ¶
func (u *UnitQueueManager) ErrorChannel() <-chan error
ErrorChannel returns a read-only channel for receiving queue errors Usage example:
go func() {
for err := range manager.ErrorChannel() {
log.Printf("Queue error: %v", err)
}
}()
func (*UnitQueueManager) QueueFunctionEvent ¶
func (u *UnitQueueManager) QueueFunctionEvent(ctx context.Context, event api.FunctionWorkerEventRequest, handler func(api.FunctionWorkerEventRequest))
QueueFunctionEvent enqueues a function worker event to be processed serially for its unit
func (*UnitQueueManager) QueueStatusEvent ¶
func (u *UnitQueueManager) QueueStatusEvent( unitID string, result *api.ActionResult, sendFn func(*api.ActionResult) error, )
QueueStatusEvent queues a status update for async delivery with infinite retry. Returns immediately (non-blocking). Status is delivered in FIFO order per unit.
func (*UnitQueueManager) RegisterCancelFunc ¶
func (u *UnitQueueManager) RegisterCancelFunc(operationID uuid.UUID, cancel context.CancelFunc)
RegisterCancelFunc stores a cancel function for an operation This allows the operation to be canceled later via CancelOperation
func (*UnitQueueManager) Start ¶
func (u *UnitQueueManager) Start(ctx context.Context)
Start initializes the queue manager
func (*UnitQueueManager) Stop ¶
func (u *UnitQueueManager) Stop()
Stop gracefully shuts down the queue manager and all its queues
func (*UnitQueueManager) UnregisterCancelFunc ¶
func (u *UnitQueueManager) UnregisterCancelFunc(operationID uuid.UUID)
UnregisterCancelFunc removes a cancel function when operation completes This should be called via defer to ensure cleanup even if operation fails
type WatcherManager ¶
type WatcherManager struct {
// contains filtered or unexported fields
}
WatcherManager combines watcher lifecycle management with worker pool execution. It ensures only one watcher is active per unit per watch type at any time, with newer watchers canceling and replacing older ones of the same type. This means Apply watchers only replace Apply watchers, and Destroy watchers only replace Destroy watchers. It also manages the goroutine pool for watchers.
func NewWatcherManager ¶
func NewWatcherManager(maxWorkers, maxQueueSize int) *WatcherManager
NewWatcherManager creates a new WatcherManager with the specified pool size
func (*WatcherManager) CancelAll ¶
func (m *WatcherManager) CancelAll()
CancelAll cancels all active watchers (useful for shutdown)
func (*WatcherManager) IsWatcherActive ¶
func (m *WatcherManager) IsWatcherActive(unitID uuid.UUID, watchType api.ActionType) bool
IsWatcherActive checks if a watcher of a specific type is currently active for a unit
func (*WatcherManager) Stats ¶
func (m *WatcherManager) Stats() (activeWatchers int, runningWorkers int, idleWorkers int, pendingTasks int)
Stats returns statistics about the watcher manager
func (*WatcherManager) StopAndWait ¶
func (m *WatcherManager) StopAndWait()
StopAndWait stops the worker pool and waits for all workers to finish
func (*WatcherManager) SubmitWatcher ¶
func (m *WatcherManager) SubmitWatcher(ctx context.Context, unitID uuid.UUID, watchType api.ActionType, task func(context.Context))
SubmitWatcher submits a watcher task for a unit, canceling any existing watcher. The task will be executed in the worker pool with proper lifecycle management. This function returns immediately (async). Use SubmitWatcherAndWait for sync behavior.
func (*WatcherManager) SubmitWatcherAndWait ¶
func (m *WatcherManager) SubmitWatcherAndWait(ctx context.Context, unitID uuid.UUID, watchType api.ActionType, task func(context.Context))
SubmitWatcherAndWait submits a watcher task and waits for it to complete. This keeps the caller blocked until the watcher finishes, which is important for maintaining running operation registration during override detection.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
func (*Worker) IsServing ¶ added in v0.1.35
IsServing reports whether the worker's inbound event loop is currently running. Returns false before Start and after WaitForPendingOperations.
func (*Worker) LastEventHandledAt ¶ added in v0.1.35
LastEventHandledAt returns the time the worker most recently handled an event from the ConfigHub server. Returns the zero time if the worker has not started or has not yet handled an event.
func (*Worker) WaitForPendingOperations ¶
func (b *Worker) WaitForPendingOperations()
WaitForPendingOperations waits for all in-flight operations to complete
func (*Worker) WithFunctionExecutor ¶
func (b *Worker) WithFunctionExecutor(functionExecutor executor.FunctionExecutor) *Worker
type WorkerFrontdoorClient ¶
type WorkerFrontdoorClient struct {
// contains filtered or unexported fields
}
func NewWorkerFrontdoorClient ¶
func NewWorkerFrontdoorClient(serverURL, serverPort, workerID, workerSecret string) *WorkerFrontdoorClient
NewWorkerFrontdoorClient creates a new WorkerFrontdoorClient
func (*WorkerFrontdoorClient) Close ¶
func (w *WorkerFrontdoorClient) Close() error
func (*WorkerFrontdoorClient) GetClient ¶
func (w *WorkerFrontdoorClient) GetClient() *goclientnew.ClientWithResponses
getClient returns the authenticated client (thread-safe)
Source Files