defaults.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 loader

import (
	"fmt"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	configapi "github.com/llm-d/llm-d-router/apix/config/v1alpha1"
	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/registry"
	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
	extractormetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/metrics"
	sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/anthropic"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/openai"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/vllmhttp"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/picker/maxscore"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/profilehandler/single"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/kvcacheutilization"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/prefix"
	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/queuedepth"
)

// DefaultScorerWeight is the weight used for scorers referenced in the configuration without explicit weights.
const DefaultScorerWeight = 1.0

var defaultScorerWeight = DefaultScorerWeight

func loadDefaultConfig() *configapi.EndpointPickerConfig {
	queueScorerWeight := 2.0
	kvCacheUtilizationScorerWeight := 2.0
	prefixCacheScorerWeight := 3.0
	return &configapi.EndpointPickerConfig{
		TypeMeta: metav1.TypeMeta{
			APIVersion: configapi.GroupVersion.String(),
			Kind:       "EndpointPickerConfig",
		},
		FeatureGates: []string{}, // Data layer is now enabled by default (no feature gate needed)
		Plugins: []configapi.PluginSpec{
			{
				Type: queuedepth.QueueScorerType,
			},
			{
				Type: kvcacheutilization.KvCacheUtilizationScorerType,
			},
			{
				Type: prefix.PrefixCacheScorerPluginType,
			},
			{
				Type: sourcemetrics.MetricsDataSourceType,
			},
			{
				Type: extractormetrics.MetricsExtractorType,
			},
		},
		SchedulingProfiles: []configapi.SchedulingProfile{
			{
				Name: "default",
				Plugins: []configapi.SchedulingPlugin{
					{
						PluginRef: queuedepth.QueueScorerType,
						Weight:    &queueScorerWeight,
					},
					{
						PluginRef: kvcacheutilization.KvCacheUtilizationScorerType,
						Weight:    &kvCacheUtilizationScorerWeight,
					},
					{
						PluginRef: prefix.PrefixCacheScorerPluginType,
						Weight:    &prefixCacheScorerWeight,
					},
				},
			},
		},
		DataLayer: &configapi.DataLayerConfig{
			Sources: []configapi.DataLayerSource{
				{
					PluginRef: sourcemetrics.MetricsDataSourceType,
					Extractors: []configapi.DataLayerExtractor{
						{PluginRef: extractormetrics.MetricsExtractorType},
					},
				},
			},
		},
	}
}

// applyStaticDefaults sanitizes the configuration object before plugin instantiation.
// It handles "Static" defaults: simple structural changes to the API object that do not require access to the plugin
// registry.
func applyStaticDefaults(cfg *configapi.EndpointPickerConfig) {
	for idx, pluginConfig := range cfg.Plugins {
		if pluginConfig.Name == "" {
			cfg.Plugins[idx].Name = pluginConfig.Type
		}
	}

	if cfg.FeatureGates == nil {
		cfg.FeatureGates = configapi.FeatureGates{}
	}
}

// applySystemDefaults injects required components that were omitted from the config.
// It handles "System" defaults: logic that requires inspecting instantiated plugins (via the handle) to ensure the
// system graph is complete.
func applySystemDefaults(cfg *configapi.EndpointPickerConfig, handle fwkplugin.Handle) error {
	allPlugins := handle.GetAllPluginsWithNames()
	if err := ensureSchedulingLayer(cfg, handle, allPlugins); err != nil {
		return fmt.Errorf("failed to apply scheduling system defaults: %w", err)
	}
	if err := ensureFlowControlLayer(cfg, handle, allPlugins); err != nil {
		return fmt.Errorf("failed to apply flow control system defaults: %w", err)
	}
	if err := ensureParsers(cfg, handle, allPlugins); err != nil {
		return fmt.Errorf("failed to apply parser defaults: %w", err)
	}
	if err := ensureSaturationDetector(cfg, handle, allPlugins); err != nil {
		return fmt.Errorf("failed to apply saturation detector defaults: %w", err)
	}
	if err := ensureDataLayer(cfg, handle, allPlugins); err != nil {
		return fmt.Errorf("failed to apply data layer defaults: %w", err)
	}
	return nil
}

