runner.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 runner

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"
	"regexp"
	"sync/atomic"
	"time"

	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/spf13/pflag"
	"google.golang.org/grpc"
	healthPb "google.golang.org/grpc/health/grpc_health_v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	configapi "sigs.k8s.io/gateway-api-inference-extension/apix/config/v1alpha1"
	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/common"
	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/common/observability/logging"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/common/observability/profiling"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/common/observability/tracing"
	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/config"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/config/loader"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
	fccontroller "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/controller"
	fcregistry "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/registry"
	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
	fwkrh "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requesthandling"
	extractormetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/extractor/metrics"
	sourcemetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/source/metrics"
	sourcenotifications "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/source/notifications"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/fairness/globalstrict"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/fairness/roundrobin"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/ordering/edf"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/ordering/fcfs"
	slodeadline "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/flowcontrol/usagelimits"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/admitter/latencyslo"
	reqdataprodprefix "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/dataproducer/approximateprefix"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/requestattributereporter"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/requestdataproducer/inflightload"
	latencyproducer "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/requestdataproducer/predictedlatency"
	testresponsereceived "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requestcontrol/test/responsereceived"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requesthandling/parsers/openai"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requesthandling/parsers/passthrough"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/filter/prefixcacheaffinity"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/filter/sloheadroomtier"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/picker/maxscore"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/picker/random"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/picker/weightedrandom"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/profile"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/kvcacheutilization"
	latencyscorer "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/latency"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/loraaffinity"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/prefix"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/queuedepth"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/runningrequests"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/scorer/tokenload"
	testfilter "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/scheduling/test/filter"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/handlers"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics/collectors"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/requestcontrol"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/env"
	"sigs.k8s.io/gateway-api-inference-extension/version"
)

const (
	// enableExperimentalFlowControlLayer defines the environment variable used as a feature flag for the pluggable flow
	// control layer.
	// DEPRECATION NOTICE - this env var will be removed in the next version as we switch to configuring the EPP using FeatureGates in the config file.
	enableExperimentalFlowControlLayer = "ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER"
)

var (
	setupLog = ctrl.Log.WithName("setup")
)

// NewRunner initializes a new EPP Runner and returns its pointer.
func NewRunner() *Runner {
	return &Runner{
		eppExecutableName:    "GIE",
		requestControlConfig: requestcontrol.NewConfig(), // default requestcontrol config has empty plugin list
		customCollectors:     []prometheus.Collector{},
	}
}

// Runner is used to run epp with its plugins
type Runner struct {
	eppExecutableName    string // the EPP executable name
	featureGates         map[string]bool
	requestControlConfig *requestcontrol.Config
	schedulerConfig      *scheduling.SchedulerConfig
	customCollectors     []prometheus.Collector
	parser               fwkrh.Parser
	dlRuntime            *datalayer.Runtime
}

// WithExecutableName sets the name of the executable containing the runner.
// The name is used in the version log upon startup and is otherwise opaque.
func (r *Runner) WithExecutableName(exeName string) *Runner {
	r.eppExecutableName = exeName
	return r
}

func (r *Runner) WithRequestControlConfig(requestControlConfig *requestcontrol.Config) *Runner {
	r.requestControlConfig = requestControlConfig
	return r
}

func (r *Runner) WithSchedulerConfig(schedulerConfig *scheduling.SchedulerConfig) *Runner {
	r.schedulerConfig = schedulerConfig
	return r
}

func (r *Runner) WithCustomCollectors(collectors ...prometheus.Collector) *Runner {
	r.customCollectors = collectors
	return r
}

