worker

package
v0.10.9 Latest Latest
Warning

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

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 39 Imported by: 0

Documentation

Overview

internal/worker/ca.go

internal/worker/cert_holder.go

internal/worker/client.go

internal/worker/registry.go

Index

Constants

View Source 
const ControlPlaneCommonName = "shinyhub-control-plane"

ControlPlaneCommonName is the Subject CommonName the control plane's client and server certificates carry (see ControlClientCertificate / ServerCertificate). A worker agent listener pins inbound peers to this identity so one worker's own CA-signed certificate cannot be presented as a client to command another worker.

View Source 
const ProviderRemoteDocker = "remote_docker"

ProviderRemoteDocker labels replicas started on a remote Docker worker.

Variables

View Source 
var ErrControlPlaneUnavailable = errors.New("control plane unavailable (not owner)")

ErrControlPlaneUnavailable is returned by Register when the control plane responds 503: it is reachable but not the ready owner (a standby, or an owner still warming up). Callers retry it, unlike auth/validation failures (401/400) which are permanent.

Functions

func CAKeyEncKey added in v0.9.2 

func CAKeyEncKey(authSecret string) []byte

CAKeyEncKey derives the key-encryption key that protects the worker CA private key at rest, from the auth secret. Exposed so auth.secret-rotation tooling can re-encrypt the stored CA key without duplicating the domain-separation string.

func Fingerprint 

func Fingerprint(cert *x509.Certificate) string

Fingerprint returns the hex SHA-256 of a certificate's DER, used to record the trusted client cert on the worker row.

func IsControlPlaneClientCert added in v0.9.6 

func IsControlPlaneClientCert(cert *x509.Certificate) bool

IsControlPlaneClientCert reports whether cert is the control plane's client certificate rather than a worker's own cert. A worker leaf always carries a namespaced node DNS SAN (<nodeid>.node.shinyhub.internal); the control-plane client cert carries none. Requiring the control-plane CN AND the absence of a node SAN rejects a worker even if its assigned node id ever collided with the control-plane CN.

func NewRemoteRuntime 

func NewRemoteRuntime(lookup WorkerLookup, tier string, dialer AgentDialer) process.Runtime

NewRemoteRuntime builds a tier-bound runtime that delegates to whichever worker is currently live for the tier, dialing it over the mTLS tunnel.

func NewReplicaServer 

func NewReplicaServer(cfg ReplicaServerConfig) *replicaServer

NewReplicaServer constructs a replicaServer from the given config. If AllocatePort is nil, deploy.AllocatePort is used.

func NodeIDFromCert 

func NodeIDFromCert(cert *x509.Certificate) string

NodeIDFromCert recovers the node id bound into a worker certificate, preferring the namespaced DNS SAN and falling back to the CN.

func Register 

func Register(ctx context.Context, serverURL string, req workerapi.RegisterRequest, caPEM []byte) (workerapi.RegisterResponse, error)

Register performs the join: POST the token + CSR over HTTPS pinned to caPEM (no client cert yet at join time). Returns the signed cert, CA bundle, and node id.

Types

type AgentDialer 

type AgentDialer interface {
	// DialWorker returns a client whose transport authenticates to the worker
	// and the base URL (scheme://host) to prefix request paths with.
	DialWorker(w db.Worker) (*http.Client, string, error)
	// Transport returns the RoundTripper used to reach the given worker's data
	// plane (for the proxy and health checks).
	Transport(w db.Worker) (http.RoundTripper, error)
}

AgentDialer returns an HTTP client and base URL for talking to a worker over its mTLS tunnel. It is a seam: production builds an mTLS client keyed by the worker's node id; tests supply a stub backed by httptest.

func NewMTLSDialer 

func NewMTLSDialer(mintClient func() (tls.Certificate, error), caPool *x509.CertPool) (AgentDialer, error)

NewMTLSDialer constructs the default control-plane-to-worker dialer. mintClient issues the control plane's short-lived client certificate; the dialer re-mints it past its half-life so it is never presented expired. It returns the AgentDialer interface because callers wire it into a remoteRuntime, which depends only on the interface.

type AgentServer 

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

