requestcontrol

package
v0.9.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 23, 2026 License: Apache-2.0 Imports: 37 Imported by: 0

Documentation

Overview

Package requestcontrol defines the Director component responsible for orchestrating request processing after initial parsing.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AdmissionController 

type AdmissionController interface {
	// Admit determines if a request should be admitted.
	// It is called by the Director for each incoming request.
	//
	// Args:
	//   ctx: The request context, carrying deadlines, cancellation signals, and logger.
	//   reqCtx: The handlers.RequestContext containing details about the incoming request.
	//   priority: The priority level of the request, as determined by the InferenceObjective.
	//
	// Returns:
	//   - nil: If the request is admitted and should proceed to scheduling.
	//   - errcommon.Error: If the request is rejected.
	Admit(
		ctx context.Context,
		reqCtx *handlers.RequestContext,
		priority int,
	) error
}

AdmissionController defines the interface for making admission control decisions. Implementations of this interface determine whether an incoming inference request should be accepted or rejected based on various criteria such as system load, fairness, priority, and available capacity.

type CachedEndpointCandidates 

type CachedEndpointCandidates struct {
	// contains filtered or unexported fields
}

CachedEndpointCandidates is a decorator for contracts.EndpointCandidates that caches results to reduce lock contention on the underlying Datastore.

It is designed for high-throughput paths (like the Flow Control dispatch loop)cwhere fetching fresh data every millisecond is unnecessary and expensive.

func NewCachedEndpointCandidates 

func NewCachedEndpointCandidates(ctx context.Context, delegate contracts.EndpointCandidates, ttl time.Duration) *CachedEndpointCandidates

NewCachedEndpointCandidates creates a new CachedEndpointCandidates and starts a background cleanup routine. The provided context is used to control the lifecycle of the cleanup goroutine.

func (*CachedEndpointCandidates) Locate 

func (c *CachedEndpointCandidates) Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint

Locate returns the list of endpoint candidates for the given request metadata, using a cached result if available and fresh.

type Config 

type Config struct {
	// contains filtered or unexported fields
}

Config provides a configuration for the requestcontrol plugins.

func NewConfig 

func NewConfig() *Config

NewConfig creates a new Config object and returns its pointer.

func (*Config) AddPlugins 

func (c *Config) AddPlugins(pluginObjects ...plugin.Plugin)

AddPlugins adds the given plugins to the Config. The type of each plugin is checked and added to the corresponding list of plugins in the Config. If a plugin implements multiple plugin interfaces, it will be added to each corresponding list.

func (*Config) OrderDataProducerPlugins 

func (c *Config) OrderDataProducerPlugins(sortedPluginNames []string)

OrderDataProducerPlugins reorders the DataProducer plugins in the Config based on the given sorted plugin names.

func (*Config) WithAdmissionPlugins 

func (c *Config) WithAdmissionPlugins(plugins ...fwkrc.Admitter) *Config

WithAdmissionPlugins sets the given plugins as the Admit plugins.

func (*Config) WithDataProducerPlugins 

func (c *Config) WithDataProducerPlugins(plugins ...fwkrc.DataProducer) *Config

WithDataProducerPlugins sets the given plugins as the DataProducer plugins.

func (*Config) WithPreAdmissionPlugins 

func (c *Config) WithPreAdmissionPlugins(plugins ...fwkrc.PreAdmitter) *Config

WithPreAdmissionPlugins sets the given plugins as the PreAdmitter plugins.

func (*Config) WithPreRequestPlugins 

func (c *Config) WithPreRequestPlugins(plugins ...fwkrc.PreRequest) *Config

WithPreRequestPlugins sets the given plugins as the PreRequest plugins. If the Config has PreRequest plugins already, this call replaces the existing plugins with the given ones.

func (*Config) WithResponseReceivedPlugins 

func (c *Config) WithResponseReceivedPlugins(plugins ...fwkrc.ResponseHeaderProcessor) *Config

WithResponseReceivedPlugins sets the given plugins as the ResponseReceived plugins. If the Config has ResponseReceived plugins already, this call replaces the existing plugins with the given ones.

func (*Config) WithResponseStreamingPlugins 

func (c *Config) WithResponseStreamingPlugins(plugins ...fwkrc.ResponseBodyProcessor) *Config

WithResponseStreamingPlugins sets the given plugins as the ResponseStreaming plugins. If the Config has ResponseStreaming plugins already, this call replaces the existing plugins with the given ones.

type Datastore 

type Datastore interface {
	PoolGet() (*datalayer.EndpointPool, error)
	ObjectiveGet(objectiveName string) *v1alpha2.InferenceObjective
	PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoint
	// ModelRewriteGet returns the highest-precedence rewrite rule for a given
	// model name (prioritizing exact matches over generic wildcard rules) and
	// the name of the InferenceModelRewrite object.
	ModelRewriteGet(modelName string) (*v1alpha2.InferenceModelRewriteRule, string)
}

Datastore defines the interface required by the Director.

type DatastoreEndpointCandidates 

type DatastoreEndpointCandidates struct {
	// contains filtered or unexported fields
}

DatastoreEndpointCandidates implements contracts.EndpointCandidates by querying the EPP Datastore. It centralizes the logic for resolving endpoint candidates based on request metadata (specifically Envoy subset filters).

func NewDatastoreEndpointCandidates 

