supervisor

package
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrServiceNotFound is returned when an agent service does not exist.
	ErrServiceNotFound = errors.New("agent service not found")

	// ErrServiceAlreadyExists is returned when attempting to create a service with an existing ID or name.
	ErrServiceAlreadyExists = errors.New("agent service already exists")

	// ErrInstanceNotFound is returned when a service instance is not found.
	ErrInstanceNotFound = errors.New("service instance not found")

	// ErrInstanceAlreadyExists is returned when an instance ID is already registered.
	ErrInstanceAlreadyExists = errors.New("service instance already exists")

	// ErrInvalidServiceSpec is returned when service specification validation fails.
	ErrInvalidServiceSpec = errors.New("invalid agent service specification")

	// ErrServiceTerminated is returned when operating on a terminated service.
	ErrServiceTerminated = errors.New("agent service is terminated")

	// ErrInstanceTerminated is returned when attempting to heartbeat or operate on a terminated instance.
	ErrInstanceTerminated = errors.New("service instance is terminated")

	// ErrNoHealthyInstance is returned when no healthy instance is available to route IPC traffic.
	ErrNoHealthyInstance = errors.New("no healthy service instance available")

	// ErrMaxRestartsExceeded is returned when an instance has reached the maximum allowed restart attempts.
	ErrMaxRestartsExceeded = errors.New("maximum restart retries exceeded")

	// ErrRolloutInProgress is returned when another rollout is already underway.
	ErrRolloutInProgress = errors.New("rollout upgrade already in progress")

	// ErrRolloutFailed is returned when an agent rollout fails health or stability checks.
	ErrRolloutFailed = errors.New("rollout upgrade failed")

	// ErrNoPreviousVersion is returned when attempting to rollback a service that has no prior version recorded.
	ErrNoPreviousVersion = errors.New("no previous service version recorded for rollback")

	// ErrInstanceDraining is returned when routing to an instance that is actively draining.
	ErrInstanceDraining = errors.New("service instance is currently draining")
)

Functions

This section is empty.

Types

type AllServicesStore 

type AllServicesStore interface {
	ListAllServices(context.Context) ([]*Service, error)
}

AllServicesStore is an internal controller enumeration, separate from the tenant-scoped API query.

type BackoffConfig 

type BackoffConfig struct {
	InitialInterval time.Duration `json:"initialInterval"`
	MaxInterval     time.Duration `json:"maxInterval"`
	Factor          float64       `json:"factor"`
	MaxRetries      int           `json:"maxRetries"` // 0 = unlimited retries
}

BackoffConfig defines exponential backoff settings for instance restarts.

func DefaultBackoffConfig 

func DefaultBackoffConfig() BackoffConfig

DefaultBackoffConfig returns sensible defaults for exponential backoff.

func (BackoffConfig) CalculateDelay 

func (b BackoffConfig) CalculateDelay(restartCount int) time.Duration

CalculateDelay computes the backoff duration for a given restart count.

type HealthConfig 

type HealthConfig struct {
	HeartbeatTTL       time.Duration `json:"heartbeatTtl"`
	UnhealthyThreshold int           `json:"unhealthyThreshold"`
	CheckInterval      time.Duration `json:"checkInterval"`
}

HealthConfig configures heartbeat timeouts and failure thresholds.

func DefaultHealthConfig 

func DefaultHealthConfig() HealthConfig

DefaultHealthConfig returns sensible defaults for instance health checks.

type Instance 

type Instance struct {
	ID                  string           `json:"id"`
	ServiceID           string           `json:"serviceId"`
	TenantID            string           `json:"tenantId"`
	Namespace           string           `json:"namespace"`
	AgentID             string           `json:"agentId"`
	AgentVersion        string           `json:"agentVersion,omitempty"`
	RuntimeClass        string           `json:"runtimeClass,omitempty"`
	FencingToken        uint64           `json:"fencingToken,omitempty"`
	TaskID              *uuid.UUID       `json:"taskId,omitempty"`
	LaunchSpec          json.RawMessage  `json:"launchSpec,omitempty"`
	Address             ipc.AgentAddress `json:"address"`
	Phase               InstancePhase    `json:"phase"`
	RestartCount        int              `json:"restartCount"`
	ConsecutiveFailures int              `json:"consecutiveFailures"`
	NextRestartAt       *time.Time       `json:"nextRestartAt,omitempty"`
	LastHeartbeat       time.Time        `json:"lastHeartbeat"`
	DrainingAt          *time.Time       `json:"drainingAt,omitempty"`
	DrainDeadline       *time.Time       `json:"drainDeadline,omitempty"`
	CreatedAt           time.Time        `json:"createdAt"`
	UpdatedAt           time.Time        `json:"updatedAt"`
	TerminatedAt        *time.Time       `json:"terminatedAt,omitempty"`
	ExitCode            int              `json:"exitCode,omitempty"`
	ExitReason          string           `json:"exitReason,omitempty"`
}