AgentServer is the agent's inbound mTLS HTTPS listener. The control plane dials it to issue Start/Signal/Wait/Stats/RunOnce commands and to proxy data-plane traffic through /v1/data/{token}/*.

func NewAgentServer 

func NewAgentServer(cfg AgentServerConfig) *AgentServer

NewAgentServer constructs an AgentServer from cfg.

func (*AgentServer) Listen added in v0.6.2 

func (s *AgentServer) Listen() (net.Listener, error)

Listen binds an mTLS listener on the configured ListenAddr. It is the fail-fast half of serving: agent.Run binds synchronously so a port conflict surfaces before the worker announces liveness, then hands the listener to ServeListener to serve until ctx is cancelled.

func (*AgentServer) ServeListener added in v0.6.2 

func (s *AgentServer) ServeListener(ctx context.Context, ln net.Listener) error

ServeListener serves the replica-control API and data-plane proxy on ln until ctx is cancelled. ln must already terminate TLS (e.g. from tls.Listen with TLSConfig); callers that need a known bound port can pass their own listener.

func (*AgentServer) TLSConfig 

func (s *AgentServer) TLSConfig() *tls.Config

TLSConfig returns the tls.Config for the listener. The served cert and the client-CA pool are resolved per handshake through GetConfigForClient so a renewed cert or a rotated CA bundle takes effect on the next connection without restarting the listener. Exposed so tests can verify the security posture without binding a port.

type AgentServerConfig 

type AgentServerConfig struct {
	// ListenAddr is the host:port the server binds. The agent command passes
	// the --advertise-addr value (or a port offset from it).
	ListenAddr string
	// CertSource holds the issued keypair from SignWorkerCSR (ServerAuth +
	// ClientAuth EKUs, <nodeid>.node.shinyhub.internal SAN). Reading the cert
	// through the holder lets the agent swap a renewed cert in without restarting
	// the listener, so the worker's routing surface survives cert rotation.
	CertSource *CertHolder
	// CASource holds the CA pool the worker trusts; the control plane must present
	// a client cert signed by this CA to authenticate. Reading it through the
	// holder lets a CA bundle rotated on heartbeat take effect on the next
	// handshake without restarting the listener.
	CASource *CAHolder
	// NodeID is the assigned node id, embedded in the server cert SAN.
	NodeID string
	// Replicas is the replica-control server whose Routes are mounted. Set
	// after construction once the replicaServer is built in worker.go.
	Replicas *replicaServer
}

AgentServerConfig configures the agent's inbound mTLS HTTP server.

type BundleEnsurer 

type BundleEnsurer interface {
	Ensure(ctx context.Context, digest string) (string, error)
}

BundleEnsurer makes an app bundle present on the worker's local disk, keyed by content digest, and returns the extracted bundle directory. It is an interface rather than the concrete *agent.BundleCache because the agent package imports this one; depending on the concrete type would create an import cycle.

type CA 

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

CA is the control plane's internal certificate authority. It signs short-lived worker client certificates that bind a node id, and pins itself as the trust root both sides verify against. The keypair is generated on first OpenCA and persisted under dir.

func LoadOrInitCA added in v0.7.4 

func LoadOrInitCA(store CAStore, caDir, authSecret string, joinTokens []string) (*CA, error)

LoadOrInitCA loads/initializes the shared worker CA (see loadOrInitCA) and mirrors the public CA certificate to caDir/ca-cert.pem so worker bootstrap (`shinyhub worker --ca-file <ca_dir>/ca-cert.pem`) and operator tooling can read it. The private key is never written to disk; it stays encrypted in the database only.

func OpenCA 

func OpenCA(dir string, joinTokens []string) (*CA, error)

OpenCA loads a worker CA keypair from disk, generating and persisting it on first run. Production boot sources the CA from the database via LoadOrInitCA; OpenCA remains as the disk-based constructor used by tests (and disk-only setups).

func (*CA) CertPEM 

func (c *CA) CertPEM() []byte

CertPEM returns the PEM-encoded CA certificate (the bundle workers pin).

func (*CA) ControlClientCertificate 

