server

package
v0.0.0-...-c46dd35 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

Package server implements the Registrar gRPC service, including endpoint snapshot management, change broadcasting, and external registry synchronization.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Broadcaster 

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

Broadcaster fans out endpoint events to connected agent watch streams. Each watcher is identified by a string key (typically cluster/node) and receives events on a buffered channel. Watchers carrying a service filter are indexed by service, so an endpoint change fans out to that service's consumers only (demand-scoped distribution) instead of every node. A watcher too slow to keep up is forcibly resynced: its channel is closed, ending its WatchEndpoints stream, and the agent's reconnect receives a fresh (filtered) snapshot — it must never silently miss an event and serve stale endpoints until something else triggers a resync.

func NewBroadcaster 

func NewBroadcaster(log *slog.Logger, metrics *Metrics) *Broadcaster

NewBroadcaster creates a Broadcaster. metrics may be nil to disable instrumentation.

func (*Broadcaster) Broadcast 

func (b *Broadcaster) Broadcast(events []*registrarv1.WatchEndpointsResponse)

Broadcast sends each event to the watchers subscribed to its service (the service's consumers plus full watchers). Events are sent non-blocking; a watcher whose channel is full has already missed an event, so it is forced to resync: its channel is closed, which ends its WatchEndpoints stream, and the agent reconnects to receive a fresh filtered snapshot. The alternative — silently dropping the event — leaves the agent serving stale endpoints with nothing to correct it until its next reconnect for unrelated reasons.

func (*Broadcaster) Subscribe 

func (b *Broadcaster) Subscribe(id string, services []string) <-chan *registrarv1.WatchEndpointsResponse

Subscribe registers a new watcher scoped to the given services (nil = full watch; empty = watch nothing) and returns a channel that will receive endpoint events. The caller must call Unsubscribe when done.

func (*Broadcaster) Unsubscribe 

func (b *Broadcaster) Unsubscribe(id string, ch <-chan *registrarv1.WatchEndpointsResponse)

Unsubscribe removes a watcher and closes its channel. The channel returned by the matching Subscribe call must be passed so that a stale caller (whose subscription was already replaced by a reconnect with the same id) does not close or delete the newer subscription's channel.

func (*Broadcaster) WatcherCount 

func (b *Broadcaster) WatcherCount() int

WatcherCount returns the number of currently subscribed watchers.

type Metrics 

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

Metrics holds the registrar server's OTel instruments. All methods are nil-receiver-safe so the server runs unchanged when telemetry is disabled (pass a nil *Metrics).

These instruments exist to make missed updates and stale state observable: a nonzero dropped-events counter means at least one agent was force-resynced (or, before PR4, silently diverged), and the snapshot-version gauge is the registrar half of the registrar-vs-agent version-skew query.

func NewMetrics 

func NewMetrics(meter metric.Meter) (*Metrics, error)

NewMetrics registers the registrar server instruments on the given meter.

type RegistrarServer 

type RegistrarServer struct {
	registrarv1.UnimplementedRegistrarServiceServer

	// Embed xds.Server for gRPC lifecycle management.
	xds.Server
	// contains filtered or unexported fields
}

RegistrarServer implements the RegistrarServiceServer gRPC interface. It delegates writes to an external registry and serves reads from a local snapshot. Change notifications are pushed to agents via the broadcaster.

func NewRegistrarServer 

func NewRegistrarServer(
	reg registry.Registry,
	snapshot *Snapshot,
	broadcaster *Broadcaster,
	address string,
	log *slog.Logger,
	metrics *Metrics,
	grpcOpts ...grpc.ServerOption,
) *RegistrarServer

NewRegistrarServer creates a RegistrarServer and registers it on the gRPC server. Additional gRPC server options (e.g., TLS credentials) can be passed via grpcOpts. metrics may be nil to disable instrumentation.

func (*RegistrarServer) GateOnSync 

func (s *RegistrarServer) GateOnSync(ch <-chan struct{})

GateOnSync makes snapshot-serving RPCs (WatchEndpoints, ListAllEndpoints) block until ch is closed (the syncer's first completed cycle). Nil leaves the RPCs ungated.

func (*RegistrarServer) ListAllConfig 

func (s *RegistrarServer) ListAllConfig(ctx context.Context, _ *registrarv1.ListAllConfigRequest) (*registrarv1.ListAllConfigResponse, error)

ListAllConfig returns the clusterset-wide config projections (proposal 026, multi-cluster config propagation). Agents pull this from the spoke registrar rather than reading the registry directly (the standing directive: agents don't talk to the store). When the backend has no cross-cluster config plane (kubernetes/dynamodb), the registry does not implement ConfigExporter and this returns an empty set — config stays cluster-local, which is the correct degenerate behaviour.

func (*RegistrarServer) ListAllEndpoints 

func (s *RegistrarServer) ListAllEndpoints(ctx context.Context, req *registrarv1.ListAllEndpointsRequest) (*registrarv1.ListAllEndpointsResponse, error)

ListAllEndpoints returns all endpoints from the local snapshot.

func (*RegistrarServer) RegisterEndpoint 

func (s *RegistrarServer) RegisterEndpoint(ctx context.Context, req *registrarv1.RegisterEndpointRequest) (*registrarv1.RegisterEndpointResponse, error)

RegisterEndpoint applies the change to the snapshot and broadcasts it immediately (discovery moves at watch latency), then writes the external registry — through the write-behind queue when enabled, so a failing external write can never make a serving pod invisible to the mesh (rev-68 roll regression). Without a queue (tests/legacy) the write is synchronous and failures fail the RPC.

func (*RegistrarServer) UnregisterEndpoint 

func (s *RegistrarServer) UnregisterEndpoint(ctx context.Context, req *registrarv1.UnregisterEndpointRequest) (*registrarv1.UnregisterEndpointResponse, error)

UnregisterEndpoint applies the removal to the snapshot and broadcasts it immediately, then writes the external registry (write-behind when enabled; see RegisterEndpoint).

func (*RegistrarServer) UseWriteBehind 

func (s *RegistrarServer) UseWriteBehind(q *WriteBehindQueue)

UseWriteBehind enables snapshot-first registry mutations through q.

func (*RegistrarServer) WatchEndpoints 

func (s *RegistrarServer) WatchEndpoints(req *registrarv1.WatchEndpointsRequest, stream grpc.ServerStreamingServer[registrarv1.WatchEndpointsResponse]) error

WatchEndpoints streams endpoint events to the agent. It first sends a full snapshot (unless the client's last_version matches the current version), then forwards incremental events from the broadcaster. A request filter scopes both the snapshot and the incremental events to the named services (demand-scoped distribution): the agent re-asserts its filter on every reconnect, and an unset filter preserves the full watch.

type Snapshot 

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

Snapshot is a thread-safe, versioned in-memory store of all service endpoints. It supports computing diffs between states and applying incremental changes.

func NewSnapshot 

func NewSnapshot() *Snapshot

NewSnapshot creates an empty Snapshot starting at version 0.

func (*Snapshot) Apply 

func (s *Snapshot) Apply(events []*registrarv1.WatchEndpointsResponse) (string, []*registrarv1.WatchEndpointsResponse)

Apply applies a set of events to the snapshot, updating it in place. It returns the new version string plus the service-catalog transitions the events caused (a service's endpoint count crossing 0<->1 emits SERVICE_ADDED/SERVICE_REMOVED): deriving transitions inside Apply makes the catalog impossible to desync from the endpoint data it summarizes. Transitions are unversioned; the caller stamps and broadcasts them with the batch.

func (*Snapshot) Diff 

func (s *Snapshot) Diff(newEndpoints map[string]map[registryv1.Service_Protocol][]*registryv1.ServiceEndpoint) []*registrarv1.WatchEndpointsResponse

Diff compares a new set of endpoints against the current snapshot and returns the events needed to transition from the current state to the new state. It does not modify the snapshot.

func (*Snapshot) FullSnapshotEvents 

func (s *Snapshot) FullSnapshotEvents(filter map[string]struct{}) ([]*registrarv1.WatchEndpointsResponse, string)

FullSnapshotEvents returns the current contents of the snapshot as a slice of FULL_SNAPSHOT events. This is used to send the initial state to a new watcher. A non-nil filter scopes the result to those service names (a nil filter is the unfiltered, cluster-wide snapshot): a demand-scoped watcher discards everything outside its filter anyway, so the out-of-scope events are skipped before their protos are ever built.

func (*Snapshot) GetAll 

func (s *Snapshot) GetAll(protocol registryv1.Service_Protocol) map[string][]*registryv1.ServiceEndpoint

GetAll returns all endpoints organized by service name. The caller receives a copy that is safe to mutate.

func (*Snapshot) GetAllWithVersion 

func (s *Snapshot) GetAllWithVersion(protocol registryv1.Service_Protocol) (map[string][]*registryv1.ServiceEndpoint, string)

GetAllWithVersion returns all endpoints and the current version.

func (*Snapshot) Replace 

func (s *Snapshot) Replace(endpoints map[string]map[registryv1.Service_Protocol][]*registryv1.ServiceEndpoint) (string, []*registrarv1.WatchEndpointsResponse)

Replace atomically replaces the entire snapshot contents with the provided endpoints and bumps the version. It returns the new version string plus the service-catalog transitions (see Apply) between the old and new contents; the caller stamps and broadcasts them.

func (*Snapshot) ServiceNames 

func (s *Snapshot) ServiceNames() []string

ServiceNames returns the sorted names of all services currently holding at least one endpoint — the service catalog replayed to every new watcher.

func (*Snapshot) Version 

func (s *Snapshot) Version() string

Version returns the current snapshot version as a string.

type Syncer 

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

Syncer periodically polls an external registry, computes a diff against the local snapshot, and broadcasts changes to all watching agents. It implements the controller-runtime Runnable interface.

func NewSyncer 

func NewSyncer(reg registry.Registry, snapshot *Snapshot, broadcaster *Broadcaster, syncInterval time.Duration, log *slog.Logger, metrics *Metrics) *Syncer

NewSyncer creates a Syncer that polls the external registry at the given interval. metrics may be nil to disable instrumentation.

func (*Syncer) NeedLeaderElection 

func (s *Syncer) NeedLeaderElection() bool

NeedLeaderElection returns false so the syncer runs on all replicas, not just the leader. Each replica independently polls the external registry and broadcasts changes to its connected agents.

func (*Syncer) Start 

func (s *Syncer) Start(ctx context.Context) error

Start runs the sync loop until the context is cancelled. When the registry supports change notifications (registry.ChangeNotifier — the etcd backend's clientv3 watch), the loop syncs at watch speed (debounced) instead of only at the poll interval; the periodic poll remains a backstop for any event missed during a watch re-establish. Backends without notifications (e.g. DynamoDB) fall back to poll-only, unchanged.

func (*Syncer) Synced 

func (s *Syncer) Synced() <-chan struct{}

Synced returns a channel that is closed once the first sync cycle has completed and the snapshot reflects the external registry.

func (*Syncer) UseWriteBehind 

func (s *Syncer) UseWriteBehind(q *WriteBehindQueue)

UseWriteBehind wires the write-behind queue's pending-intent overlay into the sync cycle.

type WriteBehindQueue 

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

WriteBehindQueue flushes snapshot-first registry mutations to the external registry with retries, and shields the sync loop from un-observed intents. It implements controller-runtime's Runnable (all replicas).

func NewWriteBehindQueue 

func NewWriteBehindQueue(reg registry.Registry, log *slog.Logger, metrics *Metrics) *WriteBehindQueue

NewWriteBehindQueue creates the queue. metrics may be nil.

func (*WriteBehindQueue) EnqueueRegister 

func (q *WriteBehindQueue) EnqueueRegister(service string, protocol registryv1.Service_Protocol, ep *registryv1.ServiceEndpoint)

EnqueueRegister records a register/upsert intent. A newer op for the same (service, protocol, ip) supersedes any older one.

func (*WriteBehindQueue) EnqueueUnregister 

func (q *WriteBehindQueue) EnqueueUnregister(service string, protocol registryv1.Service_Protocol, ip string)

EnqueueUnregister records a removal intent, superseding any pending register.

func (*WriteBehindQueue) NeedLeaderElection 

func (q *WriteBehindQueue) NeedLeaderElection() bool

NeedLeaderElection returns false: each replica owns the external writes for the RPCs it received (peer replicas converge through the external registry until peer-watch lands).

func (*WriteBehindQueue) Overlay 

func (q *WriteBehindQueue) Overlay(state map[string]map[registryv1.Service_Protocol][]*registryv1.ServiceEndpoint)

Overlay reconciles the queue against a freshly fetched external-registry state and patches that state with still-pending intents. Called by the sync loop BEFORE Diff/Replace, so neither the broadcast events nor the snapshot regress an intent the external registry has not materialized yet.

Release rule: a flushed op whose intent the fetched state reflects (register: key present with an equal endpoint; unregister: key absent) is done and removed. Everything else is overlaid onto the state.

func (*WriteBehindQueue) Shielding 

func (q *WriteBehindQueue) Shielding(service, ip string) bool

Shielding reports whether an intent for (service, ip) is still pending or flushed-but-unobserved. Diagnostic/test helper.

func (*WriteBehindQueue) Start 

func (q *WriteBehindQueue) Start(ctx context.Context) error

Start runs the flush loop until ctx ends. Implements Runnable.

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.