Instance represents a running or terminating process instance of an AgentService.

func (*Instance) IsActive 

func (inst *Instance) IsActive() bool

IsActive returns true if the instance is in an active lifecycle state.

func (*Instance) IsDraining 

func (inst *Instance) IsDraining() bool

IsDraining returns true if the instance is currently draining.

func (*Instance) IsHealthy 

func (inst *Instance) IsHealthy() bool

IsHealthy returns true if the instance is currently running and active.

func (*Instance) IsRecovering 

func (inst *Instance) IsRecovering() bool

IsRecovering returns true if the instance is waiting for a restart backoff.

func (*Instance) IsTerminal 

func (inst *Instance) IsTerminal() bool

IsTerminal returns true if the instance has exited and cannot be reused.

type InstancePhase 

type InstancePhase string

InstancePhase indicates the operational state of an individual service instance. 

const (
	InstanceCreated    InstancePhase = "Created"
	InstanceStarting   InstancePhase = "Starting"
	InstanceRunning    InstancePhase = "Running"
	InstanceDegraded   InstancePhase = "Degraded"
	InstanceDraining   InstancePhase = "Draining"
	InstanceStopping   InstancePhase = "Stopping"
	InstanceStopped    InstancePhase = "Stopped"
	InstanceFailed     InstancePhase = "Failed"
	InstanceRecovering InstancePhase = "Recovering"
)

type InstanceSpawner 

type InstanceSpawner interface {
	SpawnInstance(ctx context.Context, svc *Service, inst *Instance) error
	StopInstance(ctx context.Context, svc *Service, inst *Instance) error
}

InstanceSpawner handles the concrete activation, launch, and stopping of service instances.

type MemoryStore 

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

MemoryStore is an in-memory, thread-safe implementation of Store.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new in-memory store.

func NewMemoryStoreWithClock 

func NewMemoryStoreWithClock(clock func() time.Time) *MemoryStore

NewMemoryStoreWithClock creates a MemoryStore with a custom clock.

func (*MemoryStore) CreateInstance 

func (m *MemoryStore) CreateInstance(ctx context.Context, inst *Instance) error

CreateInstance records a new service instance.

func (*MemoryStore) CreateService 

func (m *MemoryStore) CreateService(ctx context.Context, svc *Service) error

CreateService creates an agent service.

func (*MemoryStore) DeleteInstance 

func (m *MemoryStore) DeleteInstance(ctx context.Context, tenantID, serviceID, instanceID string) error

DeleteInstance removes an instance.

func (*MemoryStore) DeleteService 

func (m *MemoryStore) DeleteService(ctx context.Context, tenantID, serviceID string) error

DeleteService removes a service and its instances.

func (*MemoryStore) GetInstance 

func (m *MemoryStore) GetInstance(ctx context.Context, tenantID, serviceID, instanceID string) (*Instance, error)

GetInstance retrieves an instance.

func (*MemoryStore) GetService 

func (m *MemoryStore) GetService(ctx context.Context, tenantID, serviceID string) (*Service, error)

GetService retrieves a service by tenant and ID.

func (*MemoryStore) GetServiceByName 

func (m *MemoryStore) GetServiceByName(ctx context.Context, tenantID, namespace, name string) (*Service, error)

GetServiceByName retrieves a service by tenant, namespace, and name.

func (*MemoryStore) ListAllInstances 

func (m *MemoryStore) ListAllInstances(ctx context.Context, tenantID string) ([]*Instance, error)

ListAllInstances lists all instances for a tenant.

func (*MemoryStore) ListAllServices 

func (m *MemoryStore) ListAllServices(ctx context.Context) ([]*Service, error)

ListAllServices is reserved for the internal controller.

func (*MemoryStore) ListInstances 

func (m *MemoryStore) ListInstances(ctx context.Context, tenantID, serviceID string) ([]*Instance, error)

ListInstances lists all instances belonging to a service.

func (*MemoryStore) ListServices 

func (m *MemoryStore) ListServices(ctx context.Context, tenantID, namespace string) ([]*Service, error)