func (c *CA) ControlClientCertificate() (tls.Certificate, error)

ControlClientCertificate issues a short-lived ECDSA client certificate that the control plane presents to worker agents over mTLS. The cert is signed by this CA, carries only the ClientAuth EKU, and verifies against CA.Pool().

func (*CA) ListenerTLSConfig added in v0.6.2 

func (c *CA) ListenerTLSConfig(hosts ...string) (*tls.Config, error)

ListenerTLSConfig builds the TLS config for the control plane's worker-facing listener. The server cert is minted off the CA for the given hosts and served through GetCertificate, which re-mints it past its half-life so a control plane running beyond the cert's lifetime keeps serving without a restart. Client certs are requested but not required (register runs before a worker has a cert) and verified against the CA when presented.

func (*CA) Pool 

func (c *CA) Pool() *x509.CertPool

Pool returns a verifier pool trusting this CA.

func (*CA) ServerCertificate 

func (c *CA) ServerCertificate(hosts ...string) (tls.Certificate, error)

ServerCertificate signs a fresh server keypair off the CA for the control plane's worker-facing listener. When no hosts are given it defaults to loopback (127.0.0.1, ::1, localhost). IP-literal hosts become IP SANs and the rest become DNS SANs. The returned certificate carries the leaf and the CA certificate so clients pinning the CA can build a full chain.

func (*CA) SignWorkerCSR 

func (c *CA) SignWorkerCSR(nodeID string, csrPEM []byte, ttl time.Duration) ([]byte, error)

SignWorkerCSR signs a worker CSR, binding nodeID into the certificate (CN plus a DNS SAN) so the node id is recoverable from the presented cert. The cert is valid for ttl (short-lived; renewed on heartbeat).

func (*CA) VerifyJoinToken 

func (c *CA) VerifyJoinToken(token string) bool

VerifyJoinToken reports whether token matches a configured join token, using a constant-time comparison to avoid leaking token length/contents via timing.

type CAHolder added in v0.6.2 

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

CAHolder holds the CA trust bundle a worker pins, behind a read-write lock so the inbound server and outbound client can read the current pool on every handshake while a heartbeat swaps in a rotated bundle without a restart. It is safe for concurrent use.

func NewCAHolder added in v0.6.2 

func NewCAHolder(caPEM []byte) (*CAHolder, error)

NewCAHolder parses caPEM into a trust pool, erroring if no certificate parses.

func (*CAHolder) Pool added in v0.6.2 

func (h *CAHolder) Pool() *x509.CertPool

Pool returns the current trust pool. Callers must treat it as read-only.

func (*CAHolder) Set added in v0.6.2 

func (h *CAHolder) Set(caPEM []byte) (bool, error)

Set replaces the bundle when caPEM differs from the current one, reporting whether a change was applied. An unparseable bundle is rejected and leaves the current pool intact, so a malformed rotation never strands the worker without trust roots.

type CAStore added in v0.7.4 

type CAStore interface {
	GetWorkerCA() (certPEM, keyEnc []byte, found bool, err error)
	PutWorkerCAIfAbsent(certPEM, keyEnc []byte) (inserted bool, err error)
}

CAStore is the minimal store the worker CA bootstrap needs. *db.Store satisfies it. found (not an ErrNotFound sentinel) signals an empty row.

type CertHolder added in v0.6.2 

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

CertHolder stores a TLS certificate that can be atomically swapped at runtime. A renewed certificate set via Set takes effect on the next TLS handshake of every listener and client wired to the holder through GetCertificate or GetClientCertificate, so a worker can rotate its expiring cert without restarting its inbound server or rebuilding its outbound client. The zero value is not usable; construct with NewCertHolder.

func NewCertHolder added in v0.6.2 

func NewCertHolder(cert tls.Certificate) *CertHolder

NewCertHolder returns a holder seeded with cert.

func (*CertHolder) Get added in v0.6.2 

func (h *CertHolder) Get() tls.Certificate

Get returns the currently held certificate.

func (*CertHolder) GetCertificate added in v0.6.2 

func (h *CertHolder) GetCertificate(*tls.ClientHelloInfo) (*tls.Certificate, error)