func (r *Runner) Run(ctx context.Context) error {
	// Setup a very basic logger in case command line argument parsing fails
	logutil.InitSetupLogging()

	setupLog.Info(r.eppExecutableName+" build", "commit-sha", version.CommitSHA, "build-ref", version.BuildRef)

	opts := runserver.NewOptions()
	opts.AddFlags(pflag.CommandLine)
	pflag.Parse()

	if err := opts.Complete(); err != nil {
		return err
	}
	if err := opts.Validate(); err != nil {
		setupLog.Error(err, "Failed to validate flags")
		return err
	}

	// Print all flag values
	flags := make(map[string]any)
	pflag.VisitAll(func(f *pflag.Flag) {
		flags[f.Name] = f.Value
	})
	setupLog.Info("Flags processed", "flags", flags)

	logutil.InitLogging(&opts.ZapOptions)

	if opts.Tracing {
		err := tracing.InitTracing(ctx, setupLog, "gateway-api-inference-extension/epp")
		if err != nil {
			return fmt.Errorf("failed to init tracing %w", err)
		}
	}

	// --- Get Kubernetes Config ---
	cfg, err := ctrl.GetConfig()
	if err != nil {
		setupLog.Error(err, "Failed to get Kubernetes rest config")
		return err
	}

	pmc, err := backendmetrics.NewPodMetricsClientImpl(setupLog, backendmetrics.Config{
		ModelServerMetricsScheme:        opts.ModelServerMetricsScheme,
		ModelServerMetricsHTTPSInsecure: opts.ModelServerMetricsHTTPSInsecure,
		ModelServerMetricsPath:          opts.ModelServerMetricsPath,

		TotalQueuedRequestsMetric:    opts.TotalQueuedRequestsMetric,
		TotalRunningRequestsMetric:   opts.TotalRunningRequestsMetric,
		KVCacheUsagePercentageMetric: opts.KVCacheUsagePercentageMetric,
		LoRAInfoMetric:               opts.LoRAInfoMetric,
		CacheInfoMetric:              opts.CacheInfoMetric,
	})
	if err != nil {
		return err
	}

	mgr, _, err := r.setup(ctx, cfg, opts, pmc, nil)
	if err != nil {
		return err
	}

	// --- Start Manager ---
	// This blocks until a signal is received.
	setupLog.Info("Controller manager starting")
	if err := mgr.Start(ctx); err != nil {
		setupLog.Error(err, "Error starting controller manager")
		return err
	}
	setupLog.Info("Controller manager terminated")
	return nil
}

