fleet

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package fleet models worker liveness and compatibility without supervising worker processes or implementing queue delivery semantics.

Index

Constants

View Source 
const (
	// MaxIdentityBytes bounds worker status labels.
	MaxIdentityBytes = 256
	// MaxQueuesPerWorker bounds one heartbeat's queue cardinality.
	MaxQueuesPerWorker = 256
	// MaxCapabilitiesPerWorker bounds negotiated capability cardinality.
	MaxCapabilitiesPerWorker = 64
	// MaxWorkerConcurrency bounds reported goroutine concurrency.
	MaxWorkerConcurrency = 1_000_000
)

Variables

View Source 
var ErrInvalidTelemetryConfiguration = errors.New("fleet: invalid telemetry configuration")

ErrInvalidTelemetryConfiguration reports unusable fleet metric settings.

Functions

This section is empty.

Types

type Capability 

type Capability string

Capability identifies one negotiated data-plane management operation. 

const (
	CapabilityWorkerStatus   Capability = "worker_status"
	CapabilityQueueStatus    Capability = "queue_status"
	CapabilityPause          Capability = "pause"
	CapabilityResume         Capability = "resume"
	CapabilityDrain          Capability = "drain"
	CapabilityTerminate      Capability = "terminate"
	CapabilityFailures       Capability = "failures"
	CapabilityDeadLetters    Capability = "dead_letters"
	CapabilityRetry          Capability = "retry"
	CapabilityBulkRetry      Capability = "bulk_retry"
	CapabilityDelete         Capability = "delete"
	CapabilityPurge          Capability = "purge"
	CapabilityReplay         Capability = "replay"
	CapabilityRetentionCount Capability = "retention_count"
	CapabilityRetentionTime  Capability = "retention_time"
	CapabilityRetentionBytes Capability = "retention_bytes"
)

type Compatibility 

type Compatibility struct {
	State            CompatibilityState `json:"state"`
	Enabled          []Capability       `json:"enabled"`
	WorkerOnly       []Capability       `json:"worker_only"`
	ControlPlaneOnly []Capability       `json:"control_plane_only"`
}

Compatibility is the deterministic result of protocol and capability negotiation.

func Negotiate 

func Negotiate(
	supported ProtocolRange,
	worker ProtocolVersion,
	workerCapabilities []Capability,
	controlPlaneCapabilities []Capability,
) Compatibility

Negotiate safely compares a worker report with the control-plane range and enables only capabilities both sides support at a compatible version.

type CompatibilityState 

type CompatibilityState string

CompatibilityState describes a worker relative to the supported protocol range without pretending an incompatible worker is controllable. 

const (
	CompatibilityCompatible  CompatibilityState = "compatible"
	CompatibilityWorkerOlder CompatibilityState = "worker_older"
	CompatibilityWorkerNewer CompatibilityState = "worker_newer"
	CompatibilityUnknown     CompatibilityState = "unknown"
)

type DrainState 

type DrainState string

DrainState describes progress of a graceful drain request. 

const (
	DrainNotRequested DrainState = "not_requested"
	DrainRequested    DrainState = "requested"
	DrainInProgress   DrainState = "in_progress"
	DrainCompleted    DrainState = "completed"
	DrainTimedOut     DrainState = "timed_out"
)

type Heartbeat 

type Heartbeat struct {
	TenantID     string
	WorkerID     string
	Version      string
	StartedAt    time.Time
	ObservedAt   time.Time
	Queues       []string
	Concurrency  uint32
	State        State
	CurrentJobs  uint32
	DrainStatus  DrainState
	Backend      string
	Protocol     ProtocolVersion
	Capabilities []Capability
}

Heartbeat is the latest status reported by a worker.

func (Heartbeat) EffectiveState 

func (h Heartbeat) EffectiveState(now time.Time, staleAfter time.Duration) State

EffectiveState safely classifies a heartbeat at the supplied observation time. Missing, future, and malformed reports are never treated as healthy.

func (Heartbeat) Validate 

func (h Heartbeat) Validate() error

Validate rejects malformed or unbounded worker status reports.

type HeartbeatDisposition 

type HeartbeatDisposition string

HeartbeatDisposition describes how the registry handled one report. 