GetCertificate adapts the holder to tls.Config.GetCertificate so a server presents the current certificate on each handshake.

func (*CertHolder) GetClientCertificate added in v0.6.2 

func (h *CertHolder) GetClientCertificate(*tls.CertificateRequestInfo) (*tls.Certificate, error)

GetClientCertificate adapts the holder to tls.Config.GetClientCertificate so a client presents the current certificate on each handshake.

func (*CertHolder) Set added in v0.6.2 

func (h *CertHolder) Set(cert tls.Certificate)

Set replaces the held certificate. Subsequent handshakes present it.

type Client 

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

Client is a worker's mTLS HTTP client to its control plane. After bootstrap it carries the signed client cert and pins the CA; the same transport backs the data-plane tunnel the control-plane proxy dials in reverse (C1).

func NewClient 

func NewClient(serverURL string, certSource *CertHolder, caSource *CAHolder) (*Client, error)

NewClient builds an mTLS client that presents the certificate held by certSource and verifies the control plane against the trust bundle held by caSource. Sourcing both through holders lets the worker rotate its expiring cert and a rotated CA bundle without rebuilding the client.

func (*Client) FetchBundle 

func (c *Client) FetchBundle(ctx context.Context, digest string) (io.ReadCloser, error)

FetchBundle streams the bundle zip for a content digest. The caller verifies the digest on the returned stream.

func (*Client) Heartbeat 

func (c *Client) Heartbeat(ctx context.Context, version, renewCSRPEM string, incarnation int64) (workerapi.HeartbeatResponse, error)

Heartbeat posts a heartbeat and returns the control plane's response. When renewCSRPEM is non-empty it is sent as a certificate renewal request; the caller applies any cert the response carries. incarnation is the worker's currently-held generation counter, letting the control plane detect and fence a stale rejoin.

func (*Client) Transport 

func (c *Client) Transport() http.RoundTripper

Transport exposes the underlying transport for the data-plane tunnel (C1).

type RegisterParams 

type RegisterParams struct {
	Name          string
	AdvertiseAddr string
	Tier          string
	Version       string
	Fingerprint   string
}

RegisterParams is the data a joining worker supplies (node id is allocated by the registry, not chosen by the worker).

type Registry 

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

Registry is the control plane's view of joined workers: durable in the workers table, indexed in memory for routing. Multiple distinct-address workers may be up on one tier (real multi-worker capacity); the invariant is one up worker per (tier, advertise address), so a stale duplicate that rejoins at an endpoint a live worker already owns is superseded.

func NewRegistry 

func NewRegistry(store *db.Store) (*Registry, error)

NewRegistry constructs a registry and rebuilds its in-memory index from the workers table, so a control-plane restart re-adopts known workers.

func (*Registry) Forget added in v0.6.2 

func (r *Registry) Forget(nodeID string)

Forget drops a worker from the in-memory index. The monitor calls it after the store reaps a long-dead worker row so the fleet snapshot does not keep listing a node that no longer exists. Unknown node ids are a no-op. Forget never touches the store; the row is already gone.

It holds regMu so the delete is serialized against a concurrent Refresh, whose older ListWorkers snapshot could otherwise reinsert the just-reaped worker.

func (*Registry) Heartbeat added in v0.6.2 

func (r *Registry) Heartbeat(nodeID, fingerprint string, reported int64) (bool, int64, error)

Heartbeat records a worker's liveness and refreshes its trusted cert fingerprint (cert renewal) in both the store and the index. It keeps an up worker up, and promotes a not-up worker - a joining worker on its first heartbeat, or a down worker reaped for missed heartbeats or superseded while offline and now restarted under its old identity - to up only when no other up worker owns its (tier, advertise address). Gating the promotion on endpoint ownership keeps the one-up-worker-per-(tier,address) invariant: a worker the endpoint's live owner holds cannot resurrect itself alongside that owner. A joining worker becomes routable here, and the agent sends this first heartbeat only after its listener binds, so an up worker is always one that is listening.