// setup configures the internal state of the Runner, including the manager,
// datastore, and other server components. It returns the initialized Manager
// without starting it, allowing for flexible use in integration tests.
//
// The returned Datastore is **only** meant to be used in the integration test.
// Optional managerOverrides are applied to the controller manager options before creation.
func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, pmc backendmetrics.PodMetricsClient, managerOverrides []func(*ctrl.Options)) (ctrl.Manager, datastore.Datastore, error) {
	rawConfig, err := r.parseConfigurationPhaseOne(ctx, opts)
	if err != nil {
		setupLog.Error(err, "Failed to parse configuration")
		return nil, nil, err
	}
	useNewMetrics := !r.featureGates[datalayer.EnableLegacyMetricsFeatureGate]
	epf := r.setupMetricsCollection(useNewMetrics, opts, pmc)
	gknn, err := extractGKNN(opts.PoolName, opts.PoolGroup, opts.PoolNamespace, opts.EndpointSelector)
	if err != nil {
		setupLog.Error(err, "Failed to extract GKNN")
		return nil, nil, err
	}

	startCrdReconcilers := opts.EndpointSelector == "" // If endpointSelector is empty, it means it's not in the standalone mode. Then we should start the inferencePool and other CRD Reconciler.
	controllerCfg := runserver.NewControllerConfig(startCrdReconcilers)
	if err := controllerCfg.PopulateControllerConfig(cfg); err != nil {
		setupLog.Error(err, "Failed to populate controller config")
		return nil, nil, err
	}

	ds, err := setupDatastore(ctx, epf, int32(opts.ModelServerMetricsPort), startCrdReconcilers,
		opts.PoolNamespace, opts.PoolName, opts.EndpointSelector, opts.EndpointTargetPorts)
	if err != nil {
		setupLog.Error(err, "Failed to setup datastore")
		return nil, nil, err
	}
	eppConfig, err := r.parseConfigurationPhaseTwo(ctx, rawConfig, ds)
	if err != nil {
		setupLog.Error(err, "Failed to parse configuration")
		return nil, nil, err
	}

	// --- Setup Metrics Server ---
	r.customCollectors = append(r.customCollectors, collectors.NewInferencePoolMetricsCollector(ds))
	metrics.Register(r.customCollectors...)
	metrics.RecordInferenceExtensionInfo(version.CommitSHA, version.BuildRef)
	// Register metrics handler.
	// Metrics endpoint is enabled in 'config/default/kustomization.yaml'. The Metrics options configure the server.
	// More info:
	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.19.1/pkg/metrics/server
	// - https://book.kubebuilder.io/reference/metrics.html
	metricsServerOptions := metricsserver.Options{
		BindAddress: fmt.Sprintf(":%d", opts.MetricsPort),
		FilterProvider: func() func(c *rest.Config, httpClient *http.Client) (metricsserver.Filter, error) {
			if opts.MetricsEndpointAuth {
				return filters.WithAuthenticationAndAuthorization
			}

			return nil
		}(),
	}

	isLeader := &atomic.Bool{}
	isLeader.Store(false)

	mgr, err := runserver.NewDefaultManager(controllerCfg, *gknn, cfg, metricsServerOptions, opts.EnableLeaderElection, managerOverrides...)
	if err != nil {
		setupLog.Error(err, "Failed to create controller manager")
		return nil, nil, err
	}

	if opts.EnableLeaderElection {
		setupLog.Info("Leader election enabled")
		go func() {
			<-mgr.Elected()
			isLeader.Store(true)
			setupLog.Info("This instance is now the leader!")
		}()
	} else {
		// If leader election is disabled, all instances are "leaders" for readiness purposes.
		isLeader.Store(true)
	}

	if opts.EnablePprof {
		setupLog.Info("Setting pprof handlers")
		if err = profiling.SetupPprofHandlers(mgr); err != nil {
			setupLog.Error(err, "Failed to setup pprof handlers")
			return nil, nil, err
		}
	}

	// --- Initialize Core EPP Components ---
	if r.schedulerConfig == nil {
		err := errors.New("scheduler config must be set either by config api or through code")
		setupLog.Error(err, "failed to create scheduler")
		return nil, nil, err
	}

	setupLog.Info("parsed config", "scheduler-config", r.schedulerConfig)

	scheduler := scheduling.NewSchedulerWithConfig(r.schedulerConfig)

	// Data layer is enabled by default; use the 'enableLegacyMetrics' feature gate to fall back to legacy polling.
	datalayerMetricsEnabled := !r.featureGates[datalayer.EnableLegacyMetricsFeatureGate]
	if err := r.configureAndStartDatalayer(ctx, datalayerMetricsEnabled, eppConfig.DataConfig, mgr); err != nil {
		setupLog.Error(err, "failed to initialize data layer")
		return nil, nil, err
	}

	// --- Admission Control Initialization ---
	var admissionController requestcontrol.AdmissionController
	var endpointCandidates contracts.EndpointCandidates
	endpointCandidates = requestcontrol.NewDatastoreEndpointCandidates(ds, requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter))
	if r.featureGates[flowcontrol.FeatureGate] {
		endpointCandidates = requestcontrol.NewCachedEndpointCandidates(ctx, endpointCandidates, time.Millisecond*50)
		setupLog.Info("Initializing experimental Flow Control layer")
		registry, err := fcregistry.NewFlowRegistry(eppConfig.FlowControlConfig.Registry, setupLog)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to initialize Flow Registry: %w", err)
		}
		fc, err := fccontroller.NewFlowController(
			ctx,
			opts.PoolName,
			eppConfig.FlowControlConfig.Controller,
			fccontroller.Deps{
				Registry:           registry,
				SaturationDetector: eppConfig.SaturationDetector,
				EndpointCandidates: endpointCandidates,
				UsageLimitPolicy:   eppConfig.FlowControlConfig.UsageLimitPolicy,
			},
		)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to initialize Flow Controller: %w", err)
		}
		go registry.Run(ctx)
		admissionController = requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName)
	} else {
		setupLog.Info("Experimental Flow Control layer is disabled, using legacy admission control")
		admissionController = requestcontrol.NewLegacyAdmissionController(eppConfig.SaturationDetector, endpointCandidates)
	}

	director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, endpointCandidates, r.requestControlConfig)

	serverRunner := &runserver.ExtProcServerRunner{
		GrpcPort:                         opts.GRPCPort,
		GKNN:                             *gknn,
		Datastore:                        ds,
		ControllerCfg:                    controllerCfg,
		SecureServing:                    opts.SecureServing,
		HealthChecking:                   opts.HealthChecking,
		CertPath:                         opts.CertPath,
		EnableCertReload:                 opts.EnableCertReload,
		RefreshPrometheusMetricsInterval: opts.RefreshPrometheusMetricsInterval,
		MetricsStalenessThreshold:        opts.MetricsStalenessThreshold,
		Director:                         director,
		Parser:                           r.parser,
		SaturationDetector:               eppConfig.SaturationDetector,
		UseExperimentalDatalayerV2:       r.featureGates[datalayer.ExperimentalDatalayerFeatureGate] || !r.featureGates[datalayer.EnableLegacyMetricsFeatureGate],
	}

	if err := serverRunner.SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "Failed to setup EPP controllers")
		return nil, nil, err
	}

	// --- Add Runnables to Manager ---
	// Register health server.
	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), ds, opts.GRPCHealthPort, isLeader, opts.EnableLeaderElection, r.parser); err != nil {
		return nil, nil, err
	}

	// Register ext-proc server.
	if err := registerExtProcServer(mgr, serverRunner, ctrl.Log.WithName("ext-proc")); err != nil {
		return nil, nil, err
	}
	return mgr, ds, nil
}

