controllers

package
v0.0.1-dev.137 Latest Latest
Warning

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

Go to latest
Published: Jul 11, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package controllers — RUNE-121 init-step orchestration for the instance controller. Owns iteration order, runIf evaluation, per-step retry policy, and persistence of InitStepState. Runners only execute one step at a time via Runner.RunInit.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type CreateInstanceCall

type CreateInstanceCall struct {
	Service      *types.Service
	InstanceName string
	Ordinal      int
}

CreateInstanceCall records the parameters of a CreateInstance call

type DeleteInstanceCall

type DeleteInstanceCall struct {
	Instance *types.Instance
}

DeleteInstanceCall records the parameters of a DeleteInstance call

type EndpointPublisher

type EndpointPublisher interface {
	// PublishService re-publishes the full Endpoint set for a service.
	// `endpoints` may be empty (the service has no running instances)
	// in which case the publisher should clear/Delete.
	PublishService(ctx context.Context, service *types.Service, endpoints []types.Endpoint) error
	// PublishLocalInstances re-publishes the per-node identity table
	// for `nodeID` so that policy enforcement can map source IPs back
	// to a service identity.
	PublishLocalInstances(ctx context.Context, nodeID string, table map[string]types.InstanceIdentity) error
}

EndpointPublisher is the orchestrator-side surface of the networking data plane. The runed agent supplies an implementation backed by pkg/networking/endpoints + pkg/networking/localinstances.

type ExecCall

type ExecCall struct {
	Instance *types.Instance
	Options  types.ExecOptions
}

ExecCall records the parameters of an Exec call

type FakeDialControllerCall

type FakeDialControllerCall struct {
	Instance *types.Instance
	Port     uint32
}

FakeDialControllerCall records one Dial invocation on the fake controller.

type FakeHealthController

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

FakeHealthController is a lightweight test double for HealthController

func NewFakeHealthController

func NewFakeHealthController() *FakeHealthController

func (*FakeHealthController) AddInstance

func (f *FakeHealthController) AddInstance(service *types.Service, instance *types.Instance) error

func (*FakeHealthController) AddedInstances

func (f *FakeHealthController) AddedInstances() []struct {
	Service  *types.Service
	Instance *types.Instance
}

Helper accessors for assertions

func (*FakeHealthController) GetHealthStatus

func (f *FakeHealthController) GetHealthStatus(ctx context.Context, instanceID string) (*types.InstanceHealthStatus, error)

func (*FakeHealthController) RemoveInstance

func (f *FakeHealthController) RemoveInstance(instanceID string) error

func (*FakeHealthController) RemovedInstanceIDs

func (f *FakeHealthController) RemovedInstanceIDs() []string

func (*FakeHealthController) Start

func (f *FakeHealthController) Start(ctx context.Context) error

func (*FakeHealthController) Stop

func (f *FakeHealthController) Stop() error

type FakeInstanceController

