registrar

Documentation

Overview

Package registrar implements the Registry interface using a Registrar gRPC service. It caches endpoints locally from a server-streaming watch and delegates writes to the Registrar, which in turn persists them to the external registry backend.

Index

• type Config
• type RegistrarRegistry
  • func NewRegistrarRegistry(log *slog.Logger, cfg Config) *RegistrarRegistry
  • func (r *RegistrarRegistry) Changes() <-chan struct{}
  • func (r *RegistrarRegistry) Close() error
  • func (r *RegistrarRegistry) HasService(name string) bool
  • func (r *RegistrarRegistry) Initialize(ctx context.Context) error
  • func (r *RegistrarRegistry) ListAllEndpoints(ctx context.Context, protocol registryv1.Service_Protocol) (map[string][]*registryv1.ServiceEndpoint, error)
  • func (r *RegistrarRegistry) ListAllEndpointsAuthoritative(ctx context.Context, protocol registryv1.Service_Protocol) (map[string][]*registryv1.ServiceEndpoint, error)
  • func (r *RegistrarRegistry) ListConfig(ctx context.Context) ([]*registryv1.ServiceConfigProjection, error)
  • func (r *RegistrarRegistry) ListEndpoints(ctx context.Context, service string, protocol registryv1.Service_Protocol) ([]*registryv1.ServiceEndpoint, error)
  • func (r *RegistrarRegistry) Reconnects() <-chan struct{}
  • func (r *RegistrarRegistry) RegisterEndpoint(ctx context.Context, serviceName string, protocol registryv1.Service_Protocol, ...) error
  • func (r *RegistrarRegistry) SetServiceFilter(services []string)
  • func (r *RegistrarRegistry) UnregisterEndpoint(ctx context.Context, serviceName string, ip string) error
  • func (r *RegistrarRegistry) UnregisterEndpoints(ctx context.Context, serviceName string, ips []string) error
  • func (r *RegistrarRegistry) WaitReady(ctx context.Context) error

Types

type Config

type Config struct {
	// Address is the gRPC address of the Registrar service.
	Address string
	// ClusterName identifies this agent's cluster to the Registrar. Together with
	// NodeName it forms the unique watcher id; without it every agent collides on
	// the same id and the Registrar evicts their watch streams in a reconnect loop.
	ClusterName string
	// NodeName identifies this agent's node to the Registrar. See ClusterName.
	NodeName string
	// DialOptions are additional gRPC dial options (e.g., TLS credentials).
	// When empty, insecure credentials are used.
	DialOptions []grpc.DialOption
}

Config holds configuration for connecting to the Registrar service.

type RegistrarRegistry

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

RegistrarRegistry implements the Registry interface by communicating with a Registrar gRPC service. Reads are served from a local cache populated by a WatchEndpoints stream. Writes are delegated to the Registrar.

func NewRegistrarRegistry

func NewRegistrarRegistry(log *slog.Logger, cfg Config) *RegistrarRegistry

NewRegistrarRegistry creates a new RegistrarRegistry.

func (*RegistrarRegistry) Changes

func (r *RegistrarRegistry) Changes() <-chan struct{}

Changes returns a channel that receives a signal whenever the cached set of endpoints changes (an endpoint is added, updated, or removed). Signals are coalesced: consumers should treat each receive as "something changed, re-read the registry" rather than a per-event notification. It satisfies the registry.ChangeNotifier capability.

func (*RegistrarRegistry) Close

func (r *RegistrarRegistry) Close() error

Close shuts down the watch stream and closes the gRPC connection.

func (*RegistrarRegistry) HasService

func (r *RegistrarRegistry) HasService(name string) bool

HasService reports whether the named service currently has at least one endpoint anywhere in the mesh, answered from the local catalog. It satisfies the registry.ServiceCatalog capability.

func (*RegistrarRegistry) Initialize

func (r *RegistrarRegistry) Initialize(ctx context.Context) error

