apiserver

README

controller

The control plane: the API every client and every node talks to, the loops that turn what was asked for into what should run where, and the gRPC session each node agent holds open.

Two faces on one process, and the difference matters. Clients get a Kubernetes-shaped REST API — discovery, OpenAPI, table printing, RBAC — so kubectl works against it without a plugin. Nodes get a bidirectional gRPC stream instead of polling: desired state goes down, status comes up, and the node's own certificate CN is what the server confines it to.

Everything between them is level-driven. A loop reads what is stored, computes what should be, writes the difference, and holds nothing in memory that it could not rebuild from the store — the same discipline the node agent follows, for the same reason: any of it can restart at any moment and must come back indistinguishable.

The storage is a PORT, not a choice made here: this module defines what it needs and cmd/controller wires an implementation in (bolt today). The SOURCE builds on any OS — an operator may run its tests on a mac — while the artifact node-tool ships to a host is built for the node's platform like everything else in bin/.

Layout

Package	Purpose
controller (apiserver)	the HTTP surface: routes, discovery, OpenAPI, tables, metrics, the pods alias
controller/service	the layer between HTTP and storage: validation, admission, watches, subresources
controller/admission	the admission chain — defaults, policy, id allocation, reference and ownership checks
controller/authn	who the caller is: client certificates, tokens, the identity in the context
controller/authz	what the caller may do: RBAC over the stored Roles and bindings
controller/nodeserver	the node side: the gRPC session, desired-state push, status and address leases
controller/loop	the level-driven control-loop substrate every loop below is built on
controller/loops/scheduler	picks a node for each Application that names none, through a plugin framework
controller/loops/appset	expands an ApplicationSet into its children and reconciles their rollout
controller/loops/nodecsr	approves and signs node certificate requests
controller/oidc	the workload-identity issuer: short-lived tokens a workload proves itself with
controller/clientset	typed ports so an in-tree loop talks to the API without going through HTTP
controller/e2e	end-to-end tests over the assembled server
controller/internal/memory	an in-memory storage implementation, for tests

The catalog projection lives in the HTTP surface rather than in a package of its own: it is a READ of the same three Kinds in Consul's shape, so that Traefik and anything else speaking Consul can consume the fleet unmodified.

Build and test

make controller                  # from the repository root -> bin/horchestra-controller

cd controller && go test ./... -race

make comments regenerates core/api/scheme/comments_gen.go from the API doc comments, and TestGoCommentsAreCurrent fails until it has been run — any edit to a doc comment on an API type needs it.

Documentation

Index

• Constants
• func AuditID(next bunrouter.HandlerFunc) bunrouter.HandlerFunc
• func Auth(a authn.Authenticator) bunrouter.MiddlewareFunc
• func Authz(az authz.Authorizer) bunrouter.MiddlewareFunc
• func Health(w http.ResponseWriter, _ *http.Request)
• func Recover(next bunrouter.HandlerFunc) bunrouter.HandlerFunc
• func RequestLog(next bunrouter.HandlerFunc) bunrouter.HandlerFunc
• type APIServer
  • func New(sch *scheme.Scheme, svc Service, mws ...bunrouter.MiddlewareFunc) *APIServer
  • func (s *APIServer) EmulatePodsAPI()
  • func (s *APIServer) EnableNodeLogs()
  • func (s *APIServer) EnableSessionLog(local LogStreamer)
  • func (s *APIServer) ServeHTTP(w http.ResponseWriter, req *http.Request)
  • func (s *APIServer) SetAuthenticator(a authn.Authenticator)
  • func (s *APIServer) SetAuthorizer(a authz.Authorizer)
  • func (s *APIServer) SetCatalogNamespace(ns string)
  • func (s *APIServer) SetLogStreamer(ls LogStreamer)
  • func (s *APIServer) SetMetricsSource(m MetricsSource)
  • func (s *APIServer) SetNamespaceFilter(f NamespaceFilter)
  • func (s *APIServer) SetRateSource(r RateSource)
  • func (s *APIServer) SetRoutedNetwork(on bool)
• type ApplicationMetrics
• type ContainerMetrics
• type LogStreamer
• type MetricsSource
• type NamespaceFilter
• type NodeMetrics
• type NodeMetricsList
• type PodMetrics
• type PodMetricsList
• type Rate
• type RateSource
• type Sample
• type Service

Constants

const Datacenter = "horchestra"

Datacenter is the single datacenter this control plane reports. Consul's model has one per cluster and clients ask for it by name; a fleet is one datacenter until there is a second control plane to federate with, and inventing more names than there are things would only give a client somewhere wrong to point.

const DefaultCatalogNamespace = "default"

DefaultCatalogNamespace is the namespace an unqualified catalog query answers when the operator names none — Consul Enterprise's own answer, and the name a fleet's first namespace has here.

Functions

func AuditID

func AuditID(next bunrouter.HandlerFunc) bunrouter.HandlerFunc

func Auth

func Auth(a authn.Authenticator) bunrouter.MiddlewareFunc

func Authz

func Authz(az authz.Authorizer) bunrouter.MiddlewareFunc

func Health

func Health(w http.ResponseWriter, _ *http.Request)

Health answers the liveness and readiness probes.

It is mounted OUTSIDE the authenticated router, beside the composition root's other unauthenticated paths, and that is not a convenience: a kubelet probing a Pod carries no credentials, so a health endpoint behind authentication answers 401 and the process is declared dead while it is serving perfectly. The same is true of the aggregator's availability check.

One answer for both probes on purpose: this server holds its storage open for its whole life and has no phase in which it is alive but not ready, so a readiness check that consulted something else would be inventing a distinction the process does not have.

func Recover

func Recover(next bunrouter.HandlerFunc) bunrouter.HandlerFunc

Recover turns a panic in any handler or middleware below it into a 500 for that one request, logged with its stack. Without it net/http's per-connection recovery aborts the connection with no audit line and no stack, so a reachable panic degrades into an unattributable stream of dropped requests. Place it outermost, before AuditID, so a panic in the middleware chain itself is caught too.

func RequestLog

func RequestLog(next bunrouter.HandlerFunc) bunrouter.HandlerFunc

RequestLog records every mutating request (create/update/patch/delete) with its verb, path, caller identity and audit id — so a write, and especially a DELETE, is always attributable — plus the sensitive `pods/<app>/log` read, so streaming another workload's logs leaves a trail. Other reads (plain GET/watch) are not logged, to keep the audit trail focused and low-noise. Place it after Auth (so the identity is set) and before Authz (so a denied request is logged too).

Types

type APIServer

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

func New

func New(sch *scheme.Scheme, svc Service, mws ...bunrouter.MiddlewareFunc) *APIServer

New builds an APIServer that serves the Kinds registered in sch, backed by svc for all resource operations. Optional middleware (authn/authz, logging) wraps every route. It registers the typed /apis routes and the legacy /api discovery; mount the returned server with http.Handle or ServeHTTP.

func (*APIServer) EmulatePodsAPI

func (s *APIServer) EmulatePodsAPI()

func (*APIServer) EnableNodeLogs

func (s *APIServer) EnableNodeLogs()

EnableNodeLogs registers the node-log route. It is called only when the NodeLogs feature gate is on, and that is the whole enforcement: with the gate off the route DOES NOT EXIST, so the answer is the router's ordinary 404 for an unknown path — indistinguishable from a typo, with no handler behind a permission check to get past and nothing about the fleet to probe by asking.

The path is a real /apis path, which is what makes the authorization ordinary rather than a special case: the middleware classifies it as a `get` on the `log` subresource of the cluster-scoped `nodes` resource, so RBAC decides it before the handler runs. Nobody holding namespace-scoped rights comes near it — which is the reason this is NOT a fallback inside pods/log, where a namespaced path would have led to a cluster-scoped object.

It is read with `kubectl get --raw /apis/horchestra.io/v1/nodes/<name>/log`, which streams the body, so follow works with any kubectl and no client-side support at all.

func (*APIServer) EnableSessionLog

func (s *APIServer) EnableSessionLog(local LogStreamer)

EnableSessionLog registers the route one replica of this control plane uses to reach another.

A node holds ONE session, to ONE replica, and both nodes and operators arrive through a balancer — so `kubectl logs` lands on the replica holding that node's session only by luck, and with three replicas it usually does not. The replica that received the request authorizes the caller for the object as it always has, and then forwards the stream here.

It takes the LOCAL streamer explicitly, and that is the whole non-recursion argument: whatever forwards must never be what serves this route, or a stale record would let two replicas bounce a request between them. This end serves what it holds or says it holds nothing.

Authorized as `nodes/sessionlog`, which only the built-in controller grant confers — no RBAC object names it, and no operator credential carries it.

func (*APIServer) ServeHTTP

func (s *APIServer) ServeHTTP(w http.ResponseWriter, req *http.Request)

func (*APIServer) SetAuthenticator

func (s *APIServer) SetAuthenticator(a authn.Authenticator)

SetAuthenticator wires the identity check the not-found handler makes on its own behalf. Pass the same Authenticator the Auth middleware holds: the two must not be able to disagree about who a caller is, and a path that no route serves is the one place the middleware's answer never arrives.

func (*APIServer) SetAuthorizer

func (s *APIServer) SetAuthorizer(a authz.Authorizer)

SetAuthorizer wires the engine that authorizes the pods alias. The alias is a legacy /api/v1 route, which AttributesFromRequest classifies as a non-resource request (it parses only /apis paths) and the Casbin engine then allows unconditionally — so the Authz middleware decides nothing here and the check has to happen in the handler, where the Application's namespace is finally known. Left unset (e.g. under auth compiled out), every namespace is allowed.

func (*APIServer) SetCatalogNamespace

func (s *APIServer) SetCatalogNamespace(ns string)

SetCatalogNamespace names the namespace an unqualified catalog query answers.

It is configurable because `default` is Consul's convention and not necessarily a fleet's: a deployment that puts its workloads in `platform` and never uses `default` would otherwise answer every client that cannot send `?ns=` with an empty catalog — and an empty catalog reads as "the service is gone", not as "you asked the wrong question". The alternative, flattening every namespace into an unqualified answer, is the one thing this must not do: one tenant's `api` would shadow another's for a client with no way to ask better.

func (*APIServer) SetLogStreamer

func (s *APIServer) SetLogStreamer(ls LogStreamer)

SetLogStreamer wires the backend that `pods/<app>/log` streams through (the controller↔agent gRPC transport satisfies it). Without it the log endpoint reports unavailable. Call before serving.

func (*APIServer) SetMetricsSource

func (s *APIServer) SetMetricsSource(m MetricsSource)

SetMetricsSource wires the backend that `applications/<name>/metrics` and the Prometheus exporter read. Without it both report that no measurement is available, rather than pretending a workload used nothing.

func (*APIServer) SetNamespaceFilter

func (s *APIServer) SetNamespaceFilter(f NamespaceFilter)

SetNamespaceFilter wires the per-caller filter for the Namespace collection endpoint, so a user lists the namespaces they can access without cluster-wide rights. Left unset (a build with auth compiled out), the endpoint returns every namespace.

func (*APIServer) SetRateSource

func (s *APIServer) SetRateSource(r RateSource)

SetRateSource wires what `kubectl top` reads.

func (*APIServer) SetRoutedNetwork

func (s *APIServer) SetRoutedNetwork(on bool)

SetRoutedNetwork tells the server whether workloads get networks of their own on this fleet, which decides whether `kubectl get application -o wide` carries an IP column at all. See columnsFor.

type ApplicationMetrics

type ApplicationMetrics struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata"`

	// Node is where the sample was taken; Timestamp is when, by the node's clock.
	Node      string      `json:"node,omitempty"`
	Timestamp metav1.Time `json:"timestamp,omitzero"`
	// Usage is cumulative since the workload started, so two samples give a rate over
	// whatever window the reader cares about — and one sample gives none, on purpose.
	CPUMicroseconds          int64 `json:"cpuMicroseconds"`
	CPUThrottledMicroseconds int64 `json:"cpuThrottledMicroseconds"`
	MemoryBytes              int64 `json:"memoryBytes"`
	MemoryPeakBytes          int64 `json:"memoryPeakBytes"`
	PIDs                     int64 `json:"pids"`
	OOMKills                 int64 `json:"oomKills"`
}

ApplicationMetrics is what a single workload consumed, served at the application's own metrics subresource. The shape is deliberately not metrics.k8s.io: that API exists so `kubectl top` can find it, and faking a whole foreign group to borrow one command is a worse trade than a subresource of the object this control plane actually has — which, now that subresources authorize as themselves, is a permission an operator can grant on its own (`applications/metrics`, verb get) without handing over the Applications too.

type ContainerMetrics

type ContainerMetrics struct {
	Name  string                       `json:"name"`
	Usage map[string]resource.Quantity `json:"usage"`
}

ContainerMetrics names the container the usage belongs to. An Application is one process, so there is exactly one entry and it takes the application's own name — the same identity the pods alias presents.

type LogStreamer

type LogStreamer interface {
	StreamLogs(ctx context.Context, node, app string, follow bool, tail int64) (<-chan []byte, func() error, error)
	// StreamNodeLogs streams a node agent's own unit journal. Reachable only when the NodeLogs
	// gate is on, and then only through a route that authorizes `nodes/log`.
	StreamNodeLogs(ctx context.Context, node string, follow bool, tail int64) (<-chan []byte, func() error, error)
}

LogStreamer streams an application's logs from the node it runs on (the controller<->agent gRPC transport satisfies it). Absent (nil), the log endpoint reports it is unavailable.

type MetricsSource

type MetricsSource interface {
	Metrics(namespace, name string) (Sample, bool)
	AllMetrics() []Sample
	// AllNodeMetrics is what the MACHINES are consuming, which is not the sum of their
	// workloads: the system, the agent and everything else on the host are in it, and a
	// capacity is held against that total rather than against its tenants.
	AllNodeMetrics() []Sample
}

MetricsSource is the last measured consumption per workload, held by whatever received it — the node transport does. The apiserver only serves what it is given; it does no collection and keeps no history of its own.

type NamespaceFilter

type NamespaceFilter func(ctx context.Context, id *authn.Identity) (accessible sets.Set[string], seesAll bool, err error)

NamespaceFilter reports the namespaces a caller may see in the self-service Namespace listing: accessible is the set of namespace names (used when seesAll is false), and seesAll is true for an admin (every namespace). It must not require any cluster-wide list permission — that is the whole point of self-service listing.

type NodeMetrics

type NodeMetrics struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata"`

	Timestamp metav1.Time                  `json:"timestamp"`
	Window    metav1.Duration              `json:"window"`
	Usage     map[string]resource.Quantity `json:"usage"`
}

type NodeMetricsList

type NodeMetricsList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata"`
	Items           []NodeMetrics `json:"items"`
}

type PodMetrics

type PodMetrics struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata"`

	Timestamp metav1.Time        `json:"timestamp"`
	Window    metav1.Duration    `json:"window"`
	Container []ContainerMetrics `json:"containers"`
}

PodMetrics and NodeMetrics are the metrics.k8s.io shapes, spelled out here rather than pulled from k8s.io/metrics: two structs are cheaper than a dependency whose only other content is a client this server does not use.

type PodMetricsList

type PodMetricsList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata"`
	Items           []PodMetrics `json:"items"`
}

type Rate

type Rate struct {
	MilliCores  int64
	MemoryBytes int64
	Window      time.Duration
	At          time.Time
}

Rate mirrors the node transport's derived rate; see Sample for why the shape is duplicated rather than imported.

type RateSource

type RateSource interface {
	Rate(namespace, name string) (Rate, bool)
	AllRates() map[string]Rate
	NodeRate(node string) (Rate, bool)
	AllNodeRates() map[string]Rate
}

RateSource is consumption per unit time, which is what `kubectl top` displays — cores, not cumulative microseconds. A single counter cannot answer it, so the source only has one once two samples exist.

type Sample

type Sample struct {
	Namespace        string
	Name             string
	Node             string
	CPUUsec          int64
	CPUThrottledUsec int64
	MemoryBytes      int64
	MemoryPeakBytes  int64
	PIDs             int64
	OOMKills         int64
	At               time.Time
	Received         time.Time
}

Sample mirrors the node transport's measurement so this package does not import it — the dependency runs the other way, and a shared struct is cheaper than an interface per field.

type Service

type Service interface {
	Get(ctx context.Context, m types.ObjectMeta) (types.Object, error)
	List(ctx context.Context, m types.ObjectMeta, opts metav1.ListOptions) ([]types.Object, error)
	// ListAt is List plus the cursor it was taken at, for the caller that will resume a watch
	// there — the HTTP list, which has to publish it, and the initial-events stream.
	ListAt(ctx context.Context, m types.ObjectMeta, opts metav1.ListOptions) ([]types.Object, string, error)
	Watch(ctx context.Context, m types.ObjectMeta, opts metav1.ListOptions) (<-chan metav1.WatchEvent, error)
	Create(ctx context.Context, gvk schema.GroupVersionKind, data []byte, ns string) (types.Object, error)
	Update(ctx context.Context, gvk schema.GroupVersionKind, data []byte, ns, name string) (types.Object, error)
	UpdateSubresource(ctx context.Context, gvk schema.GroupVersionKind, subresource string, data []byte, ns string) (types.Object, error)
	// UpdateThrough writes the whole object, telling admission which subresource door the write
	// came through — the freeze/thaw actions, which change the spec and must not be writable BY
	// the spec.
	UpdateThrough(ctx context.Context, gvk schema.GroupVersionKind, door string, data []byte, ns, name string) (types.Object, error)
	Patch(ctx context.Context, m types.ObjectMeta, pt k8stypes.PatchType, data []byte) (types.Object, error)
	Delete(ctx context.Context, m types.ObjectMeta, opts metav1.DeleteOptions) error
	Rollback(ctx context.Context, m types.ObjectMeta, uid string, targetRV int64) (types.Object, error)
}