type FakeInstanceController struct {
	CreateInstanceCalls      []CreateInstanceCall
	RetryCreateInstanceCalls []RetryCreateInstanceCall
	RecreateInstanceCalls    []RecreateInstanceCall
	UpdateInstanceCalls      []UpdateInstanceCall
	StopInstanceCalls        []StopInstanceCall
	DeleteInstanceCalls      []DeleteInstanceCall
	RestartInstanceCalls     []RestartInstanceCall
	GetStatusCalls           []string // Instance IDs
	GetLogsCalls             []string // Instance IDs
	ExecCalls                []ExecCall

	// Port-forward (RUNE-122)
	DialCalls    []FakeDialControllerCall
	DialError    error
	LastDialPeer net.Conn

	// Custom behavior options
	CreateInstanceFunc      func(ctx context.Context, service *types.Service, instanceName string, ordinal int) (*types.Instance, error)
	RetryCreateInstanceFunc func(ctx context.Context, service *types.Service, instance *types.Instance) error
	RecreateInstanceFunc    func(ctx context.Context, service *types.Service, instance *types.Instance) (*types.Instance, error)
	UpdateInstanceFunc      func(ctx context.Context, service *types.Service, instance *types.Instance) error
	StopInstanceFunc        func(ctx context.Context, instance *types.Instance) error
	DeleteInstanceFunc      func(ctx context.Context, instance *types.Instance) error
	RestartInstanceFunc     func(ctx context.Context, instance *types.Instance, reason InstanceRestartReason) error
	GetStatusFunc           func(ctx context.Context, instance *types.Instance) (*types.InstanceStatusInfo, error)
	GetLogsFunc             func(ctx context.Context, instance *types.Instance, opts types.LogOptions) (io.ReadCloser, error)
	ExecFunc                func(ctx context.Context, instance *types.Instance, options types.ExecOptions) (types.ExecStream, error)
	IsCompatibleFunc        func(ctx context.Context, instance *types.Instance, service *types.Service) (bool, string)

	// Default error responses
	CreateInstanceError      error
	RetryCreateInstanceError error
	RecreateInstanceError    error
	UpdateInstanceError      error
	StopInstanceError        error
	DeleteInstanceError      error
	RestartInstanceError     error
	GetStatusError           error
	GetLogsError             error
	ExecError                error

	// Mock responses
	ExecStdout   []byte
	ExecStderr   []byte
	ExecExitCode int
	// contains filtered or unexported fields
}

FakeInstanceController implements the InstanceController interface for testing purposes

func NewFakeInstanceController

func NewFakeInstanceController() *FakeInstanceController

NewFakeInstanceController creates a new test instance controller

func (*FakeInstanceController) AddInstance

func (c *FakeInstanceController) AddInstance(instance *types.Instance)

AddInstance allows tests to add an instance directly to the controller's storage

func (*FakeInstanceController) CreateInstance

func (c *FakeInstanceController) CreateInstance(ctx context.Context, service *types.Service, instanceName string, ordinal int) (*types.Instance, error)

CreateInstance records a call to create an instance and returns a predefined or mocked result

func (*FakeInstanceController) DeleteInstance

func (c *FakeInstanceController) DeleteInstance(ctx context.Context, instance *types.Instance) error

DeleteInstance records a call to delete an instance

func (*FakeInstanceController) Dial

func (c *FakeInstanceController) Dial(ctx context.Context, instance *types.Instance, port uint32) (net.Conn, error)

Dial returns one end of a net.Pipe; the peer end is stashed on the fake controller for tests.

func (*FakeInstanceController) Exec

Exec records a call to execute a command in an instance

func (*FakeInstanceController) ExecDebug

func (c *FakeInstanceController) ExecDebug(ctx context.Context, instance *types.Instance, options types.ExecOptions) (types.ExecStream, error)

ExecDebug stubs the debug-sidecar path for tests. The fake doesn't model the sidecar lifecycle (that's covered by the real docker runner tests + manual e2e); we just return a fake stream so callers exercising the orchestrator surface don't blow up.

func (*FakeInstanceController) GetInstance

func (c *FakeInstanceController) GetInstance(ctx context.Context, namespace, instanceID string) (*types.Instance, error)

GetInstance allows tests to retrieve an instance from the controller's storage

func (*FakeInstanceController) GetInstanceByID

func (c *FakeInstanceController) GetInstanceByID(ctx context.Context, namespace, instanceID string) (*types.Instance, error)

GetInstanceByID implements InstanceController interface

func (*FakeInstanceController) GetInstanceLogs

func (c *FakeInstanceController) GetInstanceLogs(ctx context.Context, instance *types.Instance, opts types.LogOptions) (io.ReadCloser, error)

GetInstanceLogs records a call to get instance logs

func (*FakeInstanceController) GetInstanceStatus

func (c *FakeInstanceController) GetInstanceStatus(ctx context.Context, instance *types.Instance) (*types.InstanceStatusInfo, error)

GetInstanceStatus records a call to get instance status

func (*FakeInstanceController) ListInstances

func (c *FakeInstanceController) ListInstances(ctx context.Context, namespace string) ([]*types.Instance, error)