Initialize connects to the Registrar and starts the background watch stream.

func (*RegistrarRegistry) ListAllEndpoints

func (r *RegistrarRegistry) ListAllEndpoints(ctx context.Context, protocol registryv1.Service_Protocol) (map[string][]*registryv1.ServiceEndpoint, error)

ListAllEndpoints returns all endpoints from the local cache. Falls back to the Registrar RPC if the cache is empty.

func (*RegistrarRegistry) ListAllEndpointsAuthoritative

func (r *RegistrarRegistry) ListAllEndpointsAuthoritative(ctx context.Context, protocol registryv1.Service_Protocol) (map[string][]*registryv1.ServiceEndpoint, error)

ListAllEndpointsAuthoritative lists endpoints via the ListAllEndpoints RPC, bypassing the watch-fed cache. It satisfies the registry.AuthoritativeLister capability: the cache can be a stale superset of a fresh registrar's snapshot (an empty snapshot emits no FULL_SNAPSHOT events, so the cache is never cleared), and reconciliation diffing against it would silently no-op.

func (*RegistrarRegistry) ListConfig

func (r *RegistrarRegistry) ListConfig(ctx context.Context) ([]*registryv1.ServiceConfigProjection, error)

ListConfig fetches the clusterset-wide config projections via the registrar's ListAllConfig RPC (proposal 026). It satisfies registry.ConfigImporter: the agent imports cross-cluster GAMMA config through the registrar, never reading the store directly. An empty result (e.g. a backend with no cross-cluster config plane) is not an error.

func (*RegistrarRegistry) ListEndpoints

func (r *RegistrarRegistry) ListEndpoints(ctx context.Context, service string, protocol registryv1.Service_Protocol) ([]*registryv1.ServiceEndpoint, error)

ListEndpoints returns endpoints for a service from the local cache. Falls back to the Registrar RPC if the cache is empty.

func (*RegistrarRegistry) Reconnects

func (r *RegistrarRegistry) Reconnects() <-chan struct{}

Reconnects returns a channel receiving a (coalesced) signal after each successful watch stream (re)connection. It satisfies the registry.ReconnectNotifier capability.

func (*RegistrarRegistry) RegisterEndpoint

func (r *RegistrarRegistry) RegisterEndpoint(ctx context.Context, serviceName string, protocol registryv1.Service_Protocol, endpoint *registryv1.ServiceEndpoint) error

RegisterEndpoint delegates to the Registrar's RegisterEndpoint RPC.

func (*RegistrarRegistry) SetServiceFilter

func (r *RegistrarRegistry) SetServiceFilter(services []string)

SetServiceFilter scopes the endpoint watch to the given services (the node's dependency set). nil restores the full watch; an empty non-nil set watches nothing. If the effective filter changed while a stream is active, the stream is cancelled so the loop reconnects re-asserting the new filter with a cleared resume token (the registrar must resend the snapshot — the new scope may include services the old stream never delivered). It satisfies the registry.WatchScoper capability.

func (*RegistrarRegistry) UnregisterEndpoint

func (r *RegistrarRegistry) UnregisterEndpoint(ctx context.Context, serviceName string, ip string) error

UnregisterEndpoint delegates to the Registrar's UnregisterEndpoint RPC.

func (*RegistrarRegistry) UnregisterEndpoints

func (r *RegistrarRegistry) UnregisterEndpoints(ctx context.Context, serviceName string, ips []string) error

UnregisterEndpoints delegates to the Registrar's UnregisterEndpoint RPC with multiple IPs.

func (*RegistrarRegistry) WaitReady

func (r *RegistrarRegistry) WaitReady(ctx context.Context) error

WaitReady blocks until the watch cache holds a complete snapshot (the first SNAPSHOT_COMPLETE event) or ctx ends. It satisfies the registry.ReadyWaiter capability; callers bound it with a context timeout and may proceed with degraded (RPC-fallback) reads on expiry.