// NewEndpointPoolFromOptions constructs an EndpointPool from standalone options.
// This is shared between the production runner and standalone integration tests.
func NewEndpointPoolFromOptions(
	namespace string,
	name string,
	endpointSelector string,
	endpointTargetPorts []int,
) (*datalayer.EndpointPool, error) {

	if namespace == "" {
		return nil, errors.New("namespace must not be empty")
	}
	if name == "" {
		return nil, errors.New("name must not be empty")
	}
	if endpointSelector == "" {
		return nil, errors.New("endpoint selector must not be empty")
	}
	if len(endpointTargetPorts) == 0 {
		return nil, errors.New("endpoint target ports must not be empty")
	}

	selectorMap, err := labels.ConvertSelectorToLabelsMap(endpointSelector)
	if err != nil {
		return nil, fmt.Errorf("failed to parse endpoint selector %q: %w", endpointSelector, err)
	}

	pool := datalayer.NewEndpointPool(namespace, name)
	pool.Selector = selectorMap
	pool.TargetPorts = append(pool.TargetPorts, endpointTargetPorts...)

	return pool, nil
}

func setupDatastore(ctx context.Context, epFactory datalayer.EndpointFactory, modelServerMetricsPort int32,
	startCrdReconcilers bool, namespace, name, endpointSelector string, endpointTargetPorts []int) (datastore.Datastore, error) {

	if startCrdReconcilers {
		return datastore.NewDatastore(ctx, epFactory, modelServerMetricsPort), nil
	} else {
		endpointPool, err := NewEndpointPoolFromOptions(namespace, name, endpointSelector, endpointTargetPorts)
		if err != nil {
			setupLog.Error(err, "Failed to construct endpoint pool from options")
			return nil, err
		}
		return datastore.NewDatastore(ctx, epFactory, modelServerMetricsPort).WithEndpointPool(endpointPool), nil
	}
}

