Documentation
¶
Overview ¶
Package discovery wires Gantry's libp2p host and Kademlia DHT.
Design :
- Each agent runs a libp2p host with TCP and (optionally) QUIC transports and Noise security. The host's identity is persisted to a hostPath so restarts don't churn the DHT routing table. - The host participates in a cluster-scoped Kademlia DHT (`/gantry/kad/1` protocol prefix) in **server mode**. Server mode is required because every Gantry agent is also a content provider; client-mode peers would not contribute provider records. - Provider records are keyed by a CIDv1 wrapping the OCI digest's raw 32 bytes via SHA2-256 multihash. The CID derivation is deterministic and documented inline so any agent can re-derive the same CID from the same digest.
scope:
- Host + DHT bring-up, Provide / FindProviders. - Health returns the geometric-mean health score from internal/discovery/health.go (routing-table coverage, p95 lookup latency, self-test success rate); in test mode where no Monitor is wired it returns 1.0. - Bootstrap pulls from operator-supplied `Libp2pBootstrapPeers` plus the dynamic K8s pod-annotation pool (see cmd/gantry/main.go announceSelfAndBootstrap): every Gantry pod self-patches its peer.AddrInfo on `gantry.io/p2p-addrs`, Members surfaces those entries via SnapshotForBootstrap, and this package's ConnectPeers dials them with the 8/5/32 cascade.
Index ¶
- Constants
- func DigestToCID(d digest.Digest) (cid.Cid, error)
- type Host
- func (h *Host) Addrs() []multiaddr.Multiaddr
- func (h *Host) Close() error
- func (h *Host) ConnectPeers(ctx context.Context, multiaddrs []string) int
- func (h *Host) FindProviders(ctx context.Context, d digest.Digest) ([]ifaces.Provider, error)
- func (h *Host) Health() float64
- func (h *Host) LibP2P() host.Host
- func (h *Host) Monitor() *Monitor
- func (h *Host) PeerID() peer.ID
- func (h *Host) Provide(ctx context.Context, d digest.Digest) error
- func (h *Host) RoutingTableSize() int
- func (h *Host) Withdraw(_ context.Context, _ digest.Digest) error
- type Monitor
- func (m *Monitor) InBootstrapWindow(bootstrapWindow time.Duration, bootstrapRoutingTablePct int) bool
- func (m *Monitor) ObserveLatency(d time.Duration)
- func (m *Monitor) RecordSelfTest(ok bool)
- func (m *Monitor) RunSelfTestLoop(ctx context.Context, period time.Duration, selfTest func(context.Context) bool)
- func (m *Monitor) Score() float64
- func (m *Monitor) SetRoutingTableTarget(fn func() int)
- func (m *Monitor) State() string
- type MonitorOptions
- type Options
Constants ¶
const DefaultTransferPort = 5001
DefaultTransferPort is the conventional peer-transfer port used when Options.TransferPort is zero. Kept exported so callers building Options by hand can spell it explicitly.
Variables ¶
This section is empty.
Functions ¶
func DigestToCID ¶
DigestToCID maps an OCI digest to the CIDv1 used as the DHT provider key. CID derivation: Multihash(sha256, raw_digest_bytes) wrapped in CID v1 with codec=raw. The derivation is deterministic so any agent re-derives the same CID from the same digest.
Types ¶
type Host ¶
type Host struct {
// contains filtered or unexported fields
}
Host wraps a libp2p host + kad-dht and implements ifaces.DHT.
func New ¶
New builds a Host and joins the DHT in server mode. The returned Host is already announcing - Provide/FindProviders are usable immediately, though FindProviders may return empty until the routing table converges.
func (*Host) ConnectPeers ¶
ConnectPeers dials a set of multiaddr strings in parallel with a 5s per-peer timeout. Used by main.go to seed the DHT routing table from the membership view (the design doc): after members.WaitForSync, every Ready peer with a published p2p multiaddr is fed back into the libp2p host so kad-dht has direct-connect seeds even without operator-supplied bootstrap_peers config.
Returns the number of peers that successfully connected. Failures are logged at DEBUG and do not fail the call.
func (*Host) FindProviders ¶
FindProviders implements ifaces.DHT. Returns providers whose multiaddrs expose at least one IP-based transport (TCP or QUIC). Provider.NodeID is the libp2p peer.ID as a string; Provider.Addr is the first IP-based multiaddr's IP, suffixed with the conventional transfer port `:5001`. coord layer will reconcile peer.ID with k8s NodeID using Members; callers (the mirror miss path) only need a dialable transfer URL.
func (*Host) Health ¶
Health returns the geometric-mean health score (routing-table coverage, p95 lookup latency, self-test success rate). Returns 1.0 when no monitor is wired (test mode).
func (*Host) LibP2P ¶
LibP2P returns the underlying libp2p host. Used by the // coord-stream wiring (internal/coord) to attach the gRPC-over-libp2p stream handler to the same host that runs the DHT.
func (*Host) Monitor ¶
Monitor returns the underlying health monitor for callers that need finer-grained signals (e.g. `InBootstrapWindow`). May be nil when the host was constructed without monitoring (test mode).
func (*Host) RoutingTableSize ¶
RoutingTableSize returns the current kad-dht routing-table size. Used by readiness probes and the health score.
func (*Host) Withdraw ¶
Withdraw implements ifaces.DHT. libp2p kad-dht has no protocol-level withdraw - provider records expire at the 24 h TTL. The advertiser achieves the same effect by simply not re-calling Provide for withdrawn digests on its next refresh tick, so this hook exists purely as a cooperation point for the interface contract and is a no-op today. Future work may emit a libp2p custom protocol message to peers in the routing table to evict the stale record sooner, but the plan explicitly accepts TTL drainage as adequate.
type Monitor ¶
type Monitor struct {
// contains filtered or unexported fields
}
Monitor implements the design doc DHT health scoring. It is safe for concurrent use.
func NewMonitor ¶
func NewMonitor(opts MonitorOptions) *Monitor
NewMonitor builds a Monitor. RoutingTableSize must be non-nil; all other fields receive defaults.
func (*Monitor) InBootstrapWindow ¶
func (m *Monitor) InBootstrapWindow(bootstrapWindow time.Duration, bootstrapRoutingTablePct int) bool
InBootstrapWindow reports whether the agent is still in the suppression window. direct-origin-fallback origin fallback must not fire while this is true: an empty DHT result during early bootstrap is a false signal, not evidence of a cold cluster.
The suppression applies when either: - now-started < bootstrapWindow, OR - routing-table size < (routingTableTarget × bootstrapRoutingTablePct / 100).
When RoutingTableTarget is nil or returns 0 (no known cluster size), only the time component is consulted.
func (*Monitor) ObserveLatency ¶
ObserveLatency records the duration of a successful DHT lookup. Failed lookups should be recorded via the self-test path instead; per-call failures are not penalised here so a single slow node can't tank the score.
func (*Monitor) RecordSelfTest ¶
RecordSelfTest stores the outcome of one self-test cycle.
func (*Monitor) RunSelfTestLoop ¶
func (m *Monitor) RunSelfTestLoop(ctx context.Context, period time.Duration, selfTest func(context.Context) bool)
RunSelfTestLoop drives the periodic self-test cycle. It blocks until ctx is cancelled. `selfTest` should perform one Provide -> FindProviders round-trip and return whether it succeeded. Failures are debounced: a failed cycle only contributes to the score, the loop itself always continues.
func (*Monitor) Score ¶
Score returns the geometric mean of routing-table coverage, latency score, and self-test success rate. All three are in [0, 1]. With no data yet, the corresponding component reads 1.0 so a newly started agent isn't punished into Unhealthy before its window has filled.
func (*Monitor) SetRoutingTableTarget ¶
SetRoutingTableTarget swaps the routing-table-target closure atomically. Used by main to wire the membership-derived target (the design doc's `min(informer_node_count, kademlia_max_routing_table_size)`) after the informer has come online, since Monitor is constructed before memberView in discovery.New.
type MonitorOptions ¶
type MonitorOptions struct {
// Now is the time source; defaults to time.Now.
Now func() time.Time
// RoutingTableSize returns the current size of the local DHT
// routing table. Required.
RoutingTableSize func() int
// RoutingTableTarget returns the expected steady-state
// routing-table size, computed per the design doc as
// `min(informer_node_count, kademlia_max_routing_table_size)`.
// Nil or a return value <= 0 disables the routing-table
// component (it contributes 1.0).
RoutingTableTarget func() int
// LatencyWindow is the rolling window for p95 lookup latency.
// Defaults to 5min (the design doc).
LatencyWindow time.Duration
// LatencyFloor and LatencyCeiling bound the linear-interpolation
// region for the latency component. p95 ≤ floor -> 1.0; ≥ ceiling
// -> 0.0. Defaults: 200ms / 5s (the design doc).
LatencyFloor time.Duration
LatencyCeiling time.Duration
// SelfTestWindow is the number of most-recent self-test outcomes
// to retain. Defaults to 10.
SelfTestWindow int
}
MonitorOptions configures a Monitor.
type Options ¶
type Options struct {
// IdentityPath is the on-disk persistence path for the libp2p key.
// Empty means generate a fresh ephemeral identity (test mode).
IdentityPath string
// ListenAddrs is the list of multiaddrs the host advertises. Empty
// uses libp2p's defaults (which include /ip4/0.0.0.0/tcp/0).
ListenAddrs []string
// BootstrapPeers is the static list of peer multiaddrs to seed the
// DHT routing table on startup. Each entry must include /p2p/<peer.ID>.
BootstrapPeers []string
// ProtocolPrefix is the kad-dht protocol prefix, e.g. "/gantry". Empty
// uses kad-dht's default ("/ipfs"). Production should set this to
// isolate the cluster's DHT from other libp2p networks.
ProtocolPrefix string
// Logger is the structured logger; nil uses slog.Default.
Logger *slog.Logger
// RoutingTableTarget returns the expected steady-state routing-table
// size, computed per the design doc as `min(informer_node_count,
// kademlia_max_routing_table_size)`. Nil or a return value <= 0
// disables the routing-table component (Health's rt term reads
// 1.0). The closure is invoked on every score read so it reflects
// live cluster membership.
RoutingTableTarget func() int
// SelfTestPeriod is the interval between Provide(self_id) ->
// FindProviders(self_id) self-test cycles. Zero disables the
// background self-test loop (used in tests). Production default
// is 60s (the design doc).
SelfTestPeriod time.Duration
// TransferPort is the TCP port to suffix onto IP addresses returned
// by FindProviders. Zero defaults to 5001 (the design-doc value).
// In a cluster all agents share a transfer port by convention so
// inferring it from the local config is safe; the value travels
// through Options so test harnesses and operators that override the
// port don't get a hardcoded mismatch.
TransferPort int
}
Options configures the discovery host.
func FromConfig ¶
FromConfig builds Options from a *config.Config.
Source Files