ListServices lists all services in a tenant (and optionally namespace).

func (*MemoryStore) UpdateInstance 

func (m *MemoryStore) UpdateInstance(ctx context.Context, inst *Instance) error

UpdateInstance updates an instance's state or heartbeat.

func (*MemoryStore) UpdateService 

func (m *MemoryStore) UpdateService(ctx context.Context, svc *Service) error

UpdateService updates an existing service.

type MockSpawner 

type MockSpawner struct {
	Spawned map[string]*Instance
	Stopped map[string]*Instance

	OnSpawn func(ctx context.Context, svc *Service, inst *Instance) error
	OnStop  func(ctx context.Context, svc *Service, inst *Instance) error
	// contains filtered or unexported fields
}

MockSpawner is an in-memory test spawner that records spawned and stopped instances.

func NewMockSpawner 

func NewMockSpawner() *MockSpawner

NewMockSpawner creates a new MockSpawner.

func (*MockSpawner) SpawnCount 

func (m *MockSpawner) SpawnCount() int

SpawnCount returns the total number of spawned instances recorded.

func (*MockSpawner) SpawnInstance 

func (m *MockSpawner) SpawnInstance(ctx context.Context, svc *Service, inst *Instance) error

SpawnInstance records the spawned instance.

func (*MockSpawner) StopCount 

func (m *MockSpawner) StopCount() int

StopCount returns the total number of stopped instances recorded.

func (*MockSpawner) StopInstance 

func (m *MockSpawner) StopInstance(ctx context.Context, svc *Service, inst *Instance) error

StopInstance records the stopped instance.

type Option 

type Option func(*Supervisor)

Option is a functional option for configuring the Supervisor.

func WithClock 

func WithClock(clock func() time.Time) Option

WithClock sets a custom clock for deterministic testing.

func WithIPCService 

func WithIPCService(ipcSvc *ipc.Service) Option

WithIPCService attaches the kernel IPC service for mailbox awareness and autowake.

func WithMailbox 

func WithMailbox(mailbox ipc.Mailbox) Option

WithMailbox attaches a Mailbox for inspecting pending messages on autowake.

func WithSpawner 

func WithSpawner(spawner InstanceSpawner) Option

WithSpawner sets the instance spawner implementation.

type ResourceRequirements 

type ResourceRequirements struct {
	CPU      string `json:"cpu,omitempty"`
	Memory   string `json:"memory,omitempty"`
	LLMSlots int    `json:"llmSlots,omitempty"`
}

ResourceRequirements defines CPU, Memory, and LLM resource allocations.

type RestartPolicy 

type RestartPolicy string

RestartPolicy defines how the supervisor handles crashed instances. 

const (
	RestartAlways    RestartPolicy = "Always"
	RestartOnFailure RestartPolicy = "OnFailure"
	RestartNever     RestartPolicy = "Never"
)

type RolloutConfig 

type RolloutConfig struct {
	MaxUnavailable int           `json:"maxUnavailable"`
	MaxSurge       int           `json:"maxSurge"`
	MinReadyTime   time.Duration `json:"minReadyTime,omitempty"`
}

RolloutConfig controls progressive updates and rollouts.

func DefaultRolloutConfig 

func DefaultRolloutConfig() RolloutConfig

DefaultRolloutConfig returns default rollout settings (maxUnavailable=1, maxSurge=1).

type Router 

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

Router resolves logical AgentAddress targets to concrete, healthy Service instances.

func NewRouter 

func NewRouter(supervisor *Supervisor, store Store) *Router

NewRouter creates a new service-level IPC router.

func (*Router) ResolveAddress 

func (r *Router) ResolveAddress(ctx context.Context, target ipc.AgentAddress) (ipc.AgentAddress, error)

ResolveAddress maps a logical or instance AgentAddress to a running, healthy instance address. If the target address already has a specific InstanceID, it verifies that the instance is alive. If the target address is logical (empty InstanceID), it finds a healthy instance via round-robin. If AutoWake is enabled and no instances are running, it triggers a wakeup pass.

type Service 

type Service struct {
	ID        string        `json:"id"`
	TenantID  string        `json:"tenantId"`
	Namespace string        `json:"namespace"`
	Name      string        `json:"name"`
	AgentID   string        `json:"agentId"`
	Spec      ServiceSpec   `json:"spec"`
	Status    ServiceStatus `json:"status"`
	CreatedAt time.Time     `json:"createdAt"`
	UpdatedAt time.Time     `json:"updatedAt"`
}