// registerInTreePlugins registers the factory functions of all known plugins
func (r *Runner) registerInTreePlugins() {
	fwkplugin.Register(prefix.PrefixCacheScorerPluginType, prefix.PrefixCachePluginFactory)
	fwkplugin.Register(maxscore.MaxScorePickerType, maxscore.MaxScorePickerFactory)
	fwkplugin.Register(random.RandomPickerType, random.RandomPickerFactory)
	fwkplugin.Register(weightedrandom.WeightedRandomPickerType, weightedrandom.WeightedRandomPickerFactory)
	fwkplugin.Register(profile.SingleProfileHandlerType, profile.SingleProfileHandlerFactory)
	fwkplugin.Register(kvcacheutilization.KvCacheUtilizationScorerType, kvcacheutilization.KvCacheUtilizationScorerFactory)
	fwkplugin.Register(queuedepth.QueueScorerType, queuedepth.QueueScorerFactory)
	fwkplugin.Register(runningrequests.RunningRequestsSizeScorerType, runningrequests.RunningRequestsSizeScorerFactory)
	fwkplugin.Register(loraaffinity.LoraAffinityScorerType, loraaffinity.LoraAffinityScorerFactory)
	fwkplugin.Register(tokenload.TokenLoadScorerType, tokenload.TokenLoadScorerFactory)
	// Flow Control plugins
	fwkplugin.Register(globalstrict.GlobalStrictFairnessPolicyType, globalstrict.GlobalStrictFairnessPolicyFactory)
	fwkplugin.Register(roundrobin.RoundRobinFairnessPolicyType, roundrobin.RoundRobinFairnessPolicyFactory)
	fwkplugin.Register(fcfs.FCFSOrderingPolicyType, fcfs.FCFSOrderingPolicyFactory)
	fwkplugin.Register(edf.EDFOrderingPolicyType, edf.EDFOrderingPolicyFactory)
	fwkplugin.Register(slodeadline.SLODeadlineOrderingPolicyType, slodeadline.SLODeadlineOrderingPolicyFactory)
	fwkplugin.Register(usagelimits.StaticUsageLimitPolicyType, usagelimits.StaticPolicyFactory)
	fwkplugin.Register(reqdataprodprefix.ApproxPrefixCachePluginType, reqdataprodprefix.ApproxPrefixCacheFactory)
	// Latency predictor plugins
	fwkplugin.Register(latencyproducer.LatencyDataProviderPluginType, latencyproducer.PredictedLatencyFactory)
	fwkplugin.Register(latencyslo.LatencyAdmissionPluginType, latencyslo.LatencyAdmissionFactory)
	// Latency scoring and filtering plugins
	fwkplugin.Register(prefixcacheaffinity.PluginType, prefixcacheaffinity.Factory)
	fwkplugin.Register(sloheadroomtier.PluginType, sloheadroomtier.Factory)
	fwkplugin.Register(latencyscorer.LatencyScorerType, latencyscorer.Factory)
	// In-flight load producer
	fwkplugin.Register(inflightload.InFlightLoadProducerType, inflightload.InFlightLoadProducerFactory)
	// register filter for test purpose only (used in conformance tests)
	fwkplugin.Register(testfilter.HeaderBasedTestingFilterType, testfilter.HeaderBasedTestingFilterFactory)
	// register response received plugin for test purpose only (used in conformance tests)
	fwkplugin.Register(testresponsereceived.DestinationEndpointServedVerifierType, testresponsereceived.DestinationEndpointServedVerifierFactory)
	// register datalayer metrics collection plugins
	fwkplugin.Register(sourcemetrics.MetricsDataSourceType, sourcemetrics.MetricsDataSourceFactory)
	fwkplugin.Register(extractormetrics.MetricsExtractorType, extractormetrics.CoreMetricsExtractorFactory)
	// register datalayer notification source plugins
	fwkplugin.Register(sourcenotifications.NotificationSourceType, sourcenotifications.NotificationSourceFactory)
	fwkplugin.Register(sourcenotifications.EndpointNotificationSourceType, sourcenotifications.EndpointSourceFactory)
	// register request control pluigns
	fwkplugin.Register(requestattributereporter.RequestAttributeReporterType, requestattributereporter.RequestAttributeReporterPluginFactory)
	fwkplugin.Register(openai.OpenAIParserType, openai.OpenAIParserPluginFactory)
	fwkplugin.Register(vllmgrpc.VllmGRPCParserType, vllmgrpc.VllmGRPCParserPluginFactory)
	fwkplugin.Register(passthrough.PassthroughParserType, passthrough.PassthroughParserPluginFactory)
	// register saturation detector plugins
	fwkplugin.Register(concurrency.ConcurrencyDetectorType, concurrency.ConcurrencyDetectorFactory)
	fwkplugin.Register(utilization.UtilizationDetectorType, utilization.UtilizationDetectorFactory)
}