// ensureSchedulingLayer guarantees that the scheduling subsystem is structurally complete.
// It ensures a valid profile exists and injects missing architectural components (like Pickers and ProfileHandlers) if
// they are not explicitly configured.
func ensureSchedulingLayer(
	cfg *configapi.EndpointPickerConfig,
	handle fwkplugin.Handle,
	allPlugins map[string]fwkplugin.Plugin,
) error {
	if len(cfg.SchedulingProfiles) == 0 {
		defaultProfile := configapi.SchedulingProfile{Name: "default"}
		// Auto-populate the default profile with all Filter, Scorer, and Picker plugins found.
		for name, p := range allPlugins {
			switch p.(type) {
			case fwksched.Filter, fwksched.Scorer, fwksched.Picker:
				defaultProfile.Plugins = append(defaultProfile.Plugins, configapi.SchedulingPlugin{PluginRef: name})
			}
		}
		cfg.SchedulingProfiles = []configapi.SchedulingProfile{defaultProfile}
	}

	// If there is only 1 profile and no handler is explicitly configured, use the SingleProfileHandler.
	if len(cfg.SchedulingProfiles) == 1 {
		hasHandler := false
		for _, p := range allPlugins {
			if _, ok := p.(fwksched.ProfileHandler); ok {
				hasHandler = true
				break
			}
		}
		if !hasHandler {
			if err := registerDefaultPlugin(cfg, handle, single.SingleProfileHandlerType); err != nil {
				return err
			}
		}
	}

	// Find or Create a default MaxScorePicker to reuse across profiles.
	var maxScorePickerName string
	for name, p := range allPlugins {
		if _, ok := p.(fwksched.Picker); ok {
			maxScorePickerName = name
			break
		}
	}

	if maxScorePickerName == "" {
		if err := registerDefaultPlugin(cfg, handle, maxscore.MaxScorePickerType); err != nil {
			return err
		}
		maxScorePickerName = maxscore.MaxScorePickerType
	}

	for i, prof := range cfg.SchedulingProfiles {
		hasPicker := false
		for j, pluginRef := range prof.Plugins {
			p := handle.Plugin(pluginRef.PluginRef)

			if _, ok := p.(fwksched.Scorer); ok && pluginRef.Weight == nil {
				cfg.SchedulingProfiles[i].Plugins[j].Weight = &defaultScorerWeight
			}

			if _, ok := p.(fwksched.Picker); ok {
				hasPicker = true
			}
		}

		if !hasPicker {
			cfg.SchedulingProfiles[i].Plugins = append(
				cfg.SchedulingProfiles[i].Plugins,
				configapi.SchedulingPlugin{PluginRef: maxScorePickerName},
			)
		}
	}

	return nil
}

// ensureFlowControlLayer guarantees that the flow control subsystem is structurally complete.
func ensureFlowControlLayer(cfg *configapi.EndpointPickerConfig, handle fwkplugin.Handle, allPlugins map[string]fwkplugin.Plugin) error {
	if _, ok := allPlugins[registry.DefaultOrderingPolicyRef]; !ok {
		if err := registerDefaultPlugin(cfg, handle, registry.DefaultOrderingPolicyRef); err != nil {
			return err
		}
	}
	if _, ok := allPlugins[registry.DefaultFairnessPolicyRef]; !ok {
		if err := registerDefaultPlugin(cfg, handle, registry.DefaultFairnessPolicyRef); err != nil {
			return err
		}
	}
	if _, ok := allPlugins[registry.DefaultUsageLimitPolicyRef]; !ok {
		if err := registerDefaultPlugin(cfg, handle, registry.DefaultUsageLimitPolicyRef); err != nil {
			return err
		}
	}
	return nil
}

// ensureParsers guarantees that at least one parser is configured.
// If no parsers are configured, the openAI parser is configured by default.
func ensureParsers(
	cfg *configapi.EndpointPickerConfig,
	handle fwkplugin.Handle,
	allPlugins map[string]fwkplugin.Plugin,
) error {
	if cfg.RequestHandler == nil {
		cfg.RequestHandler = &configapi.RequestHandlerConfig{}
	}
	if len(cfg.RequestHandler.Parsers) == 0 {
		cfg.RequestHandler.Parsers = []configapi.ParserConfig{
			{PluginRef: openai.OpenAIParserType},
			{PluginRef: anthropic.AnthropicParserType},
			{PluginRef: vllmhttp.VllmHTTPParserType},
		}
	}
	for _, pc := range cfg.RequestHandler.Parsers {
		if _, ok := allPlugins[pc.PluginRef]; !ok {
			if err := registerDefaultPlugin(cfg, handle, pc.PluginRef); err != nil {
				return err
			}
		}
	}
	return nil
}

