Documentation
¶
Overview ¶
Package api pkg/dmsg/discovery/api/api.go c1-net-dmsg
Package api pkg/dmsg/discovery/api/charts.go c4-net-discovery
Package api pkg/dmsg/discovery/api/cxo_publisher.go c1-net-dmsg
CXO publisher for dmsg-discovery's "clients by server" view.
The hypervisor's network visualizer needs to know "which clients are currently delegated to each dmsg server" — the same view AllClientsByServer / ClientsByServer expose over HTTP. This publisher mirrors that as a TreeStore feed.
Wire shape: one presence leaf per server ¶
Each dmsg server with at least one delegated client has one leaf:
clients-by-server/<server-pk> // FrameGzip(v2, empty)
The leaf only says the server has clients; TPD uses that to mark the servers hypervisors run for their LAN. It is written when a server gains its first client and deleted when it loses its last, so client churn does not touch the feed. Version 1 carried every client's entry, and each change resent its server's whole list to every visor holding the feed, 1.4 TB a day out of prod01. Who is delegated to which server is on HTTP (GET /dmsg-discovery/servers/clients and /entry/{pk}).
Interop ¶
Readers of the version 1 lists (tpviz and the hypervisor's visualizer) skip a version 2 leaf, find no entries and fall back to HTTP. Server entries (servers/<pk>, serverfeed) and leaf paths are unchanged, so the server list and TPD's roles still read from the feed.
Server entries are ignored by this publisher — a server PK is the path-prefix bucket, not a member of the view.
HTTP-path decoupling: the HTTP register / deregister handlers call PublishSetEntry / PublishDelEntry inline. Internally those route every mutation through a single buffered-channel worker (mirrors SD's pattern) so the HTTP goroutine never blocks on the treestore.Publisher mutex — which under load is contended by subscriber I/O and can stall register throughput long enough to time out visor heartbeats. The per-server client state map is owned by that single worker goroutine, so its mutations need no lock. Overflow drops are counted; CXO data quality degrades gracefully, HTTP stays fast.
Package api pkg/dmsg/discovery/api/cxo_servers.go
Registered dmsg servers on the clients-by-server feed, so visors keep their server cache from the snapshot they already hold instead of asking for all_servers over HTTP every few minutes.
Each server is one leaf at clients-by-server/servers/<server-pk>: its signed discovery entry, version-framed and gzipped. The path sits under the feed's prefix because subscribers fetch only that prefix. Older readers ignore it: they take clients-by-server/<pk> as a batch and a deeper path only when it ends in /entry.
A server re-signs its entry on every refresh and its free session count moves with every client, so a leaf is rewritten only when something a client dials by changes: its addresses, type, version or protocol. The signed body stays valid; its timestamp and session count are as of that change.
Package api pkg/dmsg/discovery/api/error_handler.go c1-net-dmsg
Package api pkg/dmsg/discovery/api/ratelimit.go c1-net-dmsg
Per-client request rate limiting for the discovery HTTP API. This is the server-side counterpart to the client-side serve-loop floor: it protects the discovery service even from clients that DON'T have that fix, by capping how many requests any single dmsg client can issue per second. The limits are set far above any legitimate client's discovery usage; the only thing they stop is a misbehaving client busy-looping its discover/dial loop and pinning the service at 100% CPU with tens of thousands of requests a second.
Keyed by the dmsg public key — the noise-authenticated transport identity in RemoteAddr — NOT a request header, so a client can't evade the limit by opening fresh streams (the PK is constant) or by spoofing X-Real-IP. The middleware is registered BEFORE chi's RealIP middleware for exactly that reason. Fail-open: anything that isn't a dmsg PK (clearnet IPs, which may sit behind a shared proxy) passes through untouched.
Package api pkg/dmsg/discovery/api/registration_ingest.go c1-net-dmsg
Registration-over-CXO ingest: the server side of the fan-in path where visors publish their own signed discovery entry as a CXO feed instead of re-PUTting it over HTTP on a timer (each PUT a fresh Noise + PQ handshake — the load that dominates dmsg-discovery CPU). The CXO aggregator (pkg/dmsg/discovery/regcxo) calls IngestEntryFromCXO for each entry it replicates.
Package api pkg/dmsg/discovery/api/server_health.go c2-dmsg-core
Index ¶
- Variables
- func BatchLeafPath(serverPK cipher.PubKey) string
- type API
- func (a *API) AllServers(ctx context.Context, _ logrus.FieldLogger) (entries []*disc.Entry, err error)
- func (a *API) BackfillDHTMirror(ctx context.Context, log logrus.FieldLogger)
- func (a *API) Charts() *charts.Page
- func (a *API) ChartsPage(w http.ResponseWriter, r *http.Request)
- func (a *API) IngestEntryFromCXO(ctx context.Context, entry *disc.Entry, reporter cipher.PubKey)
- func (a *API) RunBackgroundTasks(ctx context.Context, log logrus.FieldLogger)
- func (a *API) RunServersCXO(ctx context.Context, log logrus.FieldLogger)
- func (a *API) SetClientsByServerCXOPublisher(p *ClientsByServerCXOPublisher)
- func (a *API) SetDHTMirror(m interface{ ... })
- func (a *API) SetServerHealthClient(c *http.Client)
- func (a *API) StartCharts(ctx context.Context, st charts.Store, log logrus.FieldLogger)
- func (a *API) WarmCXOFromStore(ctx context.Context, log logrus.FieldLogger)
- type ClientsByServerCXOPublisher
- func (p *ClientsByServerCXOPublisher) Close() error
- func (p *ClientsByServerCXOPublisher) Dropped() uint64
- func (p *ClientsByServerCXOPublisher) FeedPK() cipher.PubKey
- func (p *ClientsByServerCXOPublisher) LastError() error
- func (p *ClientsByServerCXOPublisher) PublishDelEntry(oldEntry *disc.Entry)
- func (p *ClientsByServerCXOPublisher) PublishServers(servers []*disc.Entry)
- func (p *ClientsByServerCXOPublisher) PublishSetEntry(oldEntry, newEntry *disc.Entry)
Constants ¶
This section is empty.
Variables ¶
var WhitelistPKs = httpauth.NewWhitelist()
WhitelistPKs holds the keys allowed to call the bulk deregistration endpoint.
Functions ¶
func BatchLeafPath ¶
BatchLeafPath is exported so subscribers can reconstruct the batched per-server leaf path without duplicating the format string.
Types ¶
type API ¶
type API struct {
// Stats, when set, adds the process and traffic charts to the status page.
Stats *charts.ServiceStats
http.Handler
DmsgServers []string
OfficialServers map[string]bool
// contains filtered or unexported fields
}
API represents the api of the dmsg-discovery service`
func New ¶
func New(log logrus.FieldLogger, db store.Storer, m metrics.Metrics, testMode, enableLoadTesting, enableMetrics bool, dmsgAddr, authPassphrase string, clientEntryTTL time.Duration) *API
New returns a new API object, which can be started as a server. clientEntryTTL is the Redis expiration applied to client entries; 0 disables expiration (legacy behavior).
func (*API) AllServers ¶
func (a *API) AllServers(ctx context.Context, _ logrus.FieldLogger) (entries []*disc.Entry, err error)
AllServers is used to get all the available servers registered to the dmsg-discovery.
func (*API) BackfillDHTMirror ¶
func (a *API) BackfillDHTMirror(ctx context.Context, log logrus.FieldLogger)
BackfillDHTMirror iterates all existing entries in the store and mirrors them to the DHT. This ensures the DHT has the full dataset on startup, not just entries updated since the last restart.
func (*API) ChartsPage ¶ added in v1.3.100
func (a *API) ChartsPage(w http.ResponseWriter, r *http.Request)
ChartsPage serves the charts page, or 404 when charts are not running.
func (*API) IngestEntryFromCXO ¶
IngestEntryFromCXO applies a client discovery entry received over the registration-over-CXO feed. reporter is the feed's publisher PK (the visor); it MUST equal entry.Static since a visor may only publish its own entry.
It mirrors the validation and side effects of the HTTP setEntry handler (Validate, VerifySignature, sequence-iteration guard, SetEntry, DHT mirror, clients-by-server CXO publish, heartbeat) with ONE difference: a stale-or-equal sequence is a silent idempotent no-op rather than an error. During the migration a visor dual-writes the same entry over HTTP and CXO, so whichever path lands first wins and the other simply finds the store already at that sequence. Keep this in sync with (*API).setEntry — deliberately duplicated (not refactored out of the hot HTTP path) to keep this additive, opt-in path from touching production registration behavior.
func (*API) RunBackgroundTasks ¶
func (a *API) RunBackgroundTasks(ctx context.Context, log logrus.FieldLogger)
RunBackgroundTasks is goroutine which runs in background periodic tasks of dmsg-discovery.
func (*API) RunServersCXO ¶ added in v1.3.100
func (a *API) RunServersCXO(ctx context.Context, log logrus.FieldLogger)
RunServersCXO publishes the registered servers to the feed every minute until ctx ends. A server whose entry expires drops out on the next round.
func (*API) SetClientsByServerCXOPublisher ¶
func (a *API) SetClientsByServerCXOPublisher(p *ClientsByServerCXOPublisher)
SetClientsByServerCXOPublisher installs the CXO publisher that mirrors set/del entry events into a TreeStore feed at clients-by-server/<server>/<client>/{entry,tombstone}. Pass nil to disable. Wired by cmd/dmsg/dmsg-discovery after the publisher is built; no-op until then.
func (*API) SetDHTMirror ¶
func (a *API) SetDHTMirror(m interface { Mirror(subjectPK cipher.PubKey, entry interface{}, seq uint64) Delete(subjectPK cipher.PubKey) })
SetDHTMirror sets a mirror that publishes entries to the DHT on every successful SetEntry and removes them from the DHT on DelEntry.
func (*API) SetServerHealthClient ¶ added in v1.3.100
SetServerHealthClient lets the status page read each registered dmsg server's load from its /health, with c reaching them over dmsg.
func (*API) StartCharts ¶ added in v1.3.100
StartCharts samples the discovery's counts into st until ctx ends and serves them as the charts page.
func (*API) WarmCXOFromStore ¶
func (a *API) WarmCXOFromStore(ctx context.Context, log logrus.FieldLogger)
WarmCXOFromStore pre-populates the CXO publisher's tree from the current set of client entries in the store. Without this, the publisher only Puts on register / heartbeat events — so right after a dmsg-discovery restart the publisher's tree is empty until the first event lands, and any subscriber that connects in the post-restart gap times out at firstSyncTimeout (10s) with "timeout waiting for Root".
Mirrors the existing BackfillDHTMirror pattern: walk AllClientEntries and replay each as a fresh SetEntry (oldEntry=nil so the heartbeat skip in PublishSetEntry doesn't elide the Put). Best-effort — per-entry errors don't abort the warm.
type ClientsByServerCXOPublisher ¶
type ClientsByServerCXOPublisher struct {
// contains filtered or unexported fields
}
ClientsByServerCXOPublisher mirrors dmsg-discovery's clients-by- server view into a CXO TreeStore feed. Started automatically at DMSG-D startup whenever DMSG is enabled; the API calls into it via SetClientsByServerCXOPublisher.
func StartClientsByServerCXOPublisher ¶
func StartClientsByServerCXOPublisher(dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*ClientsByServerCXOPublisher, error)
StartClientsByServerCXOPublisher constructs the publisher backed by the given dmsg client and DMSG-D secret key. The publisher's allowlist is open — the underlying HTTP routes are public reads, so the CXO mirror inherits that. Returns nil + error if the publisher can't be created (no dmsg client, listener bind failure, etc.); callers log and continue without it (HTTP path stays the source of truth).
func (*ClientsByServerCXOPublisher) Close ¶
func (p *ClientsByServerCXOPublisher) Close() error
Close stops the worker goroutine and the underlying publisher. Pending events at the moment of Close are discarded. Safe to call multiple times.
func (*ClientsByServerCXOPublisher) Dropped ¶
func (p *ClientsByServerCXOPublisher) Dropped() uint64
Dropped returns the cumulative count of dropped publish operations. Exposed for /health-style introspection.
func (*ClientsByServerCXOPublisher) FeedPK ¶
func (p *ClientsByServerCXOPublisher) FeedPK() cipher.PubKey
FeedPK returns the publisher's feed PK (DMSG-D's own PK).
func (*ClientsByServerCXOPublisher) LastError ¶
func (p *ClientsByServerCXOPublisher) LastError() error
LastError returns the most recent publish error, or nil.
func (*ClientsByServerCXOPublisher) PublishDelEntry ¶
func (p *ClientsByServerCXOPublisher) PublishDelEntry(oldEntry *disc.Entry)
PublishDelEntry mirrors a full client-entry delete: removes the client from every server it was delegated to and re-encodes those servers' batched leaves. oldEntry is the entry being deleted (caller fetches it from store before the DelEntry call). The client's absence from a server's leaf IS the deregistration signal — there is no tombstone leaf (see the package docstring). Non-blocking.
func (*ClientsByServerCXOPublisher) PublishServers ¶ added in v1.3.100
func (p *ClientsByServerCXOPublisher) PublishServers(servers []*disc.Entry)
PublishServers makes the server leaves match servers: a new or materially changed server is written and one no longer present is deleted.
func (*ClientsByServerCXOPublisher) PublishSetEntry ¶
func (p *ClientsByServerCXOPublisher) PublishSetEntry(oldEntry, newEntry *disc.Entry)
PublishSetEntry mirrors a set/update of a client entry. Updates the worker's per-server client state — adding the client to every server in the new DelegatedServers list, removing it from any server it left (compared to old) — and re-encodes each affected server's batched leaf. Does nothing for server entries (they're path buckets here, not members) or for client entries with empty DelegatedServers (the client is absent from every server's leaf, which is exactly a deregistration). Non-blocking: the diff and JSON encode of the entry happen on the caller's goroutine; the state mutation + per-server re-encode/Put runs on the worker.
Source Files