director.go

/*
Copyright 2025 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

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

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand"
	"net"
	"strings"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/llm-d/llm-d-router/apix/v1alpha2"
	errcommon "github.com/llm-d/llm-d-router/pkg/common/error"
	logutil "github.com/llm-d/llm-d-router/pkg/common/observability/logging"
	"github.com/llm-d/llm-d-router/pkg/common/observability/tracing"
	reqcommon "github.com/llm-d/llm-d-router/pkg/common/request"
	"github.com/llm-d/llm-d-router/pkg/common/routing"
	"github.com/llm-d/llm-d-router/pkg/epp/datalayer"
	"github.com/llm-d/llm-d-router/pkg/epp/datastore"
	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
	fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol"
	fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling"
	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
	"github.com/llm-d/llm-d-router/pkg/epp/handlers"
	"github.com/llm-d/llm-d-router/pkg/epp/metadata"
	"github.com/llm-d/llm-d-router/pkg/epp/metrics"
)

const (
	// dataProducerTimeout is the default per-producer execution timeout. A
	// producer overrides it by implementing requestcontrol.TimeoutAwareProducer.
	dataProducerTimeout       = 400 * time.Millisecond
	responseBodyQueueCapacity = 100
)

// primaryDecodeEndpoint returns the first endpoint chosen by the primary
// profile, or nil when the result is empty or malformed. False-return reasons
// are logged at V(logutil.DEBUG) to disambiguate misconfiguration from a real
// cache miss.
func primaryDecodeEndpoint(logger logr.Logger, result *fwksched.SchedulingResult) fwksched.Endpoint {
	debug := logger.V(logutil.DEBUG)
	if result == nil {
		debug.Info("conditional-decode: scheduling result is nil")
		return nil
	}
	primary, ok := result.ProfileResults[result.PrimaryProfileName]
	if !ok || primary == nil {
		debug.Info("conditional-decode: primary profile result missing", "primary", result.PrimaryProfileName)
		return nil
	}
	if len(primary.TargetEndpoints) == 0 {
		debug.Info("conditional-decode: primary profile produced no endpoints", "primary", result.PrimaryProfileName)
		return nil
	}
	return primary.TargetEndpoints[0]
}

// Datastore defines the interface required by the Director.
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)
}

// Scheduler defines the interface required by the Director for scheduling.
type Scheduler interface {
	Schedule(ctx context.Context, request *fwksched.InferenceRequest, candidateEndpoints []fwksched.Endpoint) (result *fwksched.SchedulingResult, err error)
}

// NewDirectorWithConfig creates a new Director instance with all dependencies.
func NewDirectorWithConfig(
	datastore Datastore,
	scheduler Scheduler,
	admissionController AdmissionController,
	endpointCandidates contracts.EndpointCandidates,
	config *Config,
) *Director {
	return &Director{
		datastore:             datastore,
		scheduler:             scheduler,
		admissionController:   admissionController,
		endpointCandidates:    endpointCandidates,
		requestControlPlugins: *config,
		defaultPriority:       0, // define default priority explicitly
	}
}

// responseBodyWork represents a unit of work to be processed by the async response body queue.
type responseBodyWork struct {
	ctx            context.Context
	request        *fwksched.InferenceRequest
	response       *fwkrc.Response
	targetEndpoint *fwkdl.EndpointMetadata
}

// responseBodyQueue is a per-request async queue for processing response body plugin calls.
// It ensures chunks are processed in order via a channel while keeping plugin execution
// off the critical streaming path.
type responseBodyQueue struct {
	ch     chan responseBodyWork
	done   chan struct{} // closed when the processing goroutine exits
	mu     sync.Mutex
	closed bool
}

func newResponseBodyQueue() *responseBodyQueue {
	return &responseBodyQueue{
		ch:   make(chan responseBodyWork, responseBodyQueueCapacity),
		done: make(chan struct{}),
	}
}

func (q *responseBodyQueue) enqueue(work responseBodyWork) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.closed {
		return false
	}
	q.ch <- work
	return true
}

func (q *responseBodyQueue) closeAndWait() {
	q.mu.Lock()
	if !q.closed {
		q.closed = true
		close(q.ch)
	}
	q.mu.Unlock()
	<-q.done
}

// 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.
type Director struct {
	datastore             Datastore
	scheduler             Scheduler
	admissionController   AdmissionController
	endpointCandidates    contracts.EndpointCandidates
	requestControlPlugins Config
	// We just need a pointer to an int32 variable since Priority is a pointer in InferenceObjective.
	// No need to set this in the constructor, since the value we want is the default (0)
	// and value types cannot be nil.
	defaultPriority int32

	// responseBodyQueues maps request contexts to their async processing channels.
	// Each request gets a dedicated channel and goroutine to ensure chunks are processed in order while not blocking the
	// streaming response path. The request context key avoids coupling independent streams that reuse the same
	// x-request-id header.
	responseBodyQueues sync.Map
}

// getInferenceObjective fetches the inferenceObjective from the datastore otherwise creates a new one based on reqCtx.
func (d *Director) getInferenceObjective(ctx context.Context, reqCtx *handlers.RequestContext) *v1alpha2.InferenceObjective {
	infObjective := d.datastore.ObjectiveGet(reqCtx.ObjectiveKey)
	if infObjective == nil {
		log.FromContext(ctx).V(logutil.VERBOSE).Info("No associated InferenceObjective found, using default", "objectiveKey", reqCtx.ObjectiveKey)
		infObjective = &v1alpha2.InferenceObjective{
			Spec: v1alpha2.InferenceObjectiveSpec{
				Priority: &d.defaultPriority,
			},
		}
	} else if infObjective.Spec.Priority == nil {
		// Default to 0 if not specified.
		infObjective.Spec.Priority = &d.defaultPriority
	}
	return infObjective
}

// 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 (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestContext, inferenceRequestBody *fwkrh.InferenceRequestBody) (_ *handlers.RequestContext, err error) {
	tracer := tracing.Tracer("llm-d-router/pkg/epp/requestcontrol")
	ctx, span := tracer.Start(ctx, "gateway.request_orchestration", trace.WithSpanKind(trace.SpanKindServer))
	defer func() {
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
		}
		span.End()
	}()

	logger := log.FromContext(ctx)

	err = d.modelRewriteIfNeeded(ctx, reqCtx, inferenceRequestBody)
	if err != nil {
		return reqCtx, err
	}

	infObjective := d.getInferenceObjective(ctx, reqCtx)
	priority := int(*infObjective.Spec.Priority)
	reqCtx.Priority = priority
	requestObjectives := fwksched.RequestObjectives{Priority: priority}

	span.SetAttributes(
		attribute.String("target_model", reqCtx.TargetModelName),
		attribute.Int("request_prio", priority),
	)

	fairnessID, _ := metadata.GetLowerCaseHeaderValue(reqCtx.Request.Headers, metadata.FlowFairnessIDKey)

	// Prepare InferenceRequest (needed for both saturation detection and Scheduler)
	reqCtx.SchedulingRequest = &fwksched.InferenceRequest{
		RequestID:        reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey],
		TargetModel:      reqCtx.TargetModelName,
		Body:             inferenceRequestBody,
		Headers:          reqCtx.Request.Headers,
		FairnessID:       fairnessID,
		Objectives:       requestObjectives,
		RequestSizeBytes: reqCtx.RequestSize,
	}

	logger = logger.WithValues("objectiveKey", reqCtx.ObjectiveKey, "incomingModelName", reqCtx.IncomingModelName, "targetModelName", reqCtx.TargetModelName, "priority", infObjective.Spec.Priority)
	ctx = log.IntoContext(ctx, logger)
	logger.V(logutil.DEBUG).Info("LLM request assembled")

	if err := d.runPreAdmissionPlugins(ctx, reqCtx.SchedulingRequest); err != nil {
		return reqCtx, err
	}
	if reqCtx.SchedulingRequest.FairnessID == "" {
		reqCtx.SchedulingRequest.FairnessID = metadata.DefaultFairnessID
	}

	// Admit may block until flow control admits the request.
	if err := d.admissionController.Admit(ctx, reqCtx, priority); err != nil {
		return reqCtx, err
	}

	endpointCandidates := d.endpointCandidates.Locate(ctx, reqCtx.Request.Metadata)
	if len(endpointCandidates) == 0 {
		return reqCtx, errcommon.Error{
			Code: errcommon.ServiceUnavailable,
			Msg:  "failed to find endpoint candidates for serving the request",
		}
	}

	snapshotOfCandidatePods := d.toSchedulerEndpoints(endpointCandidates)
	// Prepare per request data by running DataProducer plugins.
	err = d.runDataProducerPlugins(ctx, reqCtx.SchedulingRequest, snapshotOfCandidatePods)
	if err != nil {
		// Don't fail the request if DataProducer plugins fail.
		logger.Error(err, "failed to prepare per request data")
	}

	// Run admit request plugins
	if denyReason := d.runAdmissionPlugins(ctx, reqCtx.SchedulingRequest, snapshotOfCandidatePods); denyReason != nil {
		return reqCtx, errcommon.Error{Code: errcommon.Internal, Msg: fmt.Errorf("request cannot be admitted: %w", denyReason).Error()}
	}

	result, err := d.scheduler.Schedule(ctx, reqCtx.SchedulingRequest, snapshotOfCandidatePods)
	if err != nil {
		// Preserve typed errcommon.Error from the scheduler so its status code
		// (e.g. PreconditionFailed) reaches Envoy intact, even if the error
		// has been wrapped (fmt.Errorf("...: %w", err)) on its way up. Other
		// errors fall through to ResourceExhausted, the legacy "no endpoint"
		// status.
		var e errcommon.Error
		if errors.As(err, &e) {
			return reqCtx, e
		}
		return reqCtx, errcommon.Error{Code: errcommon.ResourceExhausted, Msg: fmt.Errorf("failed to find target endpoint: %w", err).Error()}
	}

	// Conditional-decode gate (RFC 7240 "Prefer: if-available"). The coordinator
	// uses this header to mark a speculative early-decode attempt: forward to a
	// decode worker only if its KV cache already covers the prompt, otherwise
	// surface 412 Precondition Failed so the coordinator restarts the pipeline
	// at encode/prefill/decode. Lives in the director (not in a profile handler)
	// so it fires regardless of which profile handler is configured.
	if routing.IsConditionalDecode(reqCtx.Request.Headers) {
		decider := d.requestControlPlugins.ConditionalDecodeDecider()
		if decider == nil {
			logger.V(logutil.DEBUG).Info("conditional-decode: no decider configured, forwarding")
		} else {
			endpoint := primaryDecodeEndpoint(logger, result)
			if endpoint == nil || decider.ShouldRejectConditionalDecode(ctx, reqCtx.SchedulingRequest, endpoint) {
				logger.V(logutil.DEBUG).Info("conditional-decode: chosen decode worker has no cached prefix, returning 412")
				return reqCtx, errcommon.Error{
					Code: errcommon.PreconditionFailed,
					Msg:  "no decode worker has the requested KV cache",
				}
			}
			logger.V(logutil.DEBUG).Info("conditional-decode: chosen decode worker has cached prefix, forwarding")
		}
	}

	reqCtx.SchedulingRequest.SchedulingResult = result

	// Prepare Request (Populates RequestContext and call PreRequest plugins)
	// Insert target endpoint to instruct Envoy to route requests to the specified target pod and attach the port number.
	// Invoke PreRequest registered plugins.
	reqCtx, err = d.prepareRequest(ctx, reqCtx, result)
	if err != nil {
		return reqCtx, err
	}
	if err := d.repackage(ctx, reqCtx, inferenceRequestBody); err != nil {
		return reqCtx, err
	}
	return reqCtx, nil
}

func (d *Director) modelRewriteIfNeeded(ctx context.Context, reqCtx *handlers.RequestContext, inferenceRequestBody *fwkrh.InferenceRequestBody) error {
	if v, ok := inferenceRequestBody.Payload.(fwkrh.PayloadMap); ok {
		// Mutate the model name inside the map, this is currently only supported if the payload is a PayloadMap.
		_, err := d.mutateModel(ctx, reqCtx, v)
		if err != nil {
			return err
		}
	}
	return nil
}

func (d *Director) mutateModel(ctx context.Context, reqCtx *handlers.RequestContext, bodyMap map[string]any) (*handlers.RequestContext, error) {
	reqCtx.IncomingModelName, _ = bodyMap["model"].(string)
	if reqCtx.TargetModelName == "" {
		reqCtx.TargetModelName = reqCtx.IncomingModelName
	}
	d.applyWeightedModelRewrite(ctx, reqCtx)
	if reqCtx.TargetModelName == "" {
		return reqCtx, errcommon.Error{Code: errcommon.BadRequest, Msg: "model not found in request body"}
	}
	bodyMap["model"] = reqCtx.TargetModelName
	return reqCtx, nil
}

func (d *Director) repackage(ctx context.Context, reqCtx *handlers.RequestContext, inferenceRequestBody *fwkrh.InferenceRequestBody) error {
	logger := log.FromContext(ctx)
	switch v := inferenceRequestBody.Payload.(type) {
	case fwkrh.PayloadMap:
		requestBodyBytes, err := json.Marshal(v)
		if err != nil {
			logger.Error(err, "Error marshalling request body")
			return errcommon.Error{Code: errcommon.Internal, Msg: "Error marshalling request body"}
		}
		reqCtx.Request.RawBody = requestBodyBytes
		reqCtx.RequestSize = len(requestBodyBytes)
	case fwkrh.PayloadProto, fwkrh.RawPayload:
		reqCtx.RequestSize = len(reqCtx.Request.RawBody)
	default:
		return errcommon.Error{Code: errcommon.BadRequest, Msg: "Unsupported llmRequest parsedBody"}
	}
	return nil
}

func (d *Director) applyWeightedModelRewrite(ctx context.Context, reqCtx *handlers.RequestContext) {
	rewriteRule, modelRewriteName := d.datastore.ModelRewriteGet(reqCtx.IncomingModelName)
	if rewriteRule == nil {
		return
	}
	reqCtx.TargetModelName = d.selectWeightedModel(ctx, rewriteRule.Targets)
	metrics.RecordInferenceModelRewriteDecision(modelRewriteName, reqCtx.IncomingModelName, reqCtx.TargetModelName)
}

func (d *Director) selectWeightedModel(ctx context.Context, models []v1alpha2.TargetModel) string {
	if len(models) == 0 {
		return ""
	}

	var totalWeight int32
	var weightedTargets int
	for _, model := range models {
		if model.Weight != nil {
			weightedTargets++
			totalWeight += *model.Weight
		}
	}
	if weightedTargets > 0 && weightedTargets < len(models) {
		log.FromContext(ctx).Info("Warning: model rewrite target weights are mixed; targets without weights will not be selected",
			"weightedTargets", weightedTargets,
			"unweightedTargets", len(models)-weightedTargets,
		)
	}

	if totalWeight == 0 {
		// If total weight is 0, distribute evenly
		return models[rand.Intn(len(models))].ModelRewrite
	}

	randomNum := rand.Intn(int(totalWeight))
	var currentWeight int32
	for _, model := range models {
		if model.Weight != nil {
			currentWeight += *model.Weight
		}
		if randomNum < int(currentWeight) {
			return model.ModelRewrite
		}
	}

	// Should not happen
	return models[len(models)-1].ModelRewrite
}

// prepareRequest populates the RequestContext and calls the registered PreRequest plugins
// for allowing plugging customized logic based on the scheduling result.
func (d *Director) prepareRequest(ctx context.Context, reqCtx *handlers.RequestContext, result *fwksched.SchedulingResult) (*handlers.RequestContext, error) {
	logger := log.FromContext(ctx)
	if result == nil || len(result.ProfileResults) == 0 {
		return reqCtx, errcommon.Error{Code: errcommon.Internal, Msg: "results must be greater than zero"}
	}
	// primary profile is used to set destination
	targetMetadatas := []*fwkdl.EndpointMetadata{}
	targetEndpoints := []string{}

	for _, pod := range result.ProfileResults[result.PrimaryProfileName].TargetEndpoints {
		curMetadata := pod.GetMetadata()
		curEndpoint := net.JoinHostPort(curMetadata.GetIPAddress(), curMetadata.GetPort())
		targetMetadatas = append(targetMetadatas, curMetadata)
		targetEndpoints = append(targetEndpoints, curEndpoint)
	}

	multiEndpointString := strings.Join(targetEndpoints, ",")
	logger.V(logutil.VERBOSE).Info("Request handled", "objectiveKey", reqCtx.ObjectiveKey, "incomingModelName", reqCtx.IncomingModelName, "targetModel", reqCtx.TargetModelName, "endpoint", multiEndpointString)

	reqCtx.TargetPod = targetMetadatas[0]
	reqCtx.TargetEndpoint = multiEndpointString

	d.runPreRequestPlugins(ctx, reqCtx.SchedulingRequest, result)

	return reqCtx, nil
}

func (d *Director) toSchedulerEndpoints(endpoints []fwkdl.Endpoint) []fwksched.Endpoint {
	result := make([]fwksched.Endpoint, len(endpoints))
	for i, endpoint := range endpoints {
		result[i] = fwksched.NewEndpoint(endpoint.GetMetadata(), endpoint.GetMetrics(), endpoint.GetAttributes())
	}

	return result
}

// HandleResponseHeader is called when the response headers are received.
func (d *Director) HandleResponseHeader(ctx context.Context, reqCtx *handlers.RequestContext) *handlers.RequestContext {
	if len(d.requestControlPlugins.responseReceivedPlugins) == 0 {
		return reqCtx
	}
	response := &fwkrc.Response{
		RequestID:   reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey],
		Headers:     reqCtx.Response.Headers,
		ReqMetadata: reqCtx.Request.Metadata,
	}
	// TODO: to extend fallback functionality, handle cases where target pod is unavailable
	// https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/1224
	d.runResponseHeaderPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
	return reqCtx
}

// 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 (d *Director) HandleResponseBody(ctx context.Context, reqCtx *handlers.RequestContext, endOfStream bool) *handlers.RequestContext {
	logger := log.FromContext(ctx).WithValues("stage", "bodyChunk")
	logger.V(logutil.TRACE).Info("Entering HandleResponseBodyChunk")
	if len(d.requestControlPlugins.responseStreamingPlugins) == 0 {
		logger.V(logutil.TRACE).Info("Exiting HandleResponseBodyChunk")
		return reqCtx
	}

	startOfStream := !reqCtx.ResponseBodyStarted
	reqCtx.ResponseBodyStarted = true
	response := &fwkrc.Response{
		RequestID:     reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey],
		Headers:       reqCtx.Response.Headers,
		StartOfStream: startOfStream,
		EndOfStream:   endOfStream,
		Usage:         reqCtx.Usage,
	}
	requestID := reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey]

	if endOfStream {
		// Drain the async queue: close the channel and wait for the goroutine to finish
		// processing all previously queued chunks before running the final chunk synchronously.
		if val, ok := d.responseBodyQueues.LoadAndDelete(reqCtx); ok {
			q := val.(*responseBodyQueue)
			q.closeAndWait()
		}
		// Run the final chunk synchronously so DynamicMetadata is available for the response.
		d.runResponseBodyPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
		reqCtx.Response.DynamicMetadata = response.DynamicMetadata
	} else {
		// Get or create the async queue for this request.
		work := responseBodyWork{
			ctx:            ctx,
			request:        reqCtx.SchedulingRequest,
			response:       response,
			targetEndpoint: reqCtx.TargetPod,
		}
		q := d.loadOrCreateResponseBodyQueue(reqCtx)
		if !q.enqueue(work) {
			logger.V(logutil.DEBUG).Info("Skipping response body chunk because the async queue is closed", "requestID", requestID)
		}
	}
	logger.V(logutil.TRACE).Info("Exiting HandleResponseBodyChunk")
	return reqCtx
}

func (d *Director) loadOrCreateResponseBodyQueue(reqCtx *handlers.RequestContext) *responseBodyQueue {
	if val, ok := d.responseBodyQueues.Load(reqCtx); ok {
		return val.(*responseBodyQueue)
	}
	q := newResponseBodyQueue()
	val, loaded := d.responseBodyQueues.LoadOrStore(reqCtx, q)
	if loaded {
		return val.(*responseBodyQueue)
	}
	go d.processResponseBodyQueue(q)
	return q
}

func (d *Director) GetRandomEndpoint() *fwkdl.EndpointMetadata {
	pods := d.datastore.PodList(datastore.AllPodsPredicate)
	if len(pods) == 0 {
		return nil
	}
	number := rand.Intn(len(pods))
	pod := pods[number]
	return pod.GetMetadata()
}

func (d *Director) runPreRequestPlugins(ctx context.Context, request *fwksched.InferenceRequest,
	schedulingResult *fwksched.SchedulingResult) {
	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
	for _, plugin := range d.requestControlPlugins.preRequestPlugins {
		loggerDebug.Info("Running PreRequest plugin", "plugin", plugin.TypedName())
		before := time.Now()
		plugin.PreRequest(ctx, request, schedulingResult)
		metrics.RecordPluginProcessingLatency(fwkrc.PreRequestExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
		loggerDebug.Info("Completed running PreRequest plugin successfully", "plugin", plugin.TypedName())
	}
}

func (d *Director) runPreAdmissionPlugins(ctx context.Context, request *fwksched.InferenceRequest) error {
	if len(d.requestControlPlugins.preAdmissionPlugins) == 0 {
		return nil
	}
	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
	for _, plugin := range d.requestControlPlugins.preAdmissionPlugins {
		loggerDebug.Info("Running PreAdmitter plugin", "plugin", plugin.TypedName())
		before := time.Now()
		if err := plugin.PreAdmit(ctx, request); err != nil {
			return err
		}
		metrics.RecordPluginProcessingLatency(fwkrc.PreAdmissionExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
		loggerDebug.Info("Completed running PreAdmitter plugin successfully", "plugin", plugin.TypedName())
	}
	return nil
}

func (d *Director) runDataProducerPlugins(ctx context.Context,
	request *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error {
	plugins := d.requestControlPlugins.dataProducerPlugins
	if len(plugins) == 0 {
		return nil
	}
	// Each producer runs under its own timeout so a slow one does not extend the
	// budget of the others.
	for _, p := range plugins {
		if err := dataProducerPluginsWithTimeout(ctx, producerTimeout(p), []fwkrc.DataProducer{p}, request, endpoints); err != nil {
			return err
		}
	}
	return nil
}

func (d *Director) runAdmissionPlugins(ctx context.Context,
	request *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error {
	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
	for _, plugin := range d.requestControlPlugins.admissionPlugins {
		loggerDebug.Info("Running Admit plugin", "plugin", plugin.TypedName())
		before := time.Now()
		denyReason := plugin.Admit(ctx, request, endpoints)
		metrics.RecordPluginProcessingLatency(fwkrc.AdmissionExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
		if denyReason != nil {
			loggerDebug.Info("Admit plugin denied the request", "plugin", plugin.TypedName(), "reason", denyReason.Error())
			return denyReason
		}
		loggerDebug.Info("Completed running Admit plugin successfully", "plugin", plugin.TypedName())
	}
	return nil
}

func (d *Director) runResponseHeaderPlugins(ctx context.Context, request *fwksched.InferenceRequest, response *fwkrc.Response, targetEndpoint *fwkdl.EndpointMetadata) {
	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
	for _, plugin := range d.requestControlPlugins.responseReceivedPlugins {
		loggerDebug.Info("Running ResponseReceived plugin", "plugin", plugin.TypedName())
		before := time.Now()
		plugin.ResponseHeader(ctx, request, response, targetEndpoint)
		metrics.RecordPluginProcessingLatency(fwkrc.ResponseReceivedExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
		loggerDebug.Info("Completed running ResponseReceived plugin successfully", "plugin", plugin.TypedName())
	}
}

func (d *Director) runResponseBodyPlugins(ctx context.Context, request *fwksched.InferenceRequest, response *fwkrc.Response, targetEndpoint *fwkdl.EndpointMetadata) {
	loggerTrace := log.FromContext(ctx).V(logutil.TRACE)
	for _, plugin := range d.requestControlPlugins.responseStreamingPlugins {
		loggerTrace.Info("Running ResponseStreaming plugin", "plugin", plugin.TypedName())
		before := time.Now()
		plugin.ResponseBody(ctx, request, response, targetEndpoint)
		metrics.RecordPluginProcessingLatency(fwkrc.ResponseStreamingExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
		loggerTrace.Info("Completed running ResponseStreaming plugin successfully", "plugin", plugin.TypedName())
	}
}

// processResponseBodyQueue reads work items from the queue channel and runs response body
// plugins for each one sequentially. It exits when the channel is closed and signals
// completion by closing q.done.
func (d *Director) processResponseBodyQueue(q *responseBodyQueue) {
	defer close(q.done)
	for work := range q.ch {
		d.runResponseBodyPlugins(work.ctx, work.request, work.response, work.targetEndpoint)
	}
}