Documentation
¶
Index ¶
- Constants
- Variables
- func IsReservedAgentName(agentName string) bool
- func MergeWorkerHeaders(dst, worker http.Header)
- func NegotiateSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error)
- func NormalizeDeployment(d string) string
- type Access
- type AccessLevel
- type AccessResolver
- type Endpoint
- type Fallback
- type Front
- type FrontParams
- type Identity
- type Manifest
- type Registration
- type RegistrationParams
- type Registry
- type Route
- type Scope
- type Session
- type Stream
- type StreamResetError
- type Tunnels
Constants ¶
const ( // PathPrefix is the public route namespace: /agents/{agent_name}/{deployment}/{path...} PathPrefix = "/agents/" // MaxPathLength caps the escaped route path. Matching runs before the // request head is sized, so this is the only bound on it. MaxPathLength = 8 << 10 )
const DefaultDeployment = "default"
DefaultDeployment is the URL segment that addresses workers registered with an empty deployment name (self-hosted workers typically set none).
const DefaultMaxStreams = 256
DefaultMaxStreams is the soft per-session concurrency cap used only for capacity weighting; QUIC's own stream limit is the hard bound.
const ( // DefaultTunnelDrainTimeout is how long a WebSocket tunnel outlives its // worker starting to drain. DefaultTunnelDrainTimeout = 30 * time.Second )
const MaxManifestRoutes = 256
const UnnamedAgentSegment = "_"
UnnamedAgentSegment is the URL segment that addresses workers registered with an empty agent name.
Variables ¶
var ErrNoSession = errors.New("registration has no data-plane session")
ErrNoSession means the registration has no live data-plane session, so no stream can be opened toward the worker.
Functions ¶
func IsReservedAgentName ¶
IsReservedAgentName reports whether a name cannot address a worker. These are unreserved characters (RFC 3986 §2.3), so percent-encoding them yields no distinct form to address them by.
func MergeWorkerHeaders ¶
MergeWorkerHeaders adds a worker's response head to dst, which may hold headers set by serving middleware. A worker's Access-Control-* and Vary values replace the middleware's. Preflights never reach the worker: they carry no credential to admit them to a private route.
func NegotiateSettings ¶
func NegotiateSettings(req *livekit.RegisterWorkerRequest) (*livekit.AgentHttp_AgentEndpointSettings, error)
NegotiateSettings validates a registration's endpoint manifest and negotiates the data-plane protocol version. The data plane is a WebTransport session (no attach token, no fixed connection pool), so the version is all there is to agree on.
func NormalizeDeployment ¶
NormalizeDeployment maps an empty deployment to the segment that addresses it in a URL. Every embedder must key its scopes through this, or a worker that registered without a deployment is unreachable from any node but its own.
Types ¶
type Access ¶
type Access struct {
// Scope is the deployment's serving state this request is placed against.
// nil means no worker here holds it; Fallback may still place it elsewhere.
Scope *Scope
// Fallback serves the request elsewhere (e.g. a multi-node relay), already
// curried on the deployment it resolved. nil means local misses are final.
Fallback Fallback
Level AccessLevel
}
Access is what the front knows about a request's caller, together with the serving state it resolved to. The front never keys anything itself: whatever scopes a request - a tenant, an api key, a project - is resolved by the embedder and arrives here already looked up.
type AccessLevel ¶
type AccessLevel int
AccessLevel is how far a request's caller is trusted. Callers compare against it, so a new level must be inserted at its correct rank.
const ( // AccessNone presented no credential. AccessNone AccessLevel = iota // AccessCredentialed presented a valid token carrying no agent-endpoint // grant for the addressed agent and deployment. AccessCredentialed // AccessGranted presented a token whose agent-endpoint grant covers the // addressed agent and deployment. AccessGranted )
func (AccessLevel) String ¶
func (a AccessLevel) String() string
type AccessResolver ¶
AccessResolver maps an inbound request, plus the agent and deployment its URL addresses, to the caller's access. ok is false when the request cannot be placed at all - no credential, or an unknown tenant - and the front challenges.
type Endpoint ¶
Endpoint is one validated manifest entry. Mask carries the methods exactly as declared, so it may name several.
type Fallback ¶
type Fallback func(w http.ResponseWriter, r *http.Request, level AccessLevel) bool
Fallback serves a request elsewhere (e.g. a multi-node relay); it reports whether a response was written. Returning false falls back to the local status mapping. It is curried on the deployment it was resolved for, so it carries no scope arguments.
type Front ¶
type Front struct {
// contains filtered or unexported fields
}
func NewFront ¶
func NewFront(params FrontParams) *Front
func (*Front) Drain ¶
Drain refuses new WebSocket upgrades, lets open tunnels finish until ctx is done, then closes the rest with 1001 (going away). It returns once no tunnel remains and is safe to call repeatedly.
func (*Front) OpenTunnels ¶
OpenTunnels reports the WebSocket tunnels this front is splicing.
type FrontParams ¶
type FrontParams struct {
ResolveAccess AccessResolver
Logger logger.Logger
Identity Identity
// MaxTunnelsPerWorker caps one worker's concurrent WebSocket tunnels. 0
// means half its session's MaxStreams.
MaxTunnelsPerWorker int
// TunnelDrainTimeout is how long a tunnel outlives its worker starting to
// drain. 0 means DefaultTunnelDrainTimeout.
TunnelDrainTimeout time.Duration
}
FrontParams configures a Front. Fields are read on every request once the Front is serving, so none may change after construction.
type Identity ¶
Identity resolves the agent and deployment a request addresses. Reporting false leaves them to the URL.
type Manifest ¶
type Manifest struct {
Endpoints []Endpoint
// contains filtered or unexported fields
}
Manifest is one worker's validated route table, in declaration order.
func ParseManifest ¶
func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, error)
ParseManifest validates a registration's endpoint list.
func (*Manifest) Ambiguous ¶
Ambiguous returns the declared templates whose shape forces the matcher to backtrack. The result is read-only.
func (*Manifest) MarshalLogObject ¶
func (m *Manifest) MarshalLogObject(e zapcore.ObjectEncoder) error
MarshalLogObject describes the manifest's shape.
type Registration ¶
type Registration struct {
WorkerID string
Manifest *Manifest
// contains filtered or unexported fields
}
Registration is one worker's data-plane state: its manifest and the single WebTransport session that carries both its control stream and the HTTP exchanges the node opens toward it. It lives exactly as long as that session (epoch fencing: a reconnecting worker forms a new registration, and the old session dies with it).
It carries no agent name, deployment or tenant identity: those name the Scope it was registered into, which is the embedder's to key.
func NewRegistration ¶
func NewRegistration(params RegistrationParams) *Registration
func (*Registration) HasSession ¶
func (r *Registration) HasSession() bool
HasSession reports whether the worker has a live data-plane session, i.e. can serve HTTP requests right now.
func (*Registration) InflightStreams ¶
func (r *Registration) InflightStreams() int
InflightStreams reports the open streams on this worker's session - the least-outstanding-requests signal for worker selection.
func (*Registration) IsDraining ¶
func (r *Registration) IsDraining() bool
IsDraining is false when no drain signal was supplied.
func (*Registration) OpenStream ¶
func (r *Registration) OpenStream(ctx context.Context) (Stream, error)
OpenStream opens a data-plane stream toward the worker for one HTTP exchange.
func (*Registration) SpareStreams ¶
func (r *Registration) SpareStreams() int
SpareStreams reports remaining stream capacity - the node's live serving headroom for the worker, used to weight node selection.
func (*Registration) Tunnels ¶
func (r *Registration) Tunnels() int
Tunnels reports the WebSocket tunnels (and handshakes) holding a stream on this worker.
type RegistrationParams ¶
type RegistrationParams struct {
WorkerID string
Manifest *Manifest
// Session is the worker's live data-plane session: the WebTransport session
// that also carries its control stream. One session per worker.
Session Session
// Draining reports that the worker is shedding; a shedding worker takes no
// new streams.
Draining func() bool
}
RegistrationParams is fixed for the life of the registration, which lasts exactly as long as the session.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry fences worker epochs on this node. Worker ids are server-issued and unique across the node, so this index is tenancy-blind: it exists only so a reconnecting worker supersedes its own stale epoch, wherever that epoch was scoped.
It holds the outermost lock in this package: a Scope's lock may be taken while holding it, never the reverse.
func NewRegistry ¶
func NewRegistry() *Registry
func (*Registry) Deregister ¶
func (g *Registry) Deregister(r *Registration)
Deregister removes exactly this registration; it is a no-op when a newer epoch has already superseded it.
func (*Registry) Register ¶
func (g *Registry) Register(scope *Scope, r *Registration)
Register records a registration in a scope. A worker id already present is superseded: worker ids are stable across reconnects, and the retiring session must not be able to strand the new epoch (its own Deregister is a no-op once replaced). The superseded epoch's session is closed.
type Route ¶
type Route struct {
Template *router.Template
Mask router.Mask
// contains filtered or unexported fields
}
Route is one endpoint of a deployment and every worker that declared it. The compiled table holds these by pointer: a route outlives the tables built around it, and its worker set changes without a rebuild.
type Scope ¶
type Scope struct {
// contains filtered or unexported fields
}
Scope is one deployment's serving state on this node: the registrations that hold it and their merged route table. Who a scope belongs to is the embedder's business - nothing here stores a tenancy identity, and the embedder owns the map that scopes it.
Registrations and routes move together under lock, so a worker present in the candidate set always has its routes installed.
func NewScope ¶
NewScope takes a logger the embedder has already curried with whatever identifies the scope; the package adds no identity of its own.
func (*Scope) Candidates ¶
func (s *Scope) Candidates() []*Registration
Candidates returns the registrations holding this scope. A nil scope holds none, so a caller that could not resolve one needs no separate test.
type Session ¶
type Session interface {
// OpenStream opens a bidirectional stream toward the worker for one HTTP
// exchange. It blocks until a stream can be opened (QUIC stream-limit
// backpressure) or ctx is done.
OpenStream(ctx context.Context) (Stream, error)
// OpenStreams reports the streams currently open on this session - the
// least-outstanding-requests signal for worker selection.
OpenStreams() int
// MaxStreams is the soft cap on concurrent streams, used to derive spare
// capacity for node-selection weighting.
MaxStreams() int
// Close tears the session down.
Close(reason string)
}
Session is the node's handle to one worker's data plane: a single WebTransport (QUIC) session that also carries the worker's control stream. The node opens a fresh Stream per HTTP exchange; QUIC provides the multiplexing and per-stream flow control, so there is no capsule layer.
func NewWebTransportSession ¶
func NewWebTransportSession(sess *webtransport.Session, maxStreams int) Session
NewWebTransportSession wraps a live WebTransport session as a data-plane Session. maxStreams is the soft concurrency cap used only for capacity weighting; QUIC's own stream limit is the hard bound.
type Stream ¶
type Stream interface {
io.Reader
io.Writer
// CloseWrite half-closes the send side (QUIC stream FIN). A FIN arriving
// before the body framing says the body is complete is truncation.
CloseWrite() error
// Reset aborts the stream in both directions (QUIC RESET_STREAM /
// STOP_SENDING). The code reaches the peer and is the only outcome signal
// available once no bytes can flow; the reason is local only.
Reset(code livekit.AgentHttp_HttpStreamResetCode, reason string)
// Close releases the stream after a completed exchange.
Close() error
// SetReadDeadline bounds the wait for more bytes.
SetReadDeadline(t time.Time) error
}
Stream is one HTTP exchange over a Session: a preamble, then opaque HTTP/1.1 bytes in each direction.
type StreamResetError ¶
type StreamResetError struct {
Code livekit.AgentHttp_HttpStreamResetCode
}
StreamResetError reports that the peer reset the stream, carrying the code it sent. Transports translate their own reset errors into this at the Stream boundary.
func (*StreamResetError) Error ¶
func (e *StreamResetError) Error() string
type Tunnels ¶
type Tunnels struct {
// contains filtered or unexported fields
}
Tunnels tracks hijacked connections so a drain can close them.
func NewTunnels ¶
func NewTunnels() *Tunnels
func (*Tunnels) Add ¶
Add tracks c. It reports false once a drain has begun: c is tracked regardless, so the drain still waits for it, but the caller must close it.
func (*Tunnels) Drain ¶
Drain lets the tracked connections end until ctx is done, then closes the rest. It returns once none remain and is safe to call repeatedly.
func (*Tunnels) Draining ¶
func (s *Tunnels) Draining() <-chan struct{}
Draining is closed once a drain has begun.
func (*Tunnels) IsDraining ¶
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package router matches request paths against an ordered table of starlette-style path templates.
|
Package router matches request paths against an ordered table of starlette-style path templates. |