func (r *Runner) parseConfigurationPhaseOne(ctx context.Context, opts *runserver.Options) (*configapi.EndpointPickerConfig, error) {
	logger := log.FromContext(ctx)

	var configBytes []byte
	if opts.ConfigText != "" {
		configBytes = []byte(opts.ConfigText)
	} else if opts.ConfigFile != "" { // if config was specified through a file
		var err error
		configBytes, err = os.ReadFile(opts.ConfigFile)
		if err != nil {
			return nil, fmt.Errorf("failed to load config from a file '%s' - %w", opts.ConfigFile, err)
		}
	}

	loader.RegisterFeatureGate(datalayer.ExperimentalDatalayerFeatureGate)
	loader.RegisterFeatureGate(datalayer.EnableLegacyMetricsFeatureGate)
	loader.RegisterFeatureGate(flowcontrol.FeatureGate)

	r.registerInTreePlugins()

	rawConfig, featureGates, err := loader.LoadRawConfig(configBytes, logger)
	if err != nil {
		return nil, fmt.Errorf("failed to parse config - %w", err)
	}

	r.featureGates = featureGates

	if r.featureGates[datalayer.ExperimentalDatalayerFeatureGate] {
		setupLog.Info("The data layer is now enabled by default. " +
			"Please remove the 'dataLayer' feature gate from your config. " +
			"To fall back to legacy metrics polling, use the 'enableLegacyMetrics' feature gate.")
	}

	if r.featureGates[datalayer.EnableLegacyMetricsFeatureGate] {
		setupLog.Info("Data layer: using legacy metrics polling (opt-in via 'enableLegacyMetrics' feature gate)")
	} else {
		setupLog.Info("Data layer: ENABLED (default)")
	}

	return rawConfig, nil
}

// Return a function that can be used in the EPP Handle to list pod names.
func makePodListFunc(ds datastore.Datastore) func() []types.NamespacedName {
	return func() []types.NamespacedName {
		pods := ds.PodList(datastore.AllPodsPredicate)
		names := make([]types.NamespacedName, 0, len(pods))

		for _, p := range pods {
			names = append(names, p.GetMetadata().NamespacedName)
		}
		return names
	}
}

func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *configapi.EndpointPickerConfig, ds datastore.Datastore) (*config.Config, error) {
	logger := log.FromContext(ctx)

	applyDeprecatedEnvFeatureGate(enableExperimentalFlowControlLayer, "Flow Control layer", flowcontrol.FeatureGate, rawConfig)

	handle := fwkplugin.NewEppHandle(ctx, makePodListFunc(ds))
	cfg, err := loader.InstantiateAndConfigure(rawConfig, handle, logger)

	if err != nil {
		return nil, fmt.Errorf("failed to load the configuration - %w", err)
	}

	r.schedulerConfig = cfg.SchedulerConfig

	// Add requestControl plugins
	r.requestControlConfig.AddPlugins(handle.GetAllPlugins()...)

	// Sort data plugins in DAG order (topological sort). Also check DAG for cycles.
	dag, err := datalayer.ValidateAndOrderDataDependencies(handle.GetAllPlugins())

	if err != nil {
		return nil, fmt.Errorf("failed to load the configuration - %w", err)
	}
	// The plugins will be executed in topologically sorted order to ensure that data is produced before it is consumed.
	r.requestControlConfig.OrderPrepareDataPlugins(dag)

	r.parser = handlers.NewParser(cfg.ParserConfig)
	logger.Info("loaded configuration from file/text successfully")

	return cfg, nil
}

func applyDeprecatedEnvFeatureGate(envVar, featureName, featureGate string, rawConfig *configapi.EndpointPickerConfig) {
	if _, ok := os.LookupEnv(envVar); ok {
		setupLog.Info(fmt.Sprintf("Enabling the experimental %s using environment variables is deprecated and will be removed in next version", featureName))
		if env.GetEnvBool(envVar, false, setupLog) {
			if rawConfig.FeatureGates == nil {
				rawConfig.FeatureGates = make(configapi.FeatureGates, 0)
			}
			rawConfig.FeatureGates = append(rawConfig.FeatureGates, featureGate)
		}
	}
}