const (
	HeartbeatAccepted         HeartbeatDisposition = "accepted"
	HeartbeatDuplicate        HeartbeatDisposition = "duplicate"
	HeartbeatReordered        HeartbeatDisposition = "reordered"
	HeartbeatConflict         HeartbeatDisposition = "conflict"
	HeartbeatInvalid          HeartbeatDisposition = "invalid"
	HeartbeatCapacityExceeded HeartbeatDisposition = "capacity_exceeded"
)

type HeartbeatValidationError 

type HeartbeatValidationError struct {
	Field   string
	Problem string
}

HeartbeatValidationError identifies one invalid worker-status field.

func (*HeartbeatValidationError) Error 

func (e *HeartbeatValidationError) Error() string

type Projection 

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

Projection consumes queue lifecycle events without accessing a backend. New queue identities beyond the configured limit are aggregated into the overflow bucket to keep memory use bounded.

func NewProjection 

func NewProjection(maxQueues int) *Projection

NewProjection creates a projection that retains at most maxQueues distinct queue identities. Non-positive limits retain no identities and aggregate all events into the overflow bucket.

func (*Projection) Observe 

func (p *Projection) Observe(event queue.Event)

Observe implements queue.Observer.

func (*Projection) Snapshot 

func (p *Projection) Snapshot() ProjectionSnapshot

Snapshot returns a deterministic copy safe for concurrent readers.

type ProjectionSnapshot 

type ProjectionSnapshot struct {
	Queues   []QueueSnapshot
	Overflow QueueMetrics
}

ProjectionSnapshot is an immutable point-in-time fleet projection.

type ProtocolRange 

type ProtocolRange struct {
	Minimum ProtocolVersion
	Maximum ProtocolVersion
}

ProtocolRange is the inclusive protocol range supported by the control plane during rolling upgrades.

type ProtocolVersion 

type ProtocolVersion struct {
	Major uint16 `json:"major"`
	Minor uint16 `json:"minor"`
}

ProtocolVersion identifies a worker/control-plane management protocol.

type QueueMetrics 

type QueueMetrics struct {
	Enqueued  uint64
	Succeeded uint64
	Failed    uint64
	Runtime   time.Duration
}

QueueMetrics is a bounded current-state projection of lifecycle events. Runtime is cumulative so a metrics backend can derive honest averages and rates without this package retaining an unbounded time series.

type QueueSnapshot 

type QueueSnapshot struct {
	Backend string
	Queue   string
	Metrics QueueMetrics
}

QueueSnapshot identifies one queue's current projected counters.

type Registry 

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

Registry retains a bounded set of latest worker heartbeats.

func NewRegistry 

func NewRegistry(maxWorkers int) *Registry

NewRegistry creates a registry that retains at most maxWorkers identities.

func (*Registry) RegisterTelemetry 

func (r *Registry) RegisterTelemetry(
	meter metric.Meter,
	now func() time.Time,
	staleAfter time.Duration,
) (metric.Registration, error)

RegisterTelemetry exports fixed-cardinality fleet state and rejection measurements. The caller owns the returned registration lifecycle.

func (*Registry) Snapshot 

func (r *Registry) Snapshot(now time.Time, staleAfter time.Duration) RegistrySnapshot

Snapshot returns sorted workers with liveness derived at now.

func (*Registry) SnapshotTenant 

func (r *Registry) SnapshotTenant(
	tenant string,
	now time.Time,
	staleAfter time.Duration,
) RegistrySnapshot

SnapshotTenant returns only one tenant's workers and rejection count.

func (*Registry) Upsert 

func (r *Registry) Upsert(heartbeat Heartbeat) HeartbeatDisposition

Upsert ingests one heartbeat without allowing delayed, duplicate, or conflicting reports to falsely advance worker state.

type RegistrySnapshot 

type RegistrySnapshot struct {
	Workers  []WorkerSnapshot
	Rejected uint64
}

RegistrySnapshot is a deterministic point-in-time view of retained workers.

type State 

type State string

State is the worker state reported to administrative consumers. 

const (
	StateRunning  State = "running"
	StatePaused   State = "paused"
	StateDraining State = "draining"
	StateStopped  State = "stopped"
	StateStale    State = "stale"
	StateUnknown  State = "unknown"
)

type WorkerSnapshot 

type WorkerSnapshot struct {
	Heartbeat
	State State
}

WorkerSnapshot contains a defensively copied heartbeat and its derived fail-safe state.

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.