diff --git a/pkg/agent/endpoint/conn.go b/pkg/agent/endpoint/conn.go index 9473b6ab3..98edd2f98 100644 --- a/pkg/agent/endpoint/conn.go +++ b/pkg/agent/endpoint/conn.go @@ -122,6 +122,17 @@ func (c *DataConn) OpenStreams() int { return len(c.streams) } +// SpareStreams reports how many more streams this conn can accept before its +// per-conn cap; never negative. +func (c *DataConn) SpareStreams() int { + c.mu.Lock() + defer c.mu.Unlock() + if n := int(c.params.MaxStreamsPerConn) - len(c.streams); n > 0 { + return n + } + return 0 +} + // HasCapacity reports whether a new stream may be opened. func (c *DataConn) HasCapacity() bool { c.mu.Lock() diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index 4754b4b2f..b4dbf5b6b 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -32,6 +32,24 @@ import ( "github.com/livekit/protocol/utils/guid" ) +// HeaderEndpointMiss marks a response the front produced ITSELF - a routing +// miss (no matching route, wrong method, auth required, or no local capacity) - +// as distinct from a response the worker's app returned through the bridge. A +// relay caller keys its cross-node retry on this header, so a worker's own 404 +// (e.g. GET /users/999 for a missing user) is never mistaken for "this node +// can't serve the path" and re-relayed. The value is the miss kind, so the +// caller can surface the most informative aggregate status. The header is set +// only on the private relay listener (see MarkMisses) and stripped by the relay +// caller, so it never reaches a client. +const HeaderEndpointMiss = "X-Livekit-Endpoint-Miss" + +const ( + MissNotFound = "notfound" + MissMethodNotAllowed = "methodnotallowed" + MissUnauthenticated = "unauthenticated" + MissUnavailable = "unavailable" +) + const ( // PathPrefix is the public route namespace: /agents/{deployment}/{path...} PathPrefix = "/agents/" @@ -45,15 +63,15 @@ const ( maxAttempts = 3 ) -// ScopeResolver maps an inbound request to its project scope. It returns the -// api key the request is authorized for (empty when unauthenticated) - the -// service layer implements it from validated grants. -type ScopeResolver func(r *http.Request) (apiKey string, authenticated bool) +// APIKeyResolver maps an inbound request to the api key it is authorized for +// (empty when unauthenticated) - the service layer implements it from validated +// grants. +type APIKeyResolver func(r *http.Request) (apiKey string, authenticated bool) type Front struct { - registry *Registry - resolveScope ScopeResolver - logger logger.Logger + registry *Registry + resolveAPIKey APIKeyResolver + logger logger.Logger // fallback is consulted when nothing local can serve the request (no // candidates, no route match, or every match without capacity); a @@ -62,21 +80,25 @@ type Front struct { fallback Fallback // see WithSingleKeyFallback singleKeyFallback bool + // see MarkMisses: set on the private relay listener so a relay caller can + // tell a routing miss from a worker-app response + markMisses bool } -func NewFront(registry *Registry, resolveScope ScopeResolver, log logger.Logger) *Front { +func NewFront(registry *Registry, resolveAPIKey APIKeyResolver, log logger.Logger) *Front { return &Front{ - registry: registry, - resolveScope: resolveScope, - logger: log.WithComponent("agents.endpoint"), + registry: registry, + resolveAPIKey: resolveAPIKey, + logger: log.WithComponent("agents.endpoint"), } } // FallbackRequest describes a request nothing local could serve. The request // body is untouched when the fallback runs. type FallbackRequest struct { - // Scope is the project identity the front resolved (api key in OSS) - Scope string + // APIKey is the identity the front resolved the request to (empty when + // unauthenticated) + APIKey string Authenticated bool Deployment string // Path within the deployment, '/'-rooted @@ -96,9 +118,28 @@ func (f *Front) WithFallback(fb Fallback) *Front { return f } +// MarkMisses tags the front's own routing-miss responses with HeaderEndpointMiss +// so a relay caller can distinguish them from worker-app responses. Set it on +// the private relay listener only; the public front must not (the header would +// leak to clients, and its misses are final anyway). +func (f *Front) MarkMisses() *Front { + f.markMisses = true + return f +} + +// writeMiss writes a front-originated miss, tagging it with the kind when this +// front marks misses (the relay listener) so the relay caller can retry past it +// and aggregate the most informative status. +func (f *Front) writeMiss(w http.ResponseWriter, status int, kind, msg string) { + if f.markMisses { + w.Header().Set(HeaderEndpointMiss, kind) + } + http.Error(w, msg, status) +} + // WithSingleKeyFallback resolves unauthenticated requests to the registry's -// single api key when the scope resolver yields none. Self-hosted convenience -// only: a multi-tenant front must never guess a scope from what happens to be +// single api key when the resolver yields none. Self-hosted convenience only: a +// multi-tenant front must never guess an api key from what happens to be // registered. func (f *Front) WithSingleKeyFallback() *Front { f.singleKeyFallback = true @@ -123,7 +164,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { isWS := isWebSocketUpgrade(r) - apiKey, authenticated := f.resolveScope(r) + apiKey, authenticated := f.resolveAPIKey(r) if apiKey == "" && f.singleKeyFallback { // unauthenticated: OSS serves public routes when the worker fleet // belongs to a single key @@ -131,14 +172,14 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if apiKey == "" { w.Header().Set("WWW-Authenticate", "Bearer") - http.Error(w, "authentication required", http.StatusUnauthorized) + f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required") return } candidates := f.registry.Candidates(apiKey, deployment) if len(candidates) == 0 && f.fallback == nil { w.Header().Set("Retry-After", "1") - http.Error(w, "no workers available for deployment", http.StatusServiceUnavailable) + f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no workers available for deployment") return } @@ -167,14 +208,14 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { if route == nil && f.fallback != nil { // nothing local can serve: hand off before the local status mapping if f.fallback(w, r, &FallbackRequest{ - Scope: apiKey, Authenticated: authenticated, + APIKey: apiKey, Authenticated: authenticated, Deployment: deployment, Path: path, WebSocket: isWS, }) { return } if len(candidates) == 0 && !restricted && !partial { w.Header().Set("Retry-After", "1") - http.Error(w, "no workers available for deployment", http.StatusServiceUnavailable) + f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no workers available for deployment") return } } @@ -182,9 +223,9 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { case restricted: w.Header().Set("WWW-Authenticate", "Bearer") - http.Error(w, "authentication required", http.StatusUnauthorized) + f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required") case partial: - http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + f.writeMiss(w, http.StatusMethodNotAllowed, MissMethodNotAllowed, "method not allowed") default: for _, reg := range candidates { if alt, ok := reg.Manifest.RedirectSlashes(path, r.Method, isWS); ok { @@ -194,7 +235,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } } - http.NotFound(w, r) + f.writeMiss(w, http.StatusNotFound, MissNotFound, "not found") } return } @@ -204,7 +245,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { attempted := make(map[*Registration]bool) for attempt := 0; attempt < maxAttempts; attempt++ { - reg := pickWeighted(matched, attempted) + reg := pickWorker(matched, attempted) if reg == nil { break } @@ -223,7 +264,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { // retryable. if bodyConsumed == 0 && f.fallback != nil { if f.fallback(w, r, &FallbackRequest{ - Scope: apiKey, Authenticated: authenticated, + APIKey: apiKey, Authenticated: authenticated, Deployment: deployment, Path: path, WebSocket: isWS, }) { return @@ -231,38 +272,49 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) { } w.Header().Set("Retry-After", "1") - http.Error(w, "no worker could serve the request", http.StatusServiceUnavailable) + f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no worker could serve the request") } -// pickWeighted is capacity-weighted random over non-draining workers with -// attached data connections. -func pickWeighted(regs []*Registration, ignore map[*Registration]bool) *Registration { - var sum float32 - weights := make([]float32, len(regs)) - for i, reg := range regs { +// pickWorker chooses a worker by the power of two choices: sample two eligible +// registrations at random and take the one with fewer in-flight streams (least +// outstanding requests). This approximates optimal load spreading without global +// coordination or the herding of exact least-loaded, and - unlike a +// self-reported load - the in-flight count is observed here on this node's own +// data conns, so it is accurate for adopted (satellite) registrations too. +// Eligible = not already attempted, has attached conns, not draining. +func pickWorker(regs []*Registration, ignore map[*Registration]bool) *Registration { + eligible := regs[:0:0] + for _, reg := range regs { if ignore[reg] || reg.AttachedConns() == 0 { continue } if reg.Draining != nil && reg.Draining() { continue } - w := float32(1) - if reg.Load != nil { - w = max(0.01, 1-reg.Load()) - } - weights[i] = w - sum += w + eligible = append(eligible, reg) } - if sum == 0 { + if len(eligible) == 0 { return nil } - target := rand.Float32() * sum - for i, reg := range regs { - if target -= weights[i]; weights[i] > 0 && target <= 0 { - return reg - } + return eligible[p2c(len(eligible), func(i int) int { return eligible[i].InflightStreams() })] +} + +// p2c returns the index of the less-loaded of two distinct random draws from +// [0,n) (n >= 1). With n == 2 both are always sampled, so it is exact; larger n +// trades a little optimality for O(1) work and no herding. +func p2c(n int, load func(int) int) int { + if n == 1 { + return 0 } - return nil + i := rand.IntN(n) + j := rand.IntN(n - 1) + if j >= i { // fold to a distinct second draw + j++ + } + if load(i) <= load(j) { + return i + } + return j } // bridge runs one attempt against one worker. done means a response (or abort) diff --git a/pkg/agent/endpoint/pick_test.go b/pkg/agent/endpoint/pick_test.go new file mode 100644 index 000000000..a3ff868ea --- /dev/null +++ b/pkg/agent/endpoint/pick_test.go @@ -0,0 +1,42 @@ +// Copyright 2026 LiveKit, Inc. + +package endpoint + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestP2CChoice(t *testing.T) { + // a single candidate is always chosen + require.Equal(t, 0, p2c(1, func(int) int { return 42 })) + + // with two candidates both are always drawn, so the lower-load one wins + // deterministically regardless of the random draw + for i := 0; i < 100; i++ { + require.Equal(t, 1, p2c(2, func(i int) int { return []int{5, 2}[i] })) + require.Equal(t, 0, p2c(2, func(i int) int { return []int{2, 5}[i] })) + } + + // larger n: the draws are always distinct and in range + for i := 0; i < 500; i++ { + idx := p2c(5, func(int) int { return 0 }) + require.GreaterOrEqual(t, idx, 0) + require.Less(t, idx, 5) + } + + // one clearly-idle candidate among four busy ones: whenever it is one of the + // two draws it wins, so it is picked far more than its 1/n share and the busy + // ones are relieved (this is the whole point of power-of-two-choices) + loads := []int{0, 100, 100, 100, 100} + low := 0 + for i := 0; i < 4000; i++ { + if p2c(len(loads), func(i int) int { return loads[i] }) == 0 { + low++ + } + } + // P(idle drawn) = 1 - (4/5)(3/4) = 0.4, and it always wins when drawn + require.Greater(t, low, 1000) + require.Less(t, low, 2400) +} diff --git a/pkg/agent/endpoint/registry.go b/pkg/agent/endpoint/registry.go index 4fca8171b..49a0b70fa 100644 --- a/pkg/agent/endpoint/registry.go +++ b/pkg/agent/endpoint/registry.go @@ -63,9 +63,9 @@ type Registration struct { Settings Settings Logger logger.Logger - // Load and Draining are provided by the control-plane layer that owns the - // worker (reported load rides UpdateWorkerStatus on the control connection). - Load func() float32 + // Draining is provided by the control-plane layer that owns the worker; a + // draining worker takes no new streams. Worker selection uses live in-flight + // streams, not a reported load, so no load hook is needed here. Draining func() bool // pendingAttaches counts slots reserved by validated-but-unadopted wires so @@ -199,6 +199,36 @@ func (r *Registration) AttachedConns() int { return len(r.conns) } +// InflightStreams reports the open streams across this registration's data +// conns - the requests currently being served from this node. It is the +// least-outstanding-requests signal for worker selection: a per-node view (each +// node counts only the streams it opened), which is exactly what a node needs +// to spread its own traffic, and it needs no load reported by the worker. +func (r *Registration) InflightStreams() int { + r.mu.Lock() + defer r.mu.Unlock() + n := 0 + for _, c := range r.conns { + n += c.OpenStreams() + } + return n +} + +// SpareStreams reports the total spare stream capacity across this +// registration's data conns on this node - how many more concurrent requests it +// can accept. It is the node's live, locally-observed serving headroom for the +// worker; being measured on the conns this node holds, it is accurate for +// adopted (satellite) registrations too, which report no load. +func (r *Registration) SpareStreams() int { + r.mu.Lock() + defer r.mu.Unlock() + n := 0 + for _, c := range r.conns { + n += c.SpareStreams() + } + return n +} + func (r *Registration) close() { r.mu.Lock() if r.closed { @@ -221,20 +251,6 @@ type Registry struct { mu sync.Mutex regs map[string]*Registration // by worker id byKey map[regKey][]*Registration - - // scope hooks fire when a scope (api key / project) gains its first or - // loses its last registration; the Remote uses them to (de)register the - // resolve topic on the bus - scopeCounts map[string]int - onScopeActive func(scope string) - onScopeIdle func(scope string) - - // hookMu serializes scope hook invocations. Transitions are decided against - // the registry's CURRENT state under hookMu, never from values computed - // earlier: a reconnect's activation racing the old connection's idle must - // not leave a live scope without its resolve topic. - hookMu sync.Mutex - scopeNotified map[string]bool } type regKey struct { @@ -244,62 +260,8 @@ type regKey struct { func NewRegistry() *Registry { return &Registry{ - regs: make(map[string]*Registration), - byKey: make(map[regKey][]*Registration), - scopeCounts: make(map[string]int), - scopeNotified: make(map[string]bool), - } -} - -// notifyScope reconciles the scope's hook state with the registry's current -// truth. Racing register/deregister notifications converge instead of -// interleaving. -func (g *Registry) notifyScope(scope string) { - g.hookMu.Lock() - defer g.hookMu.Unlock() - g.mu.Lock() - active := g.scopeCounts[scope] > 0 - activeHook, idleHook := g.onScopeActive, g.onScopeIdle - g.mu.Unlock() - if activeHook == nil && idleHook == nil { - return - } - if g.scopeNotified[scope] == active { - return - } - if active { - g.scopeNotified[scope] = true - activeHook(scope) - } else { - delete(g.scopeNotified, scope) - idleHook(scope) - } -} - -// SetScopeHooks installs the scope activation callbacks, replaying currently -// active scopes so a resolver layered on after registrations still registers -// its topics. Multi-node deployments use this to answer endpoint resolves only -// while they hold registrations for a scope. -func (g *Registry) SetScopeHooks(active, idle func(scope string)) { - g.hookMu.Lock() - defer g.hookMu.Unlock() - g.mu.Lock() - g.onScopeActive = active - g.onScopeIdle = idle - scopes := make([]string, 0, len(g.scopeCounts)) - for scope, n := range g.scopeCounts { - if n > 0 { - scopes = append(scopes, scope) - } - } - g.mu.Unlock() - if active != nil { - for _, s := range scopes { - if !g.scopeNotified[s] { - g.scopeNotified[s] = true - active(s) - } - } + regs: make(map[string]*Registration), + byKey: make(map[regKey][]*Registration), } } @@ -321,13 +283,8 @@ func (g *Registry) Register(r *Registration) error { } g.regs[r.WorkerID] = r g.byKey[key] = append(g.byKey[key], r) - g.scopeCounts[r.APIKey]++ g.mu.Unlock() - g.notifyScope(r.APIKey) if old != nil { - if old.APIKey != r.APIKey { - g.notifyScope(old.APIKey) - } old.close() } return nil @@ -347,10 +304,6 @@ func (g *Registry) removeLocked(r *Registration) { g.byKey[key] = regs } } - g.scopeCounts[r.APIKey]-- - if g.scopeCounts[r.APIKey] == 0 { - delete(g.scopeCounts, r.APIKey) - } } // Deregister removes exactly this registration; it is a no-op when a newer @@ -363,31 +316,40 @@ func (g *Registry) Deregister(r *Registration) { } g.removeLocked(r) g.mu.Unlock() - g.notifyScope(r.APIKey) r.close() } // ValidateAttach checks an attach without adopting the connection, so the -// attach response can be written before stream frames may flow. -func (g *Registry) ValidateAttach(workerID, instanceID, token string) error { +// attach response can be written before stream frames may flow. apiKey is the +// project the connecting grant is authorized for; it must match the +// registration's, so a leaked attach token is useless without a grant for the +// worker's own project (defense in depth over the token secret). +func (g *Registry) ValidateAttach(workerID, instanceID, apiKey, token string) error { g.mu.Lock() r, ok := g.regs[workerID] g.mu.Unlock() if !ok { return ErrUnknownWorker } + if apiKey != r.APIKey { + return ErrAttachRejected // wrong project: terminal, never adopt-retried + } return r.validateAttach(instanceID, token) } // BeginAttach validates an attach and reserves a pool slot; the caller writes -// the ack and then Completes (or Aborts) the ticket. -func (g *Registry) BeginAttach(workerID, instanceID, token string) (*AttachTicket, error) { +// the ack and then Completes (or Aborts) the ticket. apiKey binds the attach to +// the connecting grant's project (see ValidateAttach). +func (g *Registry) BeginAttach(workerID, instanceID, apiKey, token string) (*AttachTicket, error) { g.mu.Lock() r, ok := g.regs[workerID] g.mu.Unlock() if !ok { return nil, ErrUnknownWorker } + if apiKey != r.APIKey { + return nil, ErrAttachRejected // wrong project: terminal, never adopt-retried + } return r.beginAttach(instanceID, token) } diff --git a/pkg/agent/endpoint/registry_test.go b/pkg/agent/endpoint/registry_test.go index fb3e3a67a..559e51e74 100644 --- a/pkg/agent/endpoint/registry_test.go +++ b/pkg/agent/endpoint/registry_test.go @@ -34,8 +34,11 @@ func TestRegistrySupersede(t *testing.T) { require.NoError(t, g.Register(newReg)) require.Equal(t, []*Registration{newReg}, g.Candidates("key", "production")) - require.Error(t, g.ValidateAttach("AW_1", "i-1", "ATT_i-1"), "old epoch must not attach") - require.NoError(t, g.ValidateAttach("AW_1", "i-2", "ATT_i-2")) + require.Error(t, g.ValidateAttach("AW_1", "i-1", "key", "ATT_i-1"), "old epoch must not attach") + require.NoError(t, g.ValidateAttach("AW_1", "i-2", "key", "ATT_i-2")) + // a grant for another project cannot attach even with the right token + require.ErrorIs(t, g.ValidateAttach("AW_1", "i-2", "other", "ATT_i-2"), ErrAttachRejected, + "wrong-project grant must not attach") // the old control connection tears down after the new one registered g.Deregister(oldReg) diff --git a/pkg/service/agentservice.go b/pkg/service/agentservice.go index 91cca5083..8d9ec2419 100644 --- a/pkg/service/agentservice.go +++ b/pkg/service/agentservice.go @@ -144,8 +144,8 @@ type AgentService struct { type AgentHandler struct { agentServer rpc.AgentInternalServer // the server's only configured api key, when exactly one exists: the - // unauthenticated scope for public endpoints regardless of which node holds - // the workers + // unauthenticated identity for public endpoints regardless of which node + // holds the workers singleAPIKey string mu sync.Mutex logger logger.Logger @@ -222,14 +222,14 @@ func NewAgentService( } // EndpointFront is the /agents/{deployment}/{path...} handler backed by this -// node's attached workers. Project scope comes from validated grants when a +// node's attached workers. The api key comes from validated grants when a // token is present; unauthenticated requests reach public endpoints only. func (s *AgentService) EndpointFront() http.Handler { front := endpoint.NewFront(s.endpointRegistry, func(r *http.Request) (string, bool) { if claims := GetGrants(r.Context()); claims != nil { return GetAPIKey(r.Context()), true } - // unauthenticated: with a single configured key the scope is + // unauthenticated: with a single configured key the api key is // unambiguous even when this node holds no registrations (multi-node) return s.singleAPIKey, false }, s.logger) @@ -251,8 +251,10 @@ func (s *AgentService) ServeHTTP(w http.ResponseWriter, r *http.Request) { if attach { // data wire: no registration handshake, no signal loop; the wire // speaks AgentHttp.Frame exclusively and the endpoint mux owns it - // after a successful attach - HandleEndpointAttach(s.endpointRegistry, NewEndpointWireConn(conn), s.wireParams(), nil) + // after a successful attach. The connecting grant's api key binds the + // attach to its own project (a leaked attach token alone cannot bind a + // data conn to another project's worker). + HandleEndpointAttach(s.endpointRegistry, NewEndpointWireConn(conn), s.wireParams(), nil, GetAPIKey(r.Context())) return } @@ -346,14 +348,17 @@ func (s *AgentService) wireParams() endpoint.WireParams { // registry - behind a load balancer the wire may land on a node other than the // one holding the registration. The adopter may install a local registration // (e.g. a satellite fetched from the holder); afterwards the attach is retried -// locally once. nil rejects unknown workers. -type AttachAdopter func(a *livekit.AgentHttp_AttachDataConnection) error +// locally once. apiKey is the identity the connecting grant is authorized for: +// the adopter must refuse to install a registration for any other project, so +// an unentitled grant cannot adopt (or even probe) another project's worker. +// nil rejects unknown workers. +type AttachAdopter func(a *livekit.AgentHttp_AttachDataConnection, apiKey string) error // HandleEndpointAttach adopts a worker-dialed data wire: the first frame is // Attach on stream 0, validated against the registration's epoch and token; the // response carries this node's wire parameters. Shared by OSS and cloud // servers. -func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, params endpoint.WireParams, adopt AttachAdopter) { +func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, params endpoint.WireParams, adopt AttachAdopter, apiKey string) { if err := wire.SetReadDeadline(time.Now().Add(agent.RegisterTimeout)); err != nil { _ = wire.Close() return @@ -389,14 +394,14 @@ func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, p // the slot is reserved before the ack is written (never a success ack for a // wire that then loses the cap race) and adopted only after it, so the // worker cannot observe stream frames ahead of the attach outcome - ticket, err := registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), a.GetAttachToken()) + ticket, err := registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), apiKey, a.GetAttachToken()) if (errors.Is(err, endpoint.ErrUnknownWorker) || errors.Is(err, endpoint.ErrWrongEpoch)) && adopt != nil { - if aerr := adopt(a); aerr != nil { + if aerr := adopt(a, apiKey); aerr != nil { _ = respond(aerr.Error()) _ = wire.Close() return } - ticket, err = registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), a.GetAttachToken()) + ticket, err = registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), apiKey, a.GetAttachToken()) } if err != nil { _ = respond(err.Error()) @@ -516,7 +521,6 @@ func (h *AgentHandler) registerEndpoints(w *agent.Worker) *endpoint.Registration DataConnCount: settings.GetDataConnectionCount(), }, Logger: w.Logger(), - Load: w.Load, Draining: w.Draining, } if err := h.endpointRegistry.Register(reg); err != nil {