func (r *Runner) configureAndStartDatalayer(ctx context.Context, enableNewMetrics bool, cfg *datalayer.Config, mgr ctrl.Manager) error {
	disallowedExtractorType := ""
	if !enableNewMetrics {
		disallowedExtractorType = extractormetrics.MetricsExtractorType
	}

	if err := r.dlRuntime.Configure(cfg, enableNewMetrics, disallowedExtractorType, setupLog); err != nil {
		return err
	}

	if err := r.dlRuntime.Start(ctx, mgr); err != nil {
		return err
	}
	return nil
}

func (r *Runner) setupMetricsCollection(enableNewMetrics bool, opts *runserver.Options, pmc backendmetrics.PodMetricsClient) datalayer.EndpointFactory {
	r.dlRuntime = datalayer.NewRuntime(opts.RefreshMetricsInterval)
	if enableNewMetrics {
		return r.dlRuntime
	}
	return backendmetrics.NewPodMetricsFactory(pmc, opts.RefreshMetricsInterval)
}

// registerExtProcServer adds the ExtProcServerRunner as a Runnable to the manager.
func registerExtProcServer(mgr manager.Manager, runner *runserver.ExtProcServerRunner, logger logr.Logger) error {
	if err := mgr.Add(runner.AsRunnable(logger)); err != nil {
		setupLog.Error(err, "Failed to register ext-proc gRPC server runnable")
		return err
	}
	setupLog.Info("ExtProc server runner added to manager.")
	return nil
}

// registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.Datastore, port int, isLeader *atomic.Bool, leaderElectionEnabled bool, supporter appProtocolSupporter) error {
	srv := grpc.NewServer()
	healthPb.RegisterHealthServer(srv, &healthServer{
		logger:                logger,
		datastore:             ds,
		isLeader:              isLeader,
		leaderElectionEnabled: leaderElectionEnabled,
		supporter:             supporter,
	})
	if err := mgr.Add(
		runnable.NoLeaderElection(runnable.GRPCServer("health", srv, port))); err != nil {
		setupLog.Error(err, "Failed to register health server")
		return err
	}
	return nil
}

func extractDeploymentName(podName string) (string, error) {
	regex := regexp.MustCompile(`^(.+)-[a-z0-9]+-[a-z0-9]+$`)

	matches := regex.FindStringSubmatch(podName)
	if len(matches) == 2 {
		return matches[1], nil
	}
	return "", fmt.Errorf("failed to parse deployment name from pod name %s", podName)
}

func extractGKNN(poolName, poolGroup, poolNamespace, endpointSelector string) (*common.GKNN, error) {
	if poolName != "" {
		// Determine pool namespace: if --pool-namespace is non-empty, use it; else NAMESPACE env var; else default
		resolvedPoolNamespace := resolvePoolNamespace(poolNamespace)
		poolNamespacedName := types.NamespacedName{
			Name:      poolName,
			Namespace: resolvedPoolNamespace,
		}
		poolGroupKind := schema.GroupKind{
			Group: poolGroup,
			Kind:  "InferencePool",
		}
		return &common.GKNN{
			NamespacedName: poolNamespacedName,
			GroupKind:      poolGroupKind,
		}, nil
	}

	if endpointSelector != "" {
		// Determine EPP namespace: NAMESPACE env var; else default
		resolvedPoolNamespace := resolvePoolNamespace(poolNamespace)
		// Determine EPP name: POD_NAME env var
		eppPodNameEnv := os.Getenv("POD_NAME")
		if eppPodNameEnv == "" {
			return nil, errors.New("failed to get environment variable POD_NAME")

		}
		eppName, err := extractDeploymentName(eppPodNameEnv)
		if err != nil {
			return nil, err
		}
		return &common.GKNN{
			NamespacedName: types.NamespacedName{Namespace: resolvedPoolNamespace, Name: eppName},
			GroupKind:      schema.GroupKind{Kind: "Deployment", Group: "apps"},
		}, nil
	}
	return nil, errors.New("can't construct gknn as both pool-name and endpoint-selector are missing")
}

func resolvePoolNamespace(poolNamespace string) string {
	if poolNamespace != "" {
		return poolNamespace
	}
	if nsEnv := os.Getenv("NAMESPACE"); nsEnv != "" {
		return nsEnv
	}
	return runserver.DefaultPoolNamespace
}