// ensureSaturationDetector guarantees that saturation detector is configured.
// If the saturation detector is not set, the utilization detector is configured by default.
func ensureSaturationDetector(
	cfg *configapi.EndpointPickerConfig,
	handle fwkplugin.Handle,
	allPlugins map[string]fwkplugin.Plugin,
) error {
	if cfg.FlowControl == nil {
		cfg.FlowControl = &configapi.FlowControlConfig{}
	}
	sdConfig := cfg.FlowControl.SaturationDetector
	if sdConfig == nil {
		sdConfig = &configapi.SaturationDetectorConfig{
			PluginRef: utilization.UtilizationDetectorType,
		}
		cfg.FlowControl.SaturationDetector = sdConfig
	}
	if sdConfig.PluginRef == "" {
		sdConfig.PluginRef = utilization.UtilizationDetectorType
	}

	if sdConfig.PluginRef == utilization.UtilizationDetectorType {
		if _, ok := allPlugins[sdConfig.PluginRef]; !ok {
			if err := registerDefaultPlugin(cfg, handle, utilization.UtilizationDetectorType); err != nil {
				return err
			}
		}
	}

	if sd, ok := allPlugins[sdConfig.PluginRef]; ok {
		if _, isFilter := sd.(fwksched.Filter); isFilter {
			injectFilterIntoProfiles(cfg.SchedulingProfiles, sdConfig.PluginRef)
		}
	}
	return nil
}

func injectFilterIntoProfiles(profiles []configapi.SchedulingProfile, pluginRef string) {
	for i := range profiles {
		found := false
		for _, p := range profiles[i].Plugins {
			if p.PluginRef == pluginRef {
				found = true
				break
			}
		}
		if !found {
			profiles[i].Plugins = append(profiles[i].Plugins, configapi.SchedulingPlugin{PluginRef: pluginRef})
		}
	}
}

// ensureDataLayer additively injects the default metrics source and extractor unless opted out.
// Unlike other ensureXxx functions, it checks for explicit opt-out via InjectDefaults and avoids
// double-injection when the metrics source is already present in a user-supplied config.
func ensureDataLayer(cfg *configapi.EndpointPickerConfig, handle fwkplugin.Handle, allPlugins map[string]fwkplugin.Plugin) error {
	if cfg.DataLayer != nil && cfg.DataLayer.InjectDefaults != nil && !*cfg.DataLayer.InjectDefaults {
		return nil
	}
	if cfg.DataLayer != nil && hasSourceOfType(cfg.DataLayer, sourcemetrics.MetricsDataSourceType) {
		return nil
	}

	if _, ok := allPlugins[sourcemetrics.MetricsDataSourceType]; !ok {
		if err := registerDefaultPlugin(cfg, handle, sourcemetrics.MetricsDataSourceType); err != nil {
			return err
		}
	}
	if _, ok := allPlugins[extractormetrics.MetricsExtractorType]; !ok {
		if err := registerDefaultPlugin(cfg, handle, extractormetrics.MetricsExtractorType); err != nil {
			return err
		}
	}

	if cfg.DataLayer == nil {
		cfg.DataLayer = &configapi.DataLayerConfig{}
	}
	cfg.DataLayer.Sources = append(cfg.DataLayer.Sources, configapi.DataLayerSource{
		PluginRef: sourcemetrics.MetricsDataSourceType,
		Extractors: []configapi.DataLayerExtractor{{
			PluginRef: extractormetrics.MetricsExtractorType,
		}},
	})

	return nil
}

func hasSourceOfType(dl *configapi.DataLayerConfig, pluginType string) bool {
	for _, s := range dl.Sources {
		if s.PluginRef == pluginType {
			return true
		}
	}
	return false
}

// registerDefaultPlugin instantiates a plugin with empty configuration (defaults) and adds it to both the handle and
// the config spec.
func registerDefaultPlugin(
	cfg *configapi.EndpointPickerConfig,
	handle fwkplugin.Handle,
	pluginType string,
) error {
	name := pluginType
	factory, ok := fwkplugin.Registry[pluginType]
	if !ok {
		return fmt.Errorf("plugin type '%s' not found in registry", pluginType)
	}

	plugin, err := factory(name, nil, handle) // default plugins have no parameters
	if err != nil {
		return fmt.Errorf("failed to instantiate default plugin '%s': %w", name, err)
	}

	handle.AddPlugin(name, plugin)
	cfg.Plugins = append(cfg.Plugins, configapi.PluginSpec{
		Name: name,
		Type: pluginType,
	})

	return nil
}