ListInstances records a call to list instances

func (*FakeInstanceController) ListRunningInstances

func (c *FakeInstanceController) ListRunningInstances(ctx context.Context, namespace string) ([]*types.Instance, error)

func (*FakeInstanceController) RecreateInstance

func (c *FakeInstanceController) RecreateInstance(ctx context.Context, service *types.Service, instance *types.Instance) (*types.Instance, error)

RecreateInstance records a call to recreate an instance

func (*FakeInstanceController) RepublishService

func (c *FakeInstanceController) RepublishService(_ context.Context, _ *types.Service)

RepublishService is a no-op on the fake.

func (*FakeInstanceController) RepublishServiceByInstance

func (c *FakeInstanceController) RepublishServiceByInstance(_ context.Context, _ *types.Instance)

RepublishServiceByInstance is a no-op on the fake. Tests that need to observe republish calls should track them via a custom field.

func (*FakeInstanceController) Reset

func (c *FakeInstanceController) Reset()

Reset clears all recorded calls and stored instances

func (*FakeInstanceController) RestartInstance

func (c *FakeInstanceController) RestartInstance(ctx context.Context, instance *types.Instance, reason InstanceRestartReason) error

RestartInstance records a call to restart an instance

func (*FakeInstanceController) RetryCreateInstance

func (c *FakeInstanceController) RetryCreateInstance(ctx context.Context, service *types.Service, instance *types.Instance) error