reported is the incarnation the worker last learned from the control plane (at registration or a prior heartbeat). A reported incarnation behind the stored one means the control plane reaped this worker - bumping its incarnation and reassigning its replicas - while it was partitioned: the worker may still be running orphaned copies of those replicas, so the heartbeat is fenced (kept down, told to self-fence) instead of re-upped. reported == 0 is a legacy pre-fence agent and is never fenced, so a rolling upgrade does not fence agents that have not yet learned to report an incarnation. Heartbeat returns (fenced, current incarnation, error).

func (*Registry) MarkDown 

func (r *Registry) MarkDown(nodeID string) error

MarkDown transitions a worker to down in both the durable store and the in-memory routing index, so a worker whose heartbeat went stale is excluded from routing without waiting for a control-plane restart to rebuild the index. Unknown node ids are a no-op in memory; the store write still runs.

It holds regMu so the store write and the index update are one unit against a concurrent Refresh: without it, a Refresh whose ListWorkers snapshot predates this MarkDown could overwrite the index and resurrect the worker as "up".

func (*Registry) PlanPlacementForTier added in v0.7.0 

func (r *Registry) PlanPlacementForTier(tier, slug string, count int) []db.Worker

PlanPlacementForTier plans where to place count new replicas of slug across a tier's coexisting workers, spreading load deterministically. It returns one worker per replica, in assignment order; the returned slice is empty when no worker is up on the tier and may be shorter than count only in that case.

Each pick chooses the worker hosting the fewest running replicas; ties break toward the worker hosting fewer of slug's own replicas (anti-affinity, so an app's replicas spread for HA), then toward the lowest node id (deterministic). Picks within one call fold into a running tally before the next pick, so a batch deployed concurrently spreads across workers instead of stacking every replica on the lowest node id (which a per-replica read of the same pre-deploy snapshot would do). If the load query fails it plans from a zero baseline rather than refusing to place, so a transient store hiccup does not block deploys.

func (*Registry) Reap added in v0.9.5 

func (r *Registry) Reap(nodeID string) error

Reap marks a worker down AND bumps its incarnation, then refreshes the index from the store so the bumped value is visible to Heartbeat's fence check. The down-monitor calls this (not MarkDown) because a reaped worker also has its replicas stripped and reassigned elsewhere; the incarnation bump is what fences the worker if it later reconnects still running them.

func (*Registry) Refresh added in v0.8.0 

func (r *Registry) Refresh() error

Refresh rebuilds the in-memory routing index from the workers table. An instance becoming the control-plane owner calls it so its routing decisions reflect every worker row the previous owner wrote before it died (registrations, heartbeats, supersedes, reaps) - not just the rows present when this instance booted. It is a full replace: workers added since boot appear and workers removed from the store disappear. Idempotent.

It holds regMu (like Register/Heartbeat/Revoke), preserving the regMu-then-mu lock order, so the rebuild is a consistent snapshot against any in-flight registration on this instance.

func (*Registry) Register 

func (r *Registry) Register(p RegisterParams) (db.Worker, error)

Register allocates a node id, persists the worker, and indexes it. Re-register with a known node id is handled by the caller passing that id via Reregister.

func (*Registry) Revoke added in v0.6.2 

func (r *Registry) Revoke(nodeID string) error

Revoke administratively revokes a worker. It persists the revocation (which also marks the node down) and refreshes the in-memory index from the store so the node is excluded from routing immediately, without waiting for a control-plane restart. The row is kept (not deleted) so the revocation stays auditable. Returns db.ErrNotFound for an unknown node.

It holds regMu for the same reason Heartbeat and Register do: a heartbeat performs a read-decide-write of the worker's status, so revoking without that lock could land between a heartbeat's read and its write and let the heartbeat resurrect the revoked node to up. Serializing closes that window.

func (*Registry) Worker 

func (r *Registry) Worker(nodeID string) (db.Worker, bool)

Worker returns the indexed worker for a node id.

func (*Registry) WorkerForTier 

func (r *Registry) WorkerForTier(tier string) (db.Worker, bool)

WorkerForTier returns a single up worker routing a tier, if any: the first of WorkersForTier (lowest node id), so single-worker routing is deterministic.

func (*Registry) WorkersForTier added in v0.7.0 

