Documentation
¶
Overview ¶
Package endpoints owns the OrderedLog op types that publish service endpoint sets through the seam established in RUNE-039.
Endpoints are keyed by service ID. Each UpdateEndpoints op replaces the entire endpoint set for that service (last-writer-wins per service ID). DeleteEndpoints removes the entry entirely. The dataplane subsystem (RUNE-041) is the primary consumer: it watches the OrderedLog stream, filters mutations under the protected "endpoints/" prefix, and reconciles its in-memory cache + per-VIP proxy listeners.
The producer is intentionally not the orchestrator (yet). The instance/health controllers will start calling Publisher.Update once the LocalInstances pipeline lands in RUNE-063. For now, this package provides:
- The op types + appliers (Register on a backend);
- The Publisher API for any owner of the endpoint set;
- A reverse Reader so consumers can hydrate their cache after a snapshot recovery without re-replaying the entire log.
Key prefix: "endpoints/<serviceID>" — protected by the seam-leakage lint in scripts/check_orderedlog_seam.sh.
Index ¶
Constants ¶
const ( OpUpdateEndpoints = "endpoints.update" OpDeleteEndpoints = "endpoints.delete" )
Reserved op type strings.
Variables ¶
var ErrServiceIDRequired = errors.New("endpoints: service ID required")
ErrServiceIDRequired is returned when an op is constructed with an empty service ID. The applier rejects the same condition.
Functions ¶
func DecodePayload ¶
func DecodePayload(m orderedlog.Mutation) (types.ServiceEndpoints, error)
DecodePayload unmarshals an UpdateEndpoints mutation payload into a ServiceEndpoints. Returns an error for empty or delete mutations.
func IsEndpointsMutation ¶
func IsEndpointsMutation(m orderedlog.Mutation) bool
IsEndpointsMutation reports whether m was produced by this package. Consumers (e.g. the dataplane watch loop) use this to filter the stream cheaply.
func Register ¶
func Register(olog orderedlog.OrderedLog) error
Register installs the endpoints op types + appliers on the given OrderedLog. Idempotent in the sense that it tolerates orderedlog.ErrAlreadyRegistered (so multiple consumers in the same process — e.g. tests — don't fight). Must be called before any Publisher.Update / Delete.
Types ¶
type Publisher ¶
type Publisher interface {
// Update replaces the endpoint set for serviceID. Empty endpoints
// is allowed and means "this service has no ready instances right
// now"; the data plane will fail-closed on connections to its VIP
// in that case. To remove the entry entirely (e.g. service
// deletion), use Delete instead.
Update(ctx context.Context, serviceID string, endpoints []types.Endpoint) error
// Delete removes the endpoint set for serviceID. Idempotent.
Delete(ctx context.Context, serviceID string) error
}
Publisher is the public producer API. The orchestrator instance/ health controllers call this whenever the endpoint set for a service changes (instance Ready, instance Stopped, health flip).
Implementations must be safe for concurrent use; the default implementation backed by orderedlog.OrderedLog is.
func NewPublisher ¶
func NewPublisher(olog orderedlog.OrderedLog) Publisher
NewPublisher returns a Publisher that writes endpoint mutations through olog.Propose. The caller must have already called Register on the same olog (otherwise the proposes return ErrUnknownOpType).
Source Files