Documentation
¶
Overview ¶
Package controller contains the implementation of the FlowController engine.
The FlowController is the central processing engine of the Flow Control layer. It is a high-throughput component responsible for managing the lifecycle of all incoming requests. It achieves this by acting as a stateless supervisor that orchestrates a stateful worker (Processor).
Package controller contains the implementation of the FlowController engine.
Overview ¶
The FlowController is the central processing engine of the Flow Control layer. It acts as a stateless supervisor that orchestrates a pool of stateful workers (internal.ShardProcessor), managing the lifecycle of all incoming requests from initial submission to a terminal outcome (dispatch, rejection, or eviction).
Architecture: Supervisor-Worker Pattern ¶
This package implements a supervisor-worker pattern to achieve high throughput and dynamic scalability.
- The FlowController (Supervisor): The public-facing API of the system. Its primary responsibilities are to execute a distribution algorithm to select the optimal worker for a new request and to manage the lifecycle of the worker pool, ensuring it stays synchronized with the underlying shard topology defined by the contracts.FlowRegistry.
- The internal.ShardProcessor (Worker): A stateful, single-goroutine actor responsible for the entire lifecycle of requests on a single shard. The supervisor manages a pool of these workers, one for each contracts.RegistryShard.
Concurrency Model ¶
The FlowController is designed to be highly concurrent and thread-safe. It acts primarily as a stateless distributor.
- EnqueueAndWait: Can be called concurrently by many goroutines.
- Worker Management: Uses a sync.Map (workers) for concurrent access and lazy initialization of workers.
- Supervision: A single background goroutine (run) manages the worker pool lifecycle (garbage collection).
It achieves high throughput by minimizing shared state and relying on the internal ShardProcessors to handle state mutations serially (using an actor model).
Request Lifecycle and Ownership ¶
A request (represented internally as a FlowItem) has a lifecycle managed cooperatively by the Controller and a Processor. Defining ownership is critical for ensuring an item is finalized exactly once.
- Submission (Controller): The Controller attempts to hand off the item to a Processor.
- Handoff: - Success: Ownership transfers to the Processor, which is now responsible for Finalization. - Failure: Ownership remains with the Controller, which must Finalize the item.
- Processing (Processor): The Processor enqueues, manages, and eventually dispatches or rejects the item.
- Finalization: The terminal outcome is set. This can happen: - Synchronously: The Processor determines the outcome (e.g., Dispatch, Capacity Rejection). - Asynchronously: The Controller observes the request's Context expiry (TTL/Cancellation) and calls Finalize.
The FlowItem uses atomic operations to safely coordinate the Finalization state across goroutines.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// DefaultRequestTTL is the default Time-To-Live applied to requests that do not
// specify their own TTL hint.
// Optional: If zero, no TTL is applied by default and we rely solely on request context cancellation.
DefaultRequestTTL time.Duration
// ExpiryCleanupInterval is the interval at which each processor scans its queues for expired items.
// Optional: Defaults to `defaultExpiryCleanupInterval` (1 second).
ExpiryCleanupInterval time.Duration
// EnqueueChannelBufferSize is the size of the buffered channel that accepts incoming requests for each
// processor. This buffer acts as a shock absorber, decoupling the high-frequency distributor from the processor's
// serial execution loop and allowing the system to handle short bursts of traffic without blocking.
// Optional: Defaults to `defaultEnqueueChannelBufferSize` (100).
EnqueueChannelBufferSize int
}
Config holds the configuration for the `FlowController`.
func NewConfig ¶
func NewConfig(opts ...ConfigOption) (*Config, error)
NewConfig creates a new Config with the given options, applying defaults and validation.
func NewConfigFromAPI ¶
func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error)
NewConfigFromAPI creates a new Config from the API configuration.
type ConfigOption ¶
type ConfigOption func(*Config)
ConfigOption is a functional option for configuring the FlowController.
func WithDefaultRequestTTL ¶
func WithDefaultRequestTTL(d time.Duration) ConfigOption
WithDefaultRequestTTL sets the default request TTL.
func WithEnqueueChannelBufferSize ¶
func WithEnqueueChannelBufferSize(size int) ConfigOption
WithEnqueueChannelBufferSize sets the size of the enqueue channel buffer.
func WithExpiryCleanupInterval ¶
func WithExpiryCleanupInterval(d time.Duration) ConfigOption
WithExpiryCleanupInterval sets the expiry cleanup interval.
type Deps ¶
type Deps struct {
Registry contracts.FlowRegistry
SaturationDetector flowcontrol.SaturationDetector
EndpointCandidates contracts.EndpointCandidates
UsageLimitPolicy flowcontrol.UsageLimitPolicy
Clock clock.WithTicker
ProcessorFactory processorFactory
}
Deps groups the external FlowController build dependencies to construct a FlowController.
type FlowController ¶
type FlowController struct {
// contains filtered or unexported fields
}
FlowController is the central, high-throughput engine of the Flow Control layer. It is designed as a stateless distributor that orchestrates a stateful worker (Processor), following a supervisor-worker pattern.
Request Lifecycle Management:
- Asynchronous Finalization (Controller-Owned): The Controller actively monitors the request Context (TTL/Cancellation) in EnqueueAndWait. If the Context expires, the Controller immediately Finalizes the item and unblocks the caller.
- Synchronous Finalization (Processor-Owned): The Processor handles Dispatch, Capacity Rejection, and Shutdown.
- Cleanup (Processor-Owned): The Processor periodically sweeps externally finalized items to reclaim capacity.
func NewFlowController ¶
func NewFlowController( ctx context.Context, poolName string, config *Config, deps Deps, ) *FlowController
NewFlowController creates and starts a new FlowController instance. The provided context governs the lifecycle of the controller and all its workers.
func (*FlowController) EnqueueAndWait ¶
func (fc *FlowController) EnqueueAndWait( ctx context.Context, req flowcontrol.FlowControlRequest, ) (types.QueueOutcome, error)
EnqueueAndWait is the primary, synchronous entry point to the Flow Control system. It submits a request and blocks until the request reaches a terminal outcome (dispatched, rejected, or evicted).
Design Rationale: The Synchronous Model ¶
This blocking model is deliberately chosen for its simplicity and robustness, especially in the context of Envoy External Processing (ext_proc), which operates on a stream-based protocol.
- ext_proc Alignment: A single goroutine typically manages the stream for a given HTTP request. EnqueueAndWait fits this perfectly: the request-handling goroutine calls it, blocks, and upon return, has a definitive outcome to act upon.
- Simplified State Management: The state of a "waiting" request is implicitly managed by the blocked goroutine's stack and its Context. The system only needs to signal this specific goroutine to unblock it.
- Direct Backpressure: If queues are full, EnqueueAndWait returns an error immediately, providing direct backpressure to the caller.
Source Files
Directories