controlplane

Documentation

Overview

Package controlplane defines the deployment surface of dispatch: deploy a single service, scale its agent execution nodes, produce tasks for them, and observe the result. It is the composition root — the only package that knows about queues, nodes, sandbox policies, and metrics together; each of those stays ignorant of the others.

Execution is producer/consumer: Submit and SubmitAsync enqueue tasks, and competing consumers dequeue them. Scale adjusts the local consumers; remote consumers (Kubernetes pods, serverless containers running `dispatch work`) attach to the same deployment through the Consumer interface and scale independently under their own orchestrator.

Index

• Variables
• type Consumer
• type ControlPlane
• type DeploymentStatus
• type Memory
  • func NewMemory(factory node.Factory, rec metrics.Recorder) *Memory
  • func (m *Memory) Deploy(ctx context.Context, spec ServiceSpec) error
  • func (m *Memory) Lease(ctx context.Context, name string) (task.Task, error)
  • func (m *Memory) List(ctx context.Context) ([]DeploymentStatus, error)
  • func (m *Memory) Report(ctx context.Context, name string, r task.Result) error
  • func (m *Memory) Result(ctx context.Context, name, taskID string) (task.Result, bool, error)
  • func (m *Memory) Scale(ctx context.Context, name string, replicas int) error
  • func (m *Memory) Spec(_ context.Context, name string) (ServiceSpec, error)
  • func (m *Memory) Status(ctx context.Context, name string) (DeploymentStatus, error)
  • func (m *Memory) Submit(ctx context.Context, name string, t task.Task) (task.Result, error)
  • func (m *Memory) SubmitAsync(ctx context.Context, name string, t task.Task) (string, error)
• type NodeStatus
• type ServiceSpec
  • func (s ServiceSpec) PDP() (sandbox.PDP, error)

Variables

var ErrExists = errors.New("controlplane: deployment already exists")

ErrExists is returned when deploying a name that is already taken.

var ErrNotFound = errors.New("controlplane: deployment not found")

ErrNotFound is returned when a deployment name is unknown.

Types

type Consumer

type Consumer interface {
	// Lease blocks until a task is available for the deployment or ctx
	// is done.
	Lease(ctx context.Context, name string) (task.Task, error)
	// Report records the result of a leased task.
	Report(ctx context.Context, name string, r task.Result) error
}

Consumer is the consumer-facing surface remote workers attach to: lease work, report outcomes.

type ControlPlane

type ControlPlane interface {
	// Deploy creates a deployment from spec and starts its local nodes.
	Deploy(ctx context.Context, spec ServiceSpec) error
	// Scale sets the deployment's local node count.
	Scale(ctx context.Context, name string, replicas int) error
	// Submit enqueues t and blocks until its result arrives.
	Submit(ctx context.Context, name string, t task.Task) (task.Result, error)
	// SubmitAsync enqueues t and returns its task ID immediately.
	SubmitAsync(ctx context.Context, name string, t task.Task) (string, error)
	// Result returns the result for a task if it has arrived.
	Result(ctx context.Context, name, taskID string) (task.Result, bool, error)
	// Spec returns the deployment's service spec (remote consumers fetch
	// it to reconstruct sandbox policies).
	Spec(ctx context.Context, name string) (ServiceSpec, error)
	// Status reports one deployment.
	Status(ctx context.Context, name string) (DeploymentStatus, error)
	// List reports every deployment in lexical name order.
	List(ctx context.Context) ([]DeploymentStatus, error)
}

ControlPlane is the producer-facing surface: deploy, scale, submit, observe.

type DeploymentStatus

type DeploymentStatus struct {
	Name     string       `json:"name"`
	Replicas int          `json:"replicas"`
	Nodes    []NodeStatus `json:"nodes"`
}

DeploymentStatus reports a deployment's current shape. Replicas counts local nodes only; remote consumers are owned and counted by their own orchestrator.

type Memory

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

Memory is an in-process ControlPlane and Consumer host. Deployment state, queues, and results live in memory; local nodes are worker goroutines competing on each deployment's queue, and remote consumers lease from the same queues via Lease/Report. Node creation is delegated to a node.Factory, so the execution substrate can change without touching it.

func NewMemory

func NewMemory(factory node.Factory, rec metrics.Recorder) *Memory

NewMemory returns an empty control plane that creates nodes with factory and records metrics to rec (use metrics.Nop() to disable).

func (*Memory) Deploy

func (m *Memory) Deploy(ctx context.Context, spec ServiceSpec) error

Deploy implements ControlPlane.

func (*Memory) Lease

func (m *Memory) Lease(ctx context.Context, name string) (task.Task, error)

Lease implements Consumer: remote workers compete on the same queue as local ones.

func (*Memory) List

func (m *Memory) List(ctx context.Context) ([]DeploymentStatus, error)

List implements ControlPlane.

func (*Memory) Report

func (m *Memory) Report(ctx context.Context, name string, r task.Result) error

Report implements Consumer. Task metrics for remote executions are recorded here since the remote worker's recorder lives in its own process.

func (*Memory) Result

func (m *Memory) Result(ctx context.Context, name, taskID string) (task.Result, bool, error)

Result implements ControlPlane.

func (*Memory) Scale

func (m *Memory) Scale(ctx context.Context, name string, replicas int) error

Scale implements ControlPlane.

func (*Memory) Spec

func (m *Memory) Spec(_ context.Context, name string) (ServiceSpec, error)

Spec implements ControlPlane.

func (*Memory) Status

func (m *Memory) Status(ctx context.Context, name string) (DeploymentStatus, error)

Status implements ControlPlane.

func (*Memory) Submit

func (m *Memory) Submit(ctx context.Context, name string, t task.Task) (task.Result, error)

Submit implements ControlPlane.

func (*Memory) SubmitAsync

func (m *Memory) SubmitAsync(ctx context.Context, name string, t task.Task) (string, error)

SubmitAsync implements ControlPlane.

type NodeStatus

type NodeStatus struct {
	ID     string `json:"id"`
	Health string `json:"health"`
}

NodeStatus reports one local node's identity and condition.

type ServiceSpec

type ServiceSpec struct {
	// Name uniquely identifies the deployment.
	Name string `json:"name"`
	// Replicas is the initial local node count. Zero means one; deploy
	// with a negative value for no local nodes (remote consumers only).
	Replicas int `json:"replicas,omitempty"`
	// Policies grant each tool its workspace areas and spawn targets. A
	// tool not listed here executes with no workspace access and no
	// ability to spawn (default deny). Compiled into an NGAC graph;
	// mutually exclusive with Access.
	Policies []sandbox.Policy `json:"policies,omitempty"`
	// Access is the full NGAC form: define user attributes grouping
	// agents, object attributes over workspace areas and spawn targets,
	// associations, and prohibitions. Use it when flat per-tool policies
	// cannot express the relationships you need.
	Access *ngac.Spec `json:"access,omitempty"`
}

ServiceSpec declares one service: a set of tools, the capabilities each is granted, and how many local nodes to start with.

func (ServiceSpec) PDP

func (s ServiceSpec) PDP() (sandbox.PDP, error)

PDP compiles the spec's access definition into the decision point that governs its nodes. Exactly one of Policies or Access may be set; an empty spec yields a deny-all PDP.