func (r *Registry) WorkersForTier(tier string) []db.Worker

WorkersForTier returns every up worker on a tier, sorted by node id so the order is deterministic (map iteration order is not). The returned slice is a fresh copy safe for the caller to retain.

type ReplicaServerConfig 

type ReplicaServerConfig struct {
	Runtime      process.Runtime
	DataDir      string
	NodeID       string
	Advertise    string // host:port base used to build tunnel URLs
	AllocatePort func() int
	// Bundles, when set, is called on every remote replica start to pull and
	// verify the app bundle by content digest, returning the local extracted dir.
	Bundles BundleEnsurer
}

ReplicaServerConfig configures a replicaServer. AllocatePort is injectable for tests; production wraps deploy.AllocatePort (which returns int, no error).

type ReplicaTransportBuilder added in v0.8.1 

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

ReplicaTransportBuilder derives the HTTP transport for a replica row from the DB, without requiring the placement registry. Each instance is safe for concurrent use.

For remote_docker replicas it builds a worker mTLS transport from the DB worker row (fetching via GetWorker) and caches it by worker_id so the same *http.Transport is reused across calls - mTLS transports are instance- independent (shared CA pool + rotating client cert) and safe to reuse.

For all other providers (fargate, native, docker) it returns nil, directing the proxy to use the default HTTP transport. Fargate replicas reach their awsvpc private IP over plain HTTP inside the VPC; native and docker replicas run on localhost and require no special transport.

func NewReplicaTransportBuilder added in v0.8.1 

func NewReplicaTransportBuilder(dialer AgentDialer, store WorkerGetter) *ReplicaTransportBuilder

NewReplicaTransportBuilder constructs a builder that resolves transports from the DB rather than the placement registry, so every instance (including standbys whose registry is empty) can build the correct per-worker transport.

func (*ReplicaTransportBuilder) TransportForReplica added in v0.8.1 

func (b *ReplicaTransportBuilder) TransportForReplica(r *db.Replica) (http.RoundTripper, error)

TransportForReplica returns the HTTP transport to use when forwarding requests to the given replica. It returns nil for non-remote_docker providers (the proxy falls back to its tuned default backend transport in that case).

For remote_docker replicas the transport is built once per worker_id and cached; concurrent callers for the same worker_id are serialized through a mutex so the transport is built exactly once.

The cache is unbounded and never invalidated because the cached transport remains correct for the lifetime of a worker entry:

  • The client certificate rotates automatically via the dialer's GetClientCertificate callback, so a cached *http.Transport always presents a fresh cert on the next handshake without needing to be rebuilt.
  • The ServerName is derived from the worker's stable node ID, which never changes for a given worker row.
  • A revoked worker never enters the cache: dialer.Transport returns an error for a revoked worker (checked before insertion), and the routing layer removes revoked workers from replica assignments before this function is called for them.

type WorkerGetter added in v0.8.1 

type WorkerGetter interface {
	GetWorker(nodeID string) (*db.Worker, error)
}

WorkerGetter is the minimal store interface the transport builder requires to resolve a worker row by node ID. *db.Store satisfies it.

type WorkerLookup 

type WorkerLookup interface {
	PlanPlacementForTier(tier, slug string, count int) []db.Worker
	WorkersForTier(tier string) []db.Worker
	Worker(nodeID string) (db.Worker, bool)
}

WorkerLookup resolves workers for routing. Implemented by *Registry in production; a stub is used in tests. PlanPlacementForTier plans where to place a batch of new replicas of a slug (least-loaded, spread across the tier, one worker per replica); WorkersForTier enumerates every up worker on a tier (inventory spans all of them); Worker resolves the specific worker that owns an existing handle or a pre-planned target.

Source Files

View all Source files

Directories

Path Synopsis
internal/worker/agent/agent.go
 internal/worker/agent/agent.go
Package api defines the JSON request/response types and the NDJSON streaming framing that form the wire contract between the ShinyHub control plane and a worker agent.
 Package api defines the JSON request/response types and the NDJSON streaming framing that form the wire contract between the ShinyHub control plane and a worker agent.

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.