manager

package
v0.22.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: AGPL-3.0 Imports: 51 Imported by: 0

Documentation

Overview

Package manager implements the APO-796 ServiceManager: the control-plane Service->ServiceRevision minting reconciler (platform-neutral, this file) and the data-plane resident reconciler (resident_reconciler.go) that drives the workerd resident and publishes this node's serveable routing. The minting reconciler runs inside the apiserver via apiserver.WithAdditionalController; the resident reconciler runs inside cmd/workerd-manager next to the runsc host and is strictly read-only on the API.

Index

Constants

View Source 
const (
	DefaultRootDir      = "/run/workerd-manager/root"
	DefaultImageBaseDir = "/run/workerd-manager/images"
	DefaultListenAddr   = "*:8080"
	DefaultEgressDir    = "/run/workerd-manager/egress"
)

Default locations for the resident dirs and the dispatcher listener under the shared /run/workerd-manager volume. Exported so multicluster embedders (the apoxy-cloud shared-shard workerd-manager) register flags with the same defaults instead of re-hardcoding the literals.

View Source 
const EgressFullPassRequestName = "egress-config"

EgressFullPassRequestName is the name of the synthetic singleton request every egress-relevant event coalesces into. The reconcilers ignore it and recompute the whole tenant, so bursts of events collapse into one pass; the apoxy-cloud multicluster pusher reuses this name for the same purpose.

Variables

This section is empty.

Functions

func EgressSocketPath 

func EgressSocketPath(dir, tenant string) string

EgressSocketPath is the deterministic per-tenant egress control socket under dir — deterministic so the backplane reconciler needs no discovery, keyed by the resident sandbox id so tenants can never collide.

func ResolveRegistryTag added in v0.22.0 

func ResolveRegistryTag(ctx context.Context, ref computev1alpha1.BundleRef) (string, error)

ResolveRegistryTag asks the registry which immutable digest a tag currently names. Resolution happens in the control plane, not per node, so that every node running the minted revision pulls the identical digest — a tag resolved per-node would let two backplanes serve different code under one revision.

The repository comes from host.BundleRepositoryFor rather than being built here, so it picks up the BundleRef's own credentials and whatever platform TLS the process installed (host.SetPlatformPullTLS) — an open-coded bundle.NewRepository silently drops the latter and 401s on exactly the platform-registry bundles. A control plane that holds no registry identity of its own resolves those through its own path instead; see WithTagResolver.

func Run 

func Run() error

Run is the workerd-manager entry point, invoked after sandbox.DispatchRunsc(). It brings up the single-project resident workerd (the empty tenant), serves the dispatcher control channel, and runs the resident reconciler against the project apiserver. It blocks until signalled.

This is the data-plane half of APO-796 (the minting reconciler runs in the apiserver via apiserver.WithAdditionalController), driving ResidentManager as its single-tenant degenerate case; the shared backplane drives the same manager per engaged project via ReconcileWithClient. It only runs meaningfully on linux — host.NewResidentFactory needs the gVisor core — but compiles everywhere so the package is unit-testable with fakes on darwin.

Types

type BundleFetcher 

type BundleFetcher interface {
	// Bundle returns the bundle's BundleManifest (the OCI config blob) and
	// each module's bytes keyed by cleaned in-layer path
	// (host.CleanModulePath), matching BundleManifest.Modules[i].Path.
	// Ref resolution and pull credentials both come from the BundleRef.
	Bundle(ctx context.Context, b computev1alpha1.BundleRef) (computev1alpha1.BundleManifest, map[string][]byte, error)
}

BundleFetcher pulls a bundle from an OCI registry. It is the registry seam: the resident reconciler and control server are fake-testable without a live registry.

type ControlServer 

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

ControlServer is the manager side of the dispatcher control channel: an HTTP server that serves WorkerCode payloads the resident's WorkerLoader callback pulls. It listens on a host loopback TCP address; the clrk control forwarder bridges the dispatcher's in-sandbox connections to it (see host.ResidentConfig). It is TCP, not AF_UNIX: the Sentry's plugin seccomp only allows socket() for AF_INET/AF_INET6, so the forwarder cannot dial a host unix socket.

One ControlServer serves exactly one tenant's resident: its Store is the isolation boundary, so a dispatcher can only ever resolve services of the project whose control address was sealed into its sandbox spec.

func NewControlServer 

func NewControlServer(store *Store) *ControlServer

NewControlServer returns a control server backed by store.

func (*ControlServer) Close 

func (c *ControlServer) Close() error

Close releases the bound listener for a server whose Serve was never started (an assembly-failure path); Serve's own shutdown closes it otherwise.

func (*ControlServer) Handler 

func (c *ControlServer) Handler() http.Handler

Handler is the control HTTP handler (exported for tests via httptest).

func (*ControlServer) Listen 

func (c *ControlServer) Listen(addr string) (string, error)

Listen binds the control listener on the host loopback TCP address addr and returns the concrete bound address. Split from Serve so a per-tenant caller can bind an ephemeral port ("127.0.0.1:0") and learn the real address BEFORE the resident's sandbox spec is sealed (ControlHostAddr is baked in at Create time and cannot change for the sandbox's lifetime).

func (*ControlServer) Serve 

func (c *ControlServer) Serve(ctx context.Context) error

Serve serves the control API on the listener bound by Listen until ctx is cancelled. The address is in the manager's own netns, which the Sentry's control forwarder shares, so a guest dispatcher reaches it through the forwarder's host TCP dial.

type EgressControlServer 

type EgressControlServer struct {
	workerdv1.UnimplementedEgressConfigServer
	workerdv1.UnimplementedDNSConfigServer
	// contains filtered or unexported fields
}

EgressControlServer is the manager side of the egress config plane (APO-723): a per-tenant gRPC server the backplane's ServiceReconciler (APO-726) pushes compiled egress config through. It fans each apply out to the resident's EgressController live setters via host.EgressApplier.

Unlike the dispatcher control channel (ControlServer, which the SANDBOX reaches through the Sentry's control forwarder and therefore must be loopback TCP), this listener is dialed host-side only — so it is a unix domain socket, and filesystem permissions on the socket directory are the auth boundary. One server serves exactly one tenant: a request naming any sandbox other than that tenant's resident is rejected, so a reconciler can never push config across projects.

func NewEgressControlServer 

func NewEgressControlServer(tenant string, applier host.EgressApplier, dnsApplier host.DNSApplier) *EgressControlServer

NewEgressControlServer returns an egress control server for tenant, fanning applies out through applier (the tenant's resident). dnsApplier (usually the same resident) receives DNSConfig pushes; nil disables that service's applies.

func (*EgressControlServer) ApplyDNS 

func (s *EgressControlServer) ApplyDNS(ctx context.Context, req *workerdv1.ApplyDNSRequest) (*workerdv1.ApplyDNSResponse, error)

ApplyDNS implements workerdv1.DNSConfigServer.

func (*EgressControlServer) ApplyEgress 

func (s *EgressControlServer) ApplyEgress(ctx context.Context, req *workerdv1.ApplyEgressRequest) (*workerdv1.ApplyEgressResponse, error)

ApplyEgress implements workerdv1.EgressConfigServer.

func (*EgressControlServer) Close 

func (s *EgressControlServer) Close() error

Close releases the bound listener for a server whose Serve was never started (an assembly-failure path); Serve's own shutdown (grpc Server.Stop) closes it otherwise. Idempotent — a re-run teardown must be a no-op, and in particular must never touch the socket path again: the deterministic path may already belong to a rebuilt successor server. Socket-file cleanup needs no explicit remove — *net.UnixListener unlinks its file on close, and Listen reaps a leftover from a killed process.

func (*EgressControlServer) Listen 

func (s *EgressControlServer) Listen(path string) error

Listen binds the unix domain socket at path. A stale socket file from a previous incarnation is removed first — the manager is the only legitimate binder of the path, so an existing file is always leftover, never live contention worth preserving.

func (*EgressControlServer) Serve 

func (s *EgressControlServer) Serve(ctx context.Context) error

Serve serves the egress control API on the listener bound by Listen until ctx is cancelled.

type EgressPlan 

type EgressPlan struct {
	// Services is sorted by name so pushes are deterministic.
	Services []ServiceEgressPlan
	Gateways []GatewayEgressPlan
	Routes   []RouteEgressPlan
}

EgressPlan is one tenant's full compiled egress state.

func CompileEgress 

func CompileEgress(services []ServiceEgressInput, gateways []computev1alpha1.EgressGateway, routes []computev1alpha1.EgressRoute) *EgressPlan

CompileEgress compiles one tenant's egress state. Inputs are the Services' resolved egress selections plus every EgressGateway and EgressRoute in the project. It is pure: no client, no I/O.

func (*EgressPlan) WireConfigs 

func (p *EgressPlan) WireConfigs() []*workerdv1.ServiceEgressConfig

WireConfigs returns the per-Service wire planes for the ApplyEgress push.

type EgressPusher 

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

EgressPusher is the data-plane half of the egress reconciler (APO-726): it compiles the tenant's egress plan from the project apiserver and pushes it to the CO-LOCATED resident over the per-tenant egress control socket (APO-723). One pusher instance runs per workerd-manager process and serves every tenant that process hosts; each pod pushes only to its own residents, so nothing is coordinated across pods. It is read-only on the API — status is written by the control-plane EgressStatusReconciler.

func NewEgressPusher 

func NewEgressPusher(egressDir string) *EgressPusher

NewEgressPusher returns a pusher over the egress control sockets under egressDir (the same directory the ResidentManager binds them in).

func (*EgressPusher) ReconcileWithClient 

func (p *EgressPusher) ReconcileWithClient(ctx context.Context, tenant string, c client.Client, _ ctrl.Request) (ctrl.Result, error)

ReconcileWithClient compiles the tenant's egress plan with the given project-apiserver client and pushes it to the tenant's resident. The request is ignored: any egress-relevant event triggers a full push (the wire contract is whole-state, level-triggered). The multicluster wrapper in apoxy-cloud drives this per engaged project, mirroring ResidentManager.ReconcileWithClient.

func (*EgressPusher) StopTenant 

func (p *EgressPusher) StopTenant(tenant string)

StopTenant drops the tenant's pusher state on project disengage: the generation counter and the cached control-socket connection.

func (*EgressPusher) TenantReconciler 

func (p *EgressPusher) TenantReconciler(tenant string, c client.Client) reconcile.Func

TenantReconciler adapts the pusher to a plain reconcile.Func over one fixed tenant and client — the single-project (dedicated/dev) registration, mirroring ResidentManager.TenantReconciler.

type EgressStatusReconciler 

type EgressStatusReconciler struct {
	client.Client
}

EgressStatusReconciler is the control-plane half of the egress reconciler (APO-726): it resolves each Service's egress selection against the project's EgressGateways/EgressRoutes and writes the resulting status — EgressReady on Services, Ready + per-listener attachment counts on EgressGateways, parents on EgressRoutes. It registers into the project apiserver via apiserver.WithAdditionalController, next to the minting ServiceReconciler, so status has exactly one writer (the data-plane pusher in workerd-manager stays read-only on the API, per the conditions.go contract).

Every reconcile is a full tenant pass over the (small) egress object set: the inputs are all-to-all — one gateway edit changes every attached Service's condition — so per-object requests would recompute the same plan anyway. No-op writes are suppressed.

func NewEgressStatusReconciler 

func NewEgressStatusReconciler(c client.Client) *EgressStatusReconciler

NewEgressStatusReconciler returns an EgressStatusReconciler.

func (*EgressStatusReconciler) Reconcile 

func (r *EgressStatusReconciler) Reconcile(ctx context.Context, _ reconcile.Request) (ctrl.Result, error)

Reconcile recomputes the tenant's egress plan and writes every object's status. The request is ignored: any egress-relevant event triggers a full pass.

func (*EgressStatusReconciler) SetupWithManager 

func (r *EgressStatusReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error

SetupWithManager registers the reconciler. Everything egress-relevant funnels into the singleton full-pass request.

type GatewayEgressPlan 

type GatewayEgressPlan struct {
	Name    string
	Ready   bool
	Reason  string
	Message string
	// AttachedRoutes counts attached EgressRoutes per listener name.
	AttachedRoutes map[string]int32
}

GatewayEgressPlan is the compile result for one EgressGateway object: the Ready condition and per-listener attachment counts the control plane writes. Listener data-plane fields (port, backendAddress) are owned by the gateway data-plane materializer and preserved, never computed here.

type ResidentBuilder 

type ResidentBuilder interface {
	NewResident(tenant, controlHostAddr string) (host.ResidentRuntime, error)
}

ResidentBuilder constructs per-tenant residents. *host.ResidentFactory is the production implementation; tests fake it on any platform.

type ResidentManager 

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

ResidentManager owns the per-tenant workerd residents of one manager process. Each tenant (project UUID; "" for single-project topologies) gets its own resident sandbox, warm Store, and control listener — the hard isolation boundary: a tenant's dispatcher can only reach the control server whose address was sealed into its own sandbox spec, so it can never resolve another project's services.

It follows the tunnelproxy embedding pattern (TunnelServer. ReconcileWithClient): the core is client-parameterized and holds NO multicluster machinery. Single-project callers (Run) drive it with the empty tenant and their own client; apoxy-cloud's shared backplane wraps it in an mcreconcile.Reconciler that resolves req.ClusterName to a cluster client and delegates here, and calls StopTenant when a project disengages from the shard. Tenant entries are created lazily on first reconcile, so an engaged project with no ServiceRevisions costs no sandbox.

func NewResidentManager 

func NewResidentManager(factory ResidentBuilder, controlAddr string, opts ...ResidentManagerOption) *ResidentManager

NewResidentManager returns a manager over factory. controlAddr is the fixed control address used for the empty tenant; project tenants always bind ephemeral loopback ports.

func SetupResidents 

func SetupResidents(ctx context.Context, cfg host.ResidentConfig, controlAddr string, opts ...ResidentManagerOption) (*ResidentManager, error)

SetupResidents prepares the host for resident workerds and returns the ResidentManager over them: it validates the config, creates the state/ staging/image dirs, starts the PID-1 child reaper, builds the resident factory, and sweeps orphaned sandboxes (exactly once, before any resident exists). Both Run (single-project) and multicluster embedders call this so the host bootstrap cannot drift between the two topologies. controlAddr is the fixed control bind for the empty tenant; shared shards pass "" (every project tenant gets an ephemeral loopback port).

func (*ResidentManager) Close 

func (m *ResidentManager) Close(ctx context.Context) error

Close stops every tenant. Used at process shutdown; per-tenant errors are joined so one failing teardown doesn't hide the rest.

func (*ResidentManager) EnsureTenant 

func (m *ResidentManager) EnsureTenant(ctx context.Context, tenant string, c client.Client) (*host.ResidentInstance, error)

EnsureTenant creates the tenant's entry if needed and brings its resident up eagerly, returning the running instance. Single-project Run uses it for boot-time fail-fast; shared shards skip it and let the first ServiceRevision reconcile bring the resident up lazily.

func (*ResidentManager) ReconcileWithClient 

func (m *ResidentManager) ReconcileWithClient(ctx context.Context, tenant string, c client.Client, req ctrl.Request) (ctrl.Result, error)

ReconcileWithClient refreshes one tenant using its project-scoped client. It lists revisions before creating tenant state so a stale client cannot resurrect a disengaged tenant.

func (*ResidentManager) StopTenant 

func (m *ResidentManager) StopTenant(ctx context.Context, tenant string) error

StopTenant tears down a tenant's assembly: control serve goroutine, listener, resident sandbox, and staged config. Idempotent — stopping an unknown tenant is a no-op, and a failed teardown leaves the entry in place (marked stopping, rejected by reconciles) so a retry actually retries instead of silently orphaning a running sandbox. The multicluster wrapper calls this when the project disengages from the shard.

While it runs, the tenant is marked draining: entry creation (reconcile or done-watcher rebuild) is refused, and it loops until no entry remains, so a rebuild racing the stop cannot escape it.

func (*ResidentManager) TenantReconciler 

func (m *ResidentManager) TenantReconciler(tenant string, c client.Client) reconcile.Func

TenantReconciler adapts one tenant + client pair to a reconcile.Reconciler for controller-runtime registration (the single-project path). Multicluster callers skip this and call ReconcileWithClient with the per-request cluster client instead.

type ResidentManagerOption 

type ResidentManagerOption func(*ResidentManager)

ResidentManagerOption customizes a ResidentManager.

func WithEgressDir 

func WithEgressDir(dir string) ResidentManagerOption

WithEgressDir enables the egress config plane (APO-723): each tenant gets an EgressConfig gRPC server on a unix domain socket under dir (EgressSocketPath). The directory is created on first use with 0700 — filesystem permissions are the plane's auth boundary.

type ResidentReconciler 

type ResidentReconciler struct {
	client.Client
	// contains filtered or unexported fields
}

ResidentReconciler maintains one tenant's local routes and warmed definitions.

func NewResidentReconciler 

func NewResidentReconciler(c client.Client, resident host.ResidentRuntime, store *Store) *ResidentReconciler

NewResidentReconciler returns a resident reconciler driving resident + store.

func (*ResidentReconciler) Reconcile 

func (r *ResidentReconciler) Reconcile(ctx context.Context, _ reconcile.Request) (ctrl.Result, error)

Reconcile refreshes local routing and schedules a periodic resync. It runs under a single queue key covering the whole tenant, so it refreshes every service rather than the one named in the request.

type Resolver 

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

Resolver turns a dispatcher demux id ("<service>:<revision>") into the host.WorkerDefinition the WorkerLoader callback consumes: it loads the ServiceRevision, pulls its bundle, and reconstructs the WorkerCode payload.

The resident is per-tenant — a Resolver's kube client is scoped to the single project its resident serves — so the id carries no project qualifier; the service name alone keys the WorkerLoader cache within this resident.

The client is swappable (setClient): the per-tenant client is handed in on every ReconcileWithClient call, and a re-engaged project may arrive with a fresh client while the tenant's warm state (and this Resolver, reachable from the control server's pull path) lives on.

func NewResolver 

func NewResolver(c client.Client) *Resolver

NewResolver returns a Resolver over the production OCI fetcher. The client must be scoped to the single project the resident serves.

func (*Resolver) Resolve 

func (r *Resolver) Resolve(ctx context.Context, id string) (host.WorkerDefinition, error)

Resolve builds the WorkerDefinition for a demux id. A missing ServiceRevision is reported as errRevisionNotFound (wrapped) so the control server can answer 404; pull/build failures propagate as-is (a 502 the dispatcher surfaces).

type RouteEgressPlan 

type RouteEgressPlan struct {
	Name    string
	Parents []gwapiv1.RouteParentStatus
}

RouteEgressPlan is the compile result for one EgressRoute: its status.parents entries.

type ServiceEgressInput 

type ServiceEgressInput struct {
	// Name is the compute Service name.
	Name string
	// Egress is the selection; nil means no egress block (the project
	// "default" gateway).
	Egress *computev1alpha1.ServiceEgress
}

ServiceEgressInput is one Service's egress selection, resolved by the caller: from the live ServiceRevision's template when one is serving (enforcement must match what serves), else from the Service template.

type ServiceEgressPlan 

type ServiceEgressPlan struct {
	// Name is the compute Service name.
	Name string
	// Gateway is the resolved gateway name; empty when egress is disabled.
	Gateway string
	// Ready and Reason/Message carry the EgressReady condition
	// (computev1alpha1.EgressReadyReason*).
	Ready   bool
	Reason  string
	Message string
	// Config is the compiled wire plane pushed to the resident.
	Config *workerdv1.ServiceEgressConfig
}

ServiceEgressPlan is the compile result for one Service: its wire plane and the EgressReady condition the control plane writes.

type ServiceReconciler 

type ServiceReconciler struct {
	client.Client
	// contains filtered or unexported fields
}

ServiceReconciler mints immutable ServiceRevisions from a Service's spec.template + spec.source, tracks LatestRevision/LiveRevision, and GCs old revisions. It is platform-neutral (no runsc, no workerd) and registers into the apiserver's manager via apiserver.WithAdditionalController.

func NewServiceReconciler 

func NewServiceReconciler(c client.Client, opts ...ServiceReconcilerOption) *ServiceReconciler

NewServiceReconciler returns a ServiceReconciler. The scheme is captured from the manager in SetupWithManager.

func (*ServiceReconciler) Reconcile 

func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (ctrl.Result, error)

Reconcile mints/promotes/GCs revisions for one Service.

func (*ServiceReconciler) SetupWithManager 

func (r *ServiceReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error

SetupWithManager registers the reconciler with the manager.

type ServiceReconcilerOption added in v0.22.0 

type ServiceReconcilerOption func(*ServiceReconciler)

ServiceReconcilerOption configures a ServiceReconciler.

func WithTagResolver added in v0.22.0 

func WithTagResolver(resolve TagResolver) ServiceReconcilerOption

WithTagResolver replaces the default direct-to-registry tag resolution.

The default reaches the registry over the network with whatever credentials the BundleRef carries, which is right for a customer's own registry and wrong for the platform registry: a hosted control plane authenticates to its own registry out of band, not with a credential stored on the object. That wiring is deployment-specific, so it is injected rather than assumed here.

type Store 

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

Store caches resolved WorkerDefinitions by demux id. The resident reconciler warms it (gating ServiceRevision readiness on a successful resolve), and the control server reads it on the dispatcher's pull path so a warmed revision is served without a second registry round-trip.

The cache is an optimization, not the source of truth: a cold Get resolves on demand. WorkerLoader caches the isolate by id on the workerd side, so the control server is only hit on a dispatcher cache miss (first request per revision, or after a resident restart).

func NewStore 

func NewStore(resolver *Resolver) *Store

NewStore returns a Store backed by resolver.

func (*Store) Get 

func (s *Store) Get(ctx context.Context, id string) (host.WorkerDefinition, error)

Get returns the cached definition for id, resolving (and caching) on a miss.

func (*Store) Invalidate 

func (s *Store) Invalidate(id string)

Invalidate drops a cached definition (the revision was deleted). The workerd isolate idles out on its own; M1 issues no explicit unload.

func (*Store) Warm 

func (s *Store) Warm(ctx context.Context, id string) (host.WorkerDefinition, error)

Warm resolves id and caches the result. Returns the resolve error unchanged (the caller decides readiness / HTTP status).

type TagResolver added in v0.22.0 

type TagResolver func(context.Context, computev1alpha1.BundleRef) (string, error)

TagResolver turns a tag-bearing BundleRef into the digest the tag names.

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.