func NewDatastoreEndpointCandidates(ds Datastore, opts ...EndpointCandidatesOption) *DatastoreEndpointCandidates

NewDatastoreEndpointCandidates creates a new DatastoreEndpointCandidates.

func (*DatastoreEndpointCandidates) Locate 

func (d *DatastoreEndpointCandidates) Locate(ctx context.Context, requestMetadata map[string]any) []fwkdl.Endpoint

Locate retrieves the list of endpoint candidates from the datastore that match the criteria defined in the request metadata.

It supports: 1. Returning all endpoint candidates if no specific subset filter is present. 2. Returning a filtered list of endpoint candidates if "x-gateway-destination-endpoint-subset" is present.

type Director 

type Director struct {
	// contains filtered or unexported fields
}

Director orchestrates the request handling flow after initial parsing by the handler. Its responsibilities include: - Retrieving request metadata and relevant objectives. - Determining candidate pods. - Performing admission control via the AdmissionController. - Scheduling the request to target pod(s) via the Scheduler. - Running PreRequest plugins. - Preparing the request context for the Envoy ext_proc filter to route the request. - Running PostResponse plugins.

func NewDirectorWithConfig 

func NewDirectorWithConfig(
	datastore Datastore,
	scheduler Scheduler,
	admissionController AdmissionController,
	endpointCandidates contracts.EndpointCandidates,
	config *Config,
) *Director

NewDirectorWithConfig creates a new Director instance with all dependencies.

func (*Director) GetRandomEndpoint 

func (d *Director) GetRandomEndpoint() *fwkdl.EndpointMetadata

func (*Director) HandleRequest 

func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestContext, inferenceRequestBody *fwkrh.InferenceRequestBody) (_ *handlers.RequestContext, err error)

HandleRequest orchestrates the request lifecycle. It always returns the requestContext even in the error case, as the request context is used in error handling.

func (*Director) HandleResponseBody 

func (d *Director) HandleResponseBody(ctx context.Context, reqCtx *handlers.RequestContext, endOfStream bool) *handlers.RequestContext

HandleResponseBody is invoked by the director for every chunk received in a streaming response, or exactly once for a non-streaming response.

For intermediate streaming chunks (endOfStream=false), the work is sent to a per-request async queue (channel + goroutine) so plugins run off the critical path while preserving chunk ordering. For the final chunk (endOfStream=true), the queue is drained first, then plugins run synchronously because they may produce DynamicMetadata that must be attached to the ext_proc response sent back to Envoy.

func (*Director) HandleResponseHeader 

func (d *Director) HandleResponseHeader(ctx context.Context, reqCtx *handlers.RequestContext) *handlers.RequestContext

HandleResponseHeader is called when the response headers are received.

type EndpointCandidatesConfig 

type EndpointCandidatesConfig struct {
	DisableEndpointSubsetFilter bool
}

EndpointCandidatesConfig holds configuration for the DatastoreEndpointCandidates.

type EndpointCandidatesOption 

type EndpointCandidatesOption func(*EndpointCandidatesConfig)

EndpointCandidatesOption is a function that configures the EndpointCandidatesConfig.

func WithDisableEndpointSubsetFilter 

func WithDisableEndpointSubsetFilter(disable bool) EndpointCandidatesOption

WithDisableEndpointSubsetFilter sets the DisableEndpointSubsetFilter flag.

type FlowControlAdmissionController 

type FlowControlAdmissionController struct {
	// contains filtered or unexported fields
}

FlowControlAdmissionController delegates admission decisions to the Flow Control layer. It uses the provided Flow Controller to enqueue the request and await an outcome.

func NewFlowControlAdmissionController 

func NewFlowControlAdmissionController(fc flowController, poolName string) *FlowControlAdmissionController

NewFlowControlAdmissionController creates a new FlowControlAdmissionController.

func (*FlowControlAdmissionController) Admit 

func (fcac *FlowControlAdmissionController) Admit(
	ctx context.Context,
	reqCtx *handlers.RequestContext,
	priority int,
) error

Admit implements the AdmissionController interface by checking for saturation on sheddable requests first, then deferring to the Flow Control system.

type LegacyAdmissionController 

type LegacyAdmissionController struct {
	// contains filtered or unexported fields
}

LegacyAdmissionController implements saturation-based admission control. It rejects sheddable requests (priority < 0) if the saturationDetector indicates that the system is currently saturated. Non-sheddable requests always bypass the saturation check.

func NewLegacyAdmissionController 

func NewLegacyAdmissionController(
	sd flowcontrol.SaturationDetector,
	endpointCandidates contracts.EndpointCandidates,
) *LegacyAdmissionController

NewLegacyAdmissionController creates a new LegacyAdmissionController.

func (*LegacyAdmissionController) Admit 

func (lac *LegacyAdmissionController) Admit(
	ctx context.Context,
	reqCtx *handlers.RequestContext,
	priority int,
) error

Admit implements the AdmissionController interface for the legacy strategy. It checks for saturation only for requests with priority < 0.

type Scheduler 

type Scheduler interface {
	Schedule(ctx context.Context, request *fwksched.InferenceRequest, candidateEndpoints []fwksched.Endpoint) (result *fwksched.SchedulingResult, err error)
}

Scheduler defines the interface required by the Director for scheduling.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.