RetryCreateInstance records a call to retry the create pipeline on an existing record. Default behavior: no-op success (record stays as-is in the fake's instance map). Tests that need to observe retry calls inspect RetryCreateInstanceCalls; tests that need to simulate retry failure set RetryCreateInstanceError or RetryCreateInstanceFunc.

func (*FakeInstanceController) SetEndpointPublisher

func (c *FakeInstanceController) SetEndpointPublisher(_ EndpointPublisher, _ string)

SetEndpointPublisher is a no-op on the fake.

func (*FakeInstanceController) SetEventLog

func (c *FakeInstanceController) SetEventLog(_ events.EventLog)

SetEventLog is a no-op on the fake.

func (*FakeInstanceController) SetMountResolver

func (c *FakeInstanceController) SetMountResolver(_ MountResolver)

SetMountResolver is a no-op on the fake.

func (*FakeInstanceController) StopInstance

func (c *FakeInstanceController) StopInstance(ctx context.Context, instance *types.Instance) error

StopInstance records a call to stop an instance

func (*FakeInstanceController) UpdateInstance

func (c *FakeInstanceController) UpdateInstance(ctx context.Context, service *types.Service, instance *types.Instance) error

UpdateInstance records a call to update an instance

type HealthController

type HealthController interface {
	// Start the health controller
	Start(ctx context.Context) error

	// Stop the health controller
	Stop() error

	// AddInstance adds an instance to be monitored
	AddInstance(service *types.Service, instance *types.Instance) error

	// RemoveInstance removes an instance from monitoring
	RemoveInstance(instanceID string) error

	// GetHealthStatus gets the current health status of an instance
	GetHealthStatus(ctx context.Context, instanceID string) (*types.InstanceHealthStatus, error)
}

HealthController monitors instance health

func NewHealthController

func NewHealthController(logger log.Logger, store store.Store, runnerManager manager.IRunnerManager, instanceController InstanceController) HealthController

NewHealthController creates a new health controller

type InstanceController

type InstanceController interface {
	// GetInstanceByID gets an instance by ID
	GetInstanceByID(ctx context.Context, namespace, instanceID string) (*types.Instance, error)

	// ListInstances lists all instances in a namespace
	ListInstances(ctx context.Context, namespace string) ([]*types.Instance, error)

	// GetRunningInstances lists all running instances
	ListRunningInstances(ctx context.Context, namespace string) ([]*types.Instance, error)

	// CreateInstance creates a new instance for a service. ordinal is the
	// per-replica slot index assigned by the reconciler; it is stored on the
	// instance and drives per-replica volume claimTemplate binding.
	CreateInstance(ctx context.Context, service *types.Service, instanceName string, ordinal int) (*types.Instance, error)

	// RetryCreateInstance re-runs the create pipeline against an
	// existing record that previously failed in a stuck-in-create
	// state (Status=Failed, ContainerEverCreatedAt==nil). Same UUID,
	// same Name. Called by the reconciler when the stuck record's
	// NextCreateAttemptAt backoff has elapsed.
	RetryCreateInstance(ctx context.Context, service *types.Service, instance *types.Instance) error

	// RecreateInstance recreates an instance
	RecreateInstance(ctx context.Context, service *types.Service, instance *types.Instance) (*types.Instance, error)

	// UpdateInstance updates an existing instance
	UpdateInstance(ctx context.Context, service *types.Service, instance *types.Instance) error

	// StopInstance stops an instance temporarily but keeps it in the store
	StopInstance(ctx context.Context, instance *types.Instance) error

	// DeleteInstance marks an instance for deletion and cleans up runner resources
	// The instance will remain in the store with Deleted status until garbage collection
	DeleteInstance(ctx context.Context, instance *types.Instance) error

	// GetInstanceStatus gets the current status of an instance
	GetInstanceStatus(ctx context.Context, instance *types.Instance) (*types.InstanceStatusInfo, error)

	// GetInstanceLogs gets logs for an instance
	GetInstanceLogs(ctx context.Context, instance *types.Instance, opts types.LogOptions) (io.ReadCloser, error)

	// RestartInstance restarts an instance with respect to the service's restart policy
	RestartInstance(ctx context.Context, instance *types.Instance, reason InstanceRestartReason) error

	// Exec executes a command in a running instance
	// Returns an ExecStream for bidirectional communication
	Exec(ctx context.Context, instance *types.Instance, options types.ExecOptions) (types.ExecStream, error)

	// ExecDebug spawns an ephemeral inspection sidecar from a Failed
	// instance's template (image, env, mounts), with the entrypoint
	// overridden to `sleep infinity`, and execs options.Command inside
	// the sidecar. The sidecar is removed when the returned ExecStream
	// is Closed. The original Failed container is never touched.
	ExecDebug(ctx context.Context, instance *types.Instance, options types.ExecOptions) (types.ExecStream, error)

	// Dial opens a TCP connection to the given port on a running
	// instance (RUNE-122). Returns a net.Conn owned by the caller.
	Dial(ctx context.Context, instance *types.Instance, port uint32) (net.Conn, error)

	// SetEndpointPublisher wires the networking data plane (RUNE-063).
	// May be called once at startup; nil-publisher disables publishing.
	SetEndpointPublisher(publisher EndpointPublisher, nodeID string)

	// SetMountResolver wires the agent-side volumes Subsystem
	// (RUNE-069). May be called once at startup; nil disables
	// resolver-first mount-target lookup and falls back to
	// Volume.Handle.
	SetMountResolver(resolver MountResolver)

	// SetEventLog wires the persisted event log (RUNE-126 Phase 2)
	// so status transitions surface in `rune describe`. Nil-safe.
	SetEventLog(eventLog events.EventLog)

	// RepublishServiceByInstance recomputes the data-plane endpoint set
	// for the service owning the given instance. Exposed for callers
	// outside instanceController that change instance reachability
	// (e.g. the health controller promoting Starting→Running on the
	// first readiness pass). Nil-safe when no endpoint publisher is
	// wired; safe to call repeatedly.
	RepublishServiceByInstance(ctx context.Context, instance *types.Instance)

	// RepublishService refreshes the dataplane endpoint set from
	// current store state (including live container IPs). Safe to
	// call on every reconcile tick.
	RepublishService(ctx context.Context, service *types.Service)
	// contains filtered or unexported methods
}

InstanceController manages instance lifecycle

func NewInstanceController

func NewInstanceController(store store.Store, runnerManager manager.IRunnerManager, logger log.Logger) InstanceController

NewInstanceController creates a new instance controller

type InstanceRestartReason

type InstanceRestartReason string
const (
	InstanceRestartReasonManual             InstanceRestartReason = "manual"
	InstanceRestartReasonHealthCheckFailure InstanceRestartReason = "health-check-failure"
	InstanceRestartReasonUpdate             InstanceRestartReason = "update"
	InstanceRestartReasonFailure            InstanceRestartReason = "failure"
)

type MountResolver

type MountResolver interface {
	// MountTargetFor returns the host-side mount target for the named
	// volume on this node, plus a presence flag. A false return means
	// the subsystem has not (yet) mounted the volume locally; callers
	// should treat this as a transient "not ready" condition.
	MountTargetFor(volumeID string) (string, bool)
}

MountResolver is the orchestrator-side surface of the agent's volumes Subsystem. The runed agent supplies an implementation backed by internal/agent/volumes.Subsystem.

type RecreateInstanceCall

type RecreateInstanceCall struct {
	Service  *types.Service
	Instance *types.Instance
}

RecreateInstanceCall records the parameters of a RecreateInstance call

type RestartInstanceCall

type RestartInstanceCall struct {
	Instance *types.Instance
	Reason   InstanceRestartReason
}

RestartInstanceCall records the parameters of a RestartInstance call

type RetryCreateInstanceCall

type RetryCreateInstanceCall struct {
	Service  *types.Service
	Instance *types.Instance
}

RetryCreateInstanceCall records the parameters of a RetryCreateInstance call

type RunningInstance

type RunningInstance struct {
	Instance   *types.Instance
	IsOrphaned bool
	Runner     types.RunnerType
}

RunningInstance represents an instance found running in a runner

type ScalingController

type ScalingController interface {
	// Start the scaling controller
	Start(ctx context.Context) error

	// Stop the scaling controller
	Stop()

	// CreateScalingOperation creates a new scaling operation
	CreateScalingOperation(ctx context.Context, service *types.Service, params types.ScalingOperationParams) error

	// GetActiveOperation gets the active scaling operation for a service
	GetActiveOperation(ctx context.Context, namespace, serviceName string) (*types.ScalingOperation, error)
}

ScalingController manages scaling operations for services

func NewScalingController

func NewScalingController(store store.Store, logger log.Logger) ScalingController

NewScalingController creates a new scaling controller

type ServiceController

type ServiceController interface {
	Start(ctx context.Context) error
	Stop() error
	GetServiceStatus(ctx context.Context, namespace, name string) (*types.ServiceStatusInfo, error)
	UpdateServiceStatus(ctx context.Context, service *types.Service, status types.ServiceStatus) error
	GetServiceLogs(ctx context.Context, namespace, name string, opts types.LogOptions) (io.ReadCloser, error)
	ExecInService(ctx context.Context, namespace, serviceName string, options types.ExecOptions) (types.ExecStream, error)
	DialInService(ctx context.Context, namespace, serviceName string, port uint32) (net.Conn, *types.Instance, error)
	RestartService(ctx context.Context, namespace, serviceName string) (templateGeneration int64, scale int, err error)
	StopService(ctx context.Context, namespace, serviceName string) error
	DeleteService(ctx context.Context, request *types.DeletionRequest) (*types.DeletionResponse, error)
	GetDeletionStatus(ctx context.Context, namespace, name string) (*types.DeletionOperation, error)
	// contains filtered or unexported methods
}

ServiceController implements the Controller interface for service management

func NewServiceController

func NewServiceController(
	store store.Store,
	instanceController InstanceController,
	healthController HealthController,
	logger log.Logger,
) (ServiceController, error)

NewServiceController creates a new service controller

type ServiceInstanceData

type ServiceInstanceData struct {
	Instances         []types.Instance
	OrphanedInstances []*types.Instance // Actual orphaned instance objects
}

ServiceInstanceData contains instances and orphaned instance information

type SnapshotController

type SnapshotController interface {
	Start(ctx context.Context) error
	Stop() error
}

SnapshotController owns the lifecycle of Snapshot resources: it watches the store, resolves the source Volume + StorageClass + driver, drives Pending → Creating → Ready, and on Deleting calls Driver.DeleteSnapshot before removing the row.

Introduced in RUNE-071 (Slice 10a).

func NewSnapshotController

func NewSnapshotController(opts SnapshotControllerOptions) (SnapshotController, error)

NewSnapshotController constructs a SnapshotController.

type SnapshotControllerOptions

type SnapshotControllerOptions struct {
	Store         store.Store
	Logger        log.Logger
	DriverConfigs map[string]map[string]any

	// SecretLookup resolves `secret:...` references inside the
	// merged driver parameters before Snapshot / DeleteSnapshot
	// calls. See RUNE-200 PR 3 and pkg/storage/driverparams.
	SecretLookup driverparams.SecretLookup
}

SnapshotControllerOptions configures the controller.

type StopInstanceCall

type StopInstanceCall struct {
	Instance *types.Instance
}

StopInstanceCall records the parameters of a StopInstance call

type UpdateInstanceCall

type UpdateInstanceCall struct {
	Service  *types.Service
	Instance *types.Instance
}

UpdateInstanceCall records the parameters of an UpdateInstance call

type VolumeController

type VolumeController interface {
	Start(ctx context.Context) error
	Stop() error

	// SetEventLog wires the persisted event log (RUNE-126 Phase 2)
	// so volume status transitions surface in `rune describe`. Nil-safe.
	SetEventLog(eventLog events.EventLog)
}

VolumeController owns the lifecycle of Volume resources: it watches the store, picks a driver from the resolved StorageClass, and drives the Pending → Provisioning → Available state machine. On deletion it honours the reclaim policy (delete via the driver, or leave the handle in place for retain).

Scope for this iteration is intentionally narrow:

  • No claim binding (that lives with the runner-integration ticket; this controller leaves Volume.BoundClaim untouched).

  • No expand reconciliation (the Volume size field is observed but not compared to the existing handle).

  • No snapshot scheduling (handled in RUNE-071).

Introduced in RUNE-069.

func NewVolumeController

func NewVolumeController(opts VolumeControllerOptions) (VolumeController, error)

NewVolumeController constructs a VolumeController.

type VolumeControllerOptions

type VolumeControllerOptions struct {
	Store         store.Store
	Logger        log.Logger
	DriverConfigs map[string]map[string]any

	// SecretLookup resolves `secret:<name>[.namespace.rune]/<key>`
	// references inside StorageClass / Volume parameters before the
	// controller calls the driver. Wired by cmd/runed against the
	// store-backed SecretRepo so rotation takes effect on the next
	// reconcile. nil disables resolution — secret-ref-shaped values
	// then fail Provision with a clear error (see RUNE-200 PR 3).
	SecretLookup driverparams.SecretLookup

	// DefaultStorageClass mirrors the runefile [storage].defaultStorageClass
	// knob. nil keeps the built-in default ("local"); a non-nil pointer
	// to a non-empty name promotes that class to Default:true at boot
	// (creating it as a no-op marker if needed); a non-nil pointer to
	// the empty string disables the cluster default entirely (any
	// existing Default flag is cleared, and Volumes with empty
	// storageClassName fail to resolve).
	DefaultStorageClass *string

	// PreserveOnDelete, when true, demotes ReclaimPolicy:delete to
	// retain for volumes provisioned by the in-tree "local" driver. Has
	// no effect on other drivers.
	PreserveOnDelete bool

	// Provision retry tuning. Zero values fall back to defaults
	// (5 attempts, 2s base, 30s cap) so existing callers don't need
	// updating.
	MaxProvisionAttempts int
	ProvisionBaseBackoff time.Duration
	ProvisionMaxBackoff  time.Duration
}

VolumeControllerOptions configures the controller. DriverConfigs is the per-driver-name configuration block parsed out of the runefile [storage] section — for example `{"local": {"localVolumeRoot": "/var/lib/rune"}}`. nil/missing entries are passed to the factory as nil, which all in-tree drivers tolerate.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL