Documentation
¶
Overview ¶
Package dataplane implements the per-node Rune data path (RUNE-041): per-VIP TCP/UDP userspace proxies, a kernel-side nftables reconciler (Linux only), an endpoint cache backed by the OrderedLog watch stream, and Prometheus metrics. The dataplane registers as an Agent Subsystem (internal/agent.Subsystem) so its lifecycle is governed by the agent.
On startup the subsystem subscribes to the OrderedLog from seq 0, hydrates its endpoint cache, then transitions into live-watch mode. On watch disconnect it serves the last-known endpoint set for at most StaleBudget (default 30s) before failing closed (refusing new connections; existing connections continue until torn down).
Listener lifecycle is driven by service registration. The agent's main wires a ServiceProvider that knows each service's VIP and port list. When a service is registered the subsystem opens TCP and UDP listeners on the VIP (or 127.0.0.1 in dev-mode) for each port; when it's unregistered listeners drain (TCP) or close (UDP) within DrainTimeout (default 30s). Dev-mode is signalled via Agent.Mode().
Index ¶
- Constants
- func PortReserved(reserved []int, port int) bool
- type Cache
- type Config
- type FuncVIPResolver
- type ListenerSummary
- type Metrics
- type Mode
- type NodeIDProvider
- type ProxyManager
- type ServiceProvider
- type ServiceSummary
- type Subsystem
- func (s *Subsystem) Cache() *Cache
- func (s *Subsystem) LocalInstances() *policy.LocalInstancesTable
- func (s *Subsystem) Metrics() *Metrics
- func (s *Subsystem) Name() string
- func (s *Subsystem) PolicyFor(serviceID string) *policy.Compiled
- func (s *Subsystem) Ready() <-chan struct{}
- func (s *Subsystem) RegisterService(svc *types.Service) error
- func (s *Subsystem) Start(ctx context.Context) error
- func (s *Subsystem) Stop(ctx context.Context) error
- func (s *Subsystem) UnregisterService(serviceID string)
- type VIPResolver
Constants ¶
const ( DefaultStaleBudget = 30 * time.Second DefaultDrainTimeout = 30 * time.Second DefaultReconnectMinDelay = 200 * time.Millisecond DefaultReconnectMaxDelay = 5 * time.Second )
Defaults can be overridden via Config.
const ( LocalityNone = "" LocalityPreferLocal = "prefer-local" LocalityLocalOnly = "local-only" )
LocalityPreference values mirror Service.Discovery.LocalityPreference.
Variables ¶
This section is empty.
Functions ¶
func PortReserved ¶
PortReserved reports whether port is in the dataplane's reserved set (ingress-owned ports on an edge node, typically 80/443). Exported so the ingress controller shares one definition of "reserved port" rather than carrying its own copy.
Types ¶
type Cache ¶
type Cache struct {
// contains filtered or unexported fields
}
Cache is the dataplane's in-memory copy of every service's endpoint set, keyed by service ID. It is updated by the watch loop in Subsystem and read by the proxy on every connection accept.
All methods are safe for concurrent use. Reads return defensive copies so callers can iterate without holding the lock.
func (*Cache) Get ¶
Get returns the endpoint slice for serviceID. ok is false if the service has never been seen by this cache. A nil slice with ok=true means "service known, no endpoints right now".
func (*Cache) Set ¶
Set replaces the endpoint slice for serviceID. A nil/empty slice is kept as an explicit "no endpoints" marker — the proxy fails closed instead of falling back to historic data.
func (*Cache) Snapshot ¶
func (c *Cache) Snapshot() []ServiceSummary
Snapshot returns a sorted list of (serviceID, count) pairs for diagnostics. Allocation cost is acceptable for the admin surface.
type Config ¶
type Config struct {
// OrderedLog is the source of truth for endpoint mutations.
// Required.
OrderedLog orderedlog.OrderedLog
// Services lets the dataplane look up VIP + ports for a service
// referenced by an endpoints update. Optional; service registration
// is driven by Store watch when set.
Services ServiceProvider
// Store watches service rows and opens per-VIP listeners. Required
// for production cluster networking on single-node runed.
Store store.Store
// VIPResolver supplies cluster VIPs when a service row is missing
// discovery.vip (legacy drift). Typically the VIP allocator.
VIPResolver VIPResolver
// Node identifies this node for localityPreference. Required for
// "prefer-local" / "local-only" semantics; in its absence those
// modes degrade to "no preference".
Node NodeIDProvider
// Mode is "production" (default) or "dev". Dev disables nftables
// and listens on 127.0.0.1 instead of the real VIP.
Mode Mode
// ReservedHostPorts are host ports the dataplane must NOT open VIP
// listeners on. On an edge node the ingress owns :80/:443 with a
// 0.0.0.0 wildcard bind; a <vip>:80 listener collides with it and
// fails the whole ingress subsystem. Services exposed on these
// ports are reached via the ingress, not the VIP.
ReservedHostPorts []int
// Logger; defaults to the global logger with component "dataplane".
Logger log.Logger
// StaleBudget is the time a watch disconnect can persist before
// the proxy starts refusing new connections. Defaults to 30s.
StaleBudget time.Duration
// DrainTimeout is the connection-drain deadline applied when an
// endpoint is removed or a listener is being closed. Defaults to 30s.
DrainTimeout time.Duration
}
Config bundles dataplane construction parameters.
type FuncVIPResolver ¶
FuncVIPResolver adapts a lookup function to VIPResolver.
func (FuncVIPResolver) VIPForService ¶
type ListenerSummary ¶
type ListenerSummary struct {
ServiceID string
Port int
// TargetPort is the upstream container port this listener dials.
// Defaults to Port when the service spec leaves TargetPort unset.
TargetPort int
Protocol string
Addr string
Active int64
}
ListenerSummary is the diagnostic view of a single listener.
type Metrics ¶
type Metrics struct {
ConnectionsActive *prometheus.GaugeVec // {service,protocol}
ConnectionsTotal *prometheus.CounterVec // {service,protocol,result}
EndpointHealth *prometheus.GaugeVec // {service} = healthy count
WatchLag prometheus.Gauge // seconds since last watch event
NftablesRules prometheus.Gauge // current rule count (linux only)
ListenersOpen *prometheus.GaugeVec // {service,protocol}
PolicyDrops *prometheus.CounterVec // {service,namespace,policy,reason}
PolicyAllows *prometheus.CounterVec // {service,namespace,policy}
PolicyRules *prometheus.GaugeVec // {service,namespace} -> ingress+egress rule count
PolicyLastSeq *prometheus.GaugeVec // {service,namespace} -> last refresh timestamp (unix seconds)
LocalInstances prometheus.Gauge // size of LocalInstances table
}
Metrics holds the dataplane's Prometheus collectors. The set is allocated per Subsystem so multiple subsystems in tests don't fight over the global registry; production wires the set into the runed-level registry via Register.
func (*Metrics) Register ¶
func (m *Metrics) Register(reg prometheus.Registerer) error
Register attaches every collector to reg. Idempotent failures (AlreadyRegistered) are tolerated so the agent can re-register after a controlled restart.
type Mode ¶
type Mode int
Mode mirrors agent.Mode without importing it (avoids import cycle if the agent ever depends on the dataplane). Production is the kernel-DNAT-and-real-VIP path; Dev is the loopback-listener path used by laptop development on macOS.
type NodeIDProvider ¶
type NodeIDProvider interface {
NodeID() string
}
NodeIDProvider returns the local node's stable identity. Used for localityPreference scoring (prefer-local, local-only).
func StaticNodeID ¶
func StaticNodeID(id string) NodeIDProvider
StaticNodeID wraps a string node ID into a NodeIDProvider.
type ProxyManager ¶
type ProxyManager struct {
// contains filtered or unexported fields
}
ProxyManager owns the set of per-VIP listeners and supervises their lifecycle. It is the dataplane's only network-facing component.
func (*ProxyManager) Register ¶
func (pm *ProxyManager) Register(svc *types.Service) error
Register reconciles listeners for svc. Adds new listeners for new ports, removes listeners for removed ports, leaves unchanged ones alone. Returns the first failure if any port couldn't bind.
func (*ProxyManager) Shutdown ¶
func (pm *ProxyManager) Shutdown(ctx context.Context)
Shutdown closes all listeners, draining for at most ctx's deadline.
func (*ProxyManager) Snapshot ¶
func (pm *ProxyManager) Snapshot() []ListenerSummary
Snapshot returns one entry per active listener for diagnostics.
func (*ProxyManager) Unregister ¶
func (pm *ProxyManager) Unregister(serviceID string)
Unregister tears down all listeners for serviceID.
type ServiceProvider ¶
ServiceProvider is the dataplane's read-only view of the service catalog. The dataplane needs each registered service's VIP + port list to open listeners; it doesn't care about everything else on the Service spec. The orchestrator wires the real provider in a follow-up; for now main.go uses an in-memory provider populated by the public Register / Unregister calls below.
type ServiceSummary ¶
ServiceSummary is the public diagnostic view of a single entry.
type Subsystem ¶
type Subsystem struct {
// contains filtered or unexported fields
}
Subsystem is the per-node dataplane. It implements the Subsystem interface used by internal/agent (Name / Start / Ready / Stop) but does not import the agent package directly to avoid cycles.
func New ¶
New constructs a Subsystem. It registers the endpoints op types against cfg.OrderedLog (idempotent), but does not start any goroutines until Start is called.
func (*Subsystem) Cache ¶
Cache returns the live endpoint cache. Exposed for diagnostics + tests; callers should not mutate the returned snapshot.
func (*Subsystem) LocalInstances ¶
func (s *Subsystem) LocalInstances() *policy.LocalInstancesTable
LocalInstances returns the agent's IP -> identity table for diagnostics and tests.
func (*Subsystem) Metrics ¶
Metrics returns the dataplane's Prometheus collector set so the agent's HTTP server can register them.
func (*Subsystem) Ready ¶
func (s *Subsystem) Ready() <-chan struct{}
Ready returns a channel closed once the subsystem has hydrated its cache from the OrderedLog and is serving live updates.
func (*Subsystem) RegisterService ¶
RegisterService opens listeners for svc on this node. Idempotent per service ID; mutating fields (port list, VIP) cause the listener set to be reconciled. Safe to call before Start.
func (*Subsystem) Start ¶
Start begins the watch loop and reconciler. Returns once Ready becomes ready or the context is cancelled.
func (*Subsystem) Stop ¶
Stop drains all listeners and shuts down the watch loop. Honors ctx for an upper bound on drain time.
func (*Subsystem) UnregisterService ¶
UnregisterService closes listeners for serviceID. Connections drain for up to DrainTimeout. Idempotent.