Service represents a supervised long-running agent daemon.

func (*Service) LogicalAddress 

func (s *Service) LogicalAddress() ipc.AgentAddress

LogicalAddress returns the service's base logical IPC address.

func (*Service) Validate 

func (s *Service) Validate() error

Validate verifies that the service has valid tenant, namespace, agent IDs, and spec.

type ServiceLockStore 

type ServiceLockStore interface {
	WithServiceLock(context.Context, string, string, func(Store) error) error
}

ServiceLockStore serializes mutations of one service across API and controller processes. The callback receives a store bound to the lock's transaction.

type ServicePhase 

type ServicePhase string

ServicePhase indicates the lifecycle state of an AgentService. 

const (
	ServicePending    ServicePhase = "Pending"
	ServiceActive     ServicePhase = "Active"
	ServiceDegraded   ServicePhase = "Degraded"
	ServiceSuspended  ServicePhase = "Suspended"
	ServiceFailed     ServicePhase = "Failed"
	ServiceTerminated ServicePhase = "Terminated"
)

type ServiceSpec 

type ServiceSpec struct {
	Replicas        int                  `json:"replicas"`
	AgentVersion    string               `json:"agentVersion,omitempty"`
	AgentVersionRef string               `json:"agentVersionRef,omitempty"`
	RuntimeClass    string               `json:"runtimeClass,omitempty"`
	RestartPolicy   RestartPolicy        `json:"restartPolicy"`
	Resources       ResourceRequirements `json:"resources,omitempty"`
	Backoff         BackoffConfig        `json:"backoff"`
	Health          HealthConfig         `json:"health"`
	Rollout         RolloutConfig        `json:"rollout"`
	DrainTimeout    time.Duration        `json:"drainTimeout,omitempty"`
	AutoWake        bool                 `json:"autoWake"`
	ScaleToZeroTTL  time.Duration        `json:"scaleToZeroTtl,omitempty"`
	Labels          map[string]string    `json:"labels,omitempty"`
	WorkloadSpec    json.RawMessage      `json:"workloadSpec,omitempty"`
}

ServiceSpec declares the desired state and policy configuration for an AgentService.

func (*ServiceSpec) Validate 

func (s *ServiceSpec) Validate() error

Validate checks the service specification for correctness.

type ServiceStatus 

type ServiceStatus struct {
	Phase             ServicePhase `json:"phase"`
	DesiredReplicas   int          `json:"desiredReplicas"`
	CurrentReplicas   int          `json:"currentReplicas"`
	ReadyReplicas     int          `json:"readyReplicas"`
	UpdatedReplicas   int          `json:"updatedReplicas"`
	AvailableReplicas int          `json:"availableReplicas"`
	ActiveVersion     string       `json:"activeVersion,omitempty"`
	PreviousVersion   string       `json:"previousVersion,omitempty"`
	RestartCount      int          `json:"restartCount"`
	LastTransitionAt  time.Time    `json:"lastTransitionAt"`
	Message           string       `json:"message,omitempty"`
}

ServiceStatus tracks the observed state of an AgentService.

type Store 

type Store interface {
	// Service operations
	CreateService(ctx context.Context, svc *Service) error
	GetService(ctx context.Context, tenantID, serviceID string) (*Service, error)
	GetServiceByName(ctx context.Context, tenantID, namespace, name string) (*Service, error)
	UpdateService(ctx context.Context, svc *Service) error
	DeleteService(ctx context.Context, tenantID, serviceID string) error
	ListServices(ctx context.Context, tenantID, namespace string) ([]*Service, error)

	// Instance operations
	CreateInstance(ctx context.Context, inst *Instance) error
	GetInstance(ctx context.Context, tenantID, serviceID, instanceID string) (*Instance, error)
	UpdateInstance(ctx context.Context, inst *Instance) error
	DeleteInstance(ctx context.Context, tenantID, serviceID, instanceID string) error
	ListInstances(ctx context.Context, tenantID, serviceID string) ([]*Instance, error)
	ListAllInstances(ctx context.Context, tenantID string) ([]*Instance, error)
}

Store defines persistence operations for Agent Services and their Instances.

type Supervisor 

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

Supervisor manages long-running Agent Services, monitors instance health, handles crash recovery with exponential backoffs, and balances replica counts.

func NewSupervisor 

func NewSupervisor(store Store, opts ...Option) *Supervisor

NewSupervisor constructs a new Supervisor engine.

func (*Supervisor) CreateService 

func (s *Supervisor) CreateService(ctx context.Context, svc *Service) (*Service, error)

func (*Supervisor) DeleteService 

func (s *Supervisor) DeleteService(ctx context.Context, tenant, id string) error

func (*Supervisor) DrainInstance 

func (s *Supervisor) DrainInstance(ctx context.Context, tenant, service, instance string, timeout time.Duration) error

func (*Supervisor) GetService 

func (s *Supervisor) GetService(ctx context.Context, tenantID, serviceID string) (*Service, error)

GetService retrieves a service by tenant and ID.

func (*Supervisor) ListServices 

func (s *Supervisor) ListServices(ctx context.Context, tenantID, namespace string) ([]*Service, error)

ListServices lists services for a tenant and namespace.

func (*Supervisor) Reconcile 

func (s *Supervisor) Reconcile(ctx context.Context, tenantID string) error

Reconcile passes through all services and reconciles their instances.

func (*Supervisor) ReconcileService 

func (s *Supervisor) ReconcileService(ctx context.Context, svc *Service) error

func (*Supervisor) RecordHeartbeat 

func (s *Supervisor) RecordHeartbeat(ctx context.Context, tenant, service, instance string) error

func (*Supervisor) ReportInstanceExit 

func (s *Supervisor) ReportInstanceExit(ctx context.Context, tenant, service, instance string, code int, reason string) error

func (*Supervisor) RestartService 

func (s *Supervisor) RestartService(ctx context.Context, tenant, id string) error

func (*Supervisor) RollbackService 

func (s *Supervisor) RollbackService(ctx context.Context, tenant, id string) (*Service, error)

func (*Supervisor) RolloutUpgrade 

func (s *Supervisor) RolloutUpgrade(ctx context.Context, tenant, id, version string) (*Service, error)

func (*Supervisor) ScaleService 

func (s *Supervisor) ScaleService(ctx context.Context, tenant, id string, replicas int) (*Service, error)

func (*Supervisor) StopService 

func (s *Supervisor) StopService(ctx context.Context, tenant, id string) error

func (*Supervisor) UpdateService 

func (s *Supervisor) UpdateService(ctx context.Context, svc *Service) (*Service, error)

type TaskExecution 

type TaskExecution struct {
	Task    kernelstore.Task
	Attempt *kernelstore.Attempt
	Lease   *kernelstore.Lease
}

TaskExecution is a consistent snapshot of the task and its current runtime owner. A terminal task does not need an active lease.

type TaskRepository 

type TaskRepository interface {
	CreateTask(context.Context, kernelstore.CreateTaskInput) (kernelstore.CreateTaskResult, error)
	GetTask(context.Context, string, uuid.UUID) (kernelstore.Task, error)
	RequestTaskCancellation(context.Context, string, uuid.UUID, int64) (kernelstore.Task, error)
	GetAgentVersionByRef(context.Context, string, string) (kernelstore.AgentVersion, error)
	GetServiceTaskExecution(context.Context, string, uuid.UUID) (TaskExecution, error)
}

TaskRepository is the existing durable task execution boundary, augmented with a read-only snapshot for service supervision.

type TaskSpawner 

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

TaskSpawner submits service instances to normal admission and scheduling. It never launches an ungoverned process inside the control plane.

func NewTaskSpawner 

func NewTaskSpawner(repository TaskRepository) *TaskSpawner

func (*TaskSpawner) PrepareInstance 

func (p *TaskSpawner) PrepareInstance(svc *Service, inst *Instance)

PrepareInstance pins the launch inputs to this restart generation. Its deterministic task ID also recovers a crash between task and instance writes.

func (*TaskSpawner) Rebind 

func (p *TaskSpawner) Rebind(store Store) InstanceSpawner

func (*TaskSpawner) RefreshInstance 

func (p *TaskSpawner) RefreshInstance(ctx context.Context, svc *Service, inst *Instance) error

RefreshInstance derives liveness from the currently fenced runtime lease. Stopping and draining cannot be reversed by a late heartbeat.

func (*TaskSpawner) SpawnInstance 

func (p *TaskSpawner) SpawnInstance(ctx context.Context, svc *Service, inst *Instance) error

func (*TaskSpawner) StopInstance 

func (p *TaskSpawner) StopInstance(ctx context.Context, svc *Service, inst *Instance) error

func (*TaskSpawner) ValidateService 

func (p *TaskSpawner) ValidateService(ctx context.Context, svc *Service) error

ValidateService freezes a canonical launch declaration before any instance can be created. Admission remains authoritative for policy and budgets.

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.