kserve/pkg/apis/serving/v1alpha1/llm_inference_service_validation.go

/*
Copyright 2025 The KServe 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 v1alpha1

import (
	"context"
	"fmt"

	"k8s.io/utils/ptr"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/validation/field"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/webhook"
	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"

	"github.com/kserve/kserve/pkg/apis/serving/v1alpha2"
	"github.com/kserve/kserve/pkg/utils"
)

// +kubebuilder:webhook:path=/validate-serving-kserve-io-v1alpha1-llminferenceservice,mutating=false,failurePolicy=fail,sideEffects=None,groups=serving.kserve.io,resources=llminferenceservices,verbs=create;update,versions=v1alpha1,name=llminferenceservice.kserve-webhook-server.v1alpha1.validator,admissionReviewVersions=v1

// LLMInferenceServiceValidator is responsible for validating the LLMInferenceService resource
// when it is created, updated, or deleted.
// +kubebuilder:object:generate=false
type LLMInferenceServiceValidator struct{}

var _ webhook.CustomValidator = &LLMInferenceServiceValidator{}

func (l *LLMInferenceServiceValidator) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewWebhookManagedBy(mgr).
		For(&LLMInferenceService{}).
		WithValidator(l).
		Complete()
}

func (l *LLMInferenceServiceValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
	warnings := admission.Warnings{}
	llmSvc, err := utils.Convert[*LLMInferenceService](obj)
	if err != nil {
		return warnings, err
	}

	return warnings, l.validate(ctx, nil, llmSvc)
}

func (l *LLMInferenceServiceValidator) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
	warnings := admission.Warnings{}
	llmSvc, err := utils.Convert[*LLMInferenceService](newObj)
	if err != nil {
		return warnings, err
	}
	prev, err := utils.Convert[*LLMInferenceService](oldObj)
	if err != nil {
		return warnings, err
	}
	if llmSvc.GetDeletionTimestamp() != nil {
		return warnings, nil
	}

	return warnings, l.validate(ctx, prev, llmSvc)
}

func (l *LLMInferenceServiceValidator) ValidateDelete(_ context.Context, _ runtime.Object) (admission.Warnings, error) {
	// No validation needed for deletion
	return admission.Warnings{}, nil
}

func (l *LLMInferenceServiceValidator) validate(ctx context.Context, prev *LLMInferenceService, llmSvc *LLMInferenceService) error {
	logger := log.FromContext(ctx)
	logger.Info("Validating LLMInferenceService v1alpha1", "name", llmSvc.Name, "namespace", llmSvc.Namespace)

	var allErrs field.ErrorList

	allErrs = append(allErrs, l.validateRouterCrossFieldConstraints(llmSvc)...)
	allErrs = append(allErrs, l.validateParallelismConstraints(llmSvc)...)
	allErrs = append(allErrs, l.validateSchedulerConfig(llmSvc)...)
	allErrs = append(allErrs, l.validateScaling(llmSvc)...)
	allErrs = append(allErrs, l.validateImmutable(prev, llmSvc)...)

	if len(allErrs) == 0 {
		logger.V(2).Info("LLMInferenceService v1alpha1 is valid", "llmisvc", llmSvc)
		return nil
	}

	return apierrors.NewInvalid(
		LLMInferenceServiceGVK.GroupKind(),
		llmSvc.Name, allErrs)
}

func (l *LLMInferenceServiceValidator) validateRouterCrossFieldConstraints(llmSvc *LLMInferenceService) field.ErrorList {
	router := llmSvc.Spec.Router
	if router == nil || router.Route == nil {
		return field.ErrorList{}
	}

	routerPath := field.NewPath("spec").Child("router")
	gatewayPath := routerPath.Child("gateway")
	gwRefsPath := gatewayPath.Child("refs")
	routePath := routerPath.Child("route")
	httpRoutePath := routePath.Child("http")
	httpRouteRefs := httpRoutePath.Child("refs")
	httpRouteSpec := httpRoutePath.Child("spec")

	httpRoute := router.Route.HTTP
	if httpRoute == nil {
		return field.ErrorList{}
	}

	var allErrs field.ErrorList

	// Both refs and spec cannot be used together
	if len(httpRoute.Refs) > 0 && httpRoute.Spec != nil {
		allErrs = append(allErrs, field.Invalid(
			httpRoutePath,
			httpRoute,
			fmt.Sprintf("unsupported configuration: cannot use both custom HTTPRoute refs ('%s') and an inline route spec ('%s'); "+
				"choose one",
				httpRouteRefs, httpRouteSpec,
			),
		),
		)
	}

	// User-defined routes (refs) cannot be used with managed gateway (empty gateway config)
	if len(httpRoute.Refs) > 0 && router.Gateway != nil && len(router.Gateway.Refs) == 0 {
		allErrs = append(allErrs, field.Invalid(
			httpRouteRefs,
			httpRoute.Refs,
			fmt.Sprintf("unsupported configuration: custom HTTP routes ('%s') cannot be used with a managed gateway ('%s'); "+
				"either remove '%s' or set '%s'",
				httpRouteRefs, gatewayPath, httpRouteRefs, gwRefsPath,
			),
		))
	}

	// Managed route spec referring to the gateway cannot be used with user-defined gateway refs
	if httpRoute.Spec != nil && len(httpRoute.Spec.ParentRefs) > 0 &&
		router.Gateway != nil && len(router.Gateway.Refs) > 0 {
		allErrs = append(allErrs, field.Invalid(
			httpRoutePath.Child("spec"),
			httpRoute.Spec,
			fmt.Sprintf("unsupported configuration: managed HTTP route spec ('%s') cannot be used with custom gateway refs ('%s'); "+
				"either remove '%s' or '%s.parentRefs'",
				httpRouteSpec, gwRefsPath, gwRefsPath, httpRouteSpec,
			),
		))
	}

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateParallelismConstraints(llmSvc *LLMInferenceService) field.ErrorList {
	var allErrs field.ErrorList

	allErrs = append(allErrs, l.validateWorkloadParallelism(field.NewPath("spec"), &llmSvc.Spec.WorkloadSpec)...)

	if llmSvc.Spec.Prefill != nil {
		allErrs = append(allErrs, l.validateWorkloadParallelism(field.NewPath("spec").Child("prefill"), llmSvc.Spec.Prefill)...)
	}

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateWorkloadParallelism(basePath *field.Path, workload *WorkloadSpec) field.ErrorList {
	var allErrs field.ErrorList

	if workload.Worker != nil && workload.Parallelism == nil {
		allErrs = append(allErrs, field.Invalid(
			basePath.Child("worker"),
			workload.Worker,
			"when worker is specified, parallelism must be configured for either data parallelism or pipeline parallelism",
		))
		return allErrs
	}

	if workload.Parallelism == nil {
		return field.ErrorList{}
	}

	parallelismPath := basePath.Child("parallelism")
	parallelism := workload.Parallelism

	if workload.Worker != nil && !parallelism.IsDataParallel() && !parallelism.IsPipelineParallel() {
		allErrs = append(allErrs, field.Invalid(
			basePath.Child("worker"),
			workload.Worker,
			"when worker is specified, parallelism must be configured for either data parallelism or pipeline parallelism",
		))
	}

	if parallelism.IsPipelineParallel() && parallelism.IsDataParallel() {
		allErrs = append(allErrs, field.Invalid(
			parallelismPath,
			parallelism,
			"cannot set both pipeline parallelism and data parallelism (data or dataLocal) simultaneously",
		))
	}

	// Data and DataLocal must always be set together
	if (parallelism.Data != nil) != (parallelism.DataLocal != nil) {
		if parallelism.Data != nil && parallelism.DataLocal == nil {
			allErrs = append(allErrs, field.Invalid(
				parallelismPath.Child("dataLocal"),
				parallelism.DataLocal,
				"dataLocal must be set when data is set",
			))
		}
		if parallelism.DataLocal != nil && parallelism.Data == nil {
			allErrs = append(allErrs, field.Invalid(
				parallelismPath.Child("data"),
				parallelism.Data,
				"data must be set when dataLocal is set",
			))
		}
	}

	if parallelism.Pipeline != nil && *parallelism.Pipeline <= 0 {
		allErrs = append(allErrs, field.Invalid(
			parallelismPath.Child("pipeline"),
			*parallelism.Pipeline,
			"pipeline parallelism must be greater than 0",
		))
	}

	if parallelism.Data != nil && *parallelism.Data <= 0 {
		allErrs = append(allErrs, field.Invalid(
			parallelismPath.Child("data"),
			*parallelism.Data,
			"data parallelism must be greater than 0",
		))
	}

	if parallelism.DataLocal != nil && *parallelism.DataLocal <= 0 {
		allErrs = append(allErrs, field.Invalid(
			parallelismPath.Child("dataLocal"),
			*parallelism.DataLocal,
			"dataLocal parallelism must be greater than 0",
		))
	}

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateImmutable(prev *LLMInferenceService, curr *LLMInferenceService) field.ErrorList {
	var allErrs field.ErrorList
	if prev == nil {
		return allErrs
	}

	specPath := field.NewPath("spec")

	allErrs = append(allErrs, l.validateImmutableParallelism(specPath, prev.Spec.Parallelism, curr.Spec.Parallelism)...)
	if curr.Spec.Prefill != nil && prev.Spec.Prefill != nil {
		allErrs = append(allErrs, l.validateImmutableParallelism(specPath.Child("prefill"), prev.Spec.Prefill.Parallelism, curr.Spec.Prefill.Parallelism)...)
	}

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateImmutableParallelism(basePath *field.Path, prev *ParallelismSpec, curr *ParallelismSpec) field.ErrorList {
	var allErrs field.ErrorList
	if pSize, cSize := ptr.Deref(prev.GetSize(), 1), ptr.Deref(curr.GetSize(), 1); cSize != pSize {
		allErrs = append(allErrs, immutable(
			basePath.Child("parallelism"),
			cSize,
			fmt.Sprintf("total parallelism size is immutable, previous size %d, curr size %d", pSize, cSize),
		))
	}
	return allErrs
}

func (l *LLMInferenceServiceValidator) validateSchedulerConfig(svc *LLMInferenceService) field.ErrorList {
	var allErrs field.ErrorList

	if svc.Spec.Router == nil || svc.Spec.Router.Scheduler == nil {
		return allErrs
	}

	schedulerPath := field.NewPath("spec", "router", "scheduler")

	if svc.Spec.Router.Scheduler.Replicas != nil && *svc.Spec.Router.Scheduler.Replicas <= 0 {
		allErrs = append(allErrs, field.Invalid(schedulerPath.Child("replicas"), *svc.Spec.Router.Scheduler.Replicas, "scheduler replicas must be greater than zero"))
	}

	if svc.Spec.Router.Scheduler.Config == nil {
		return allErrs
	}

	configPath := schedulerPath.Child("config")

	if svc.Spec.Router.Scheduler.Config.Ref == nil && svc.Spec.Router.Scheduler.Config.Inline == nil {
		allErrs = append(allErrs, field.Invalid(
			configPath,
			svc.Spec.Router.Scheduler.Config,
			"either inline or ref is required",
		))
	}

	if svc.Spec.Router.Scheduler.Config.Inline != nil && svc.Spec.Router.Scheduler.Config.Ref != nil {
		allErrs = append(allErrs, field.Invalid(
			configPath,
			svc.Spec.Router.Scheduler.Config,
			"both inline and ref are set, either specify inline or ref",
		))
	}

	if svc.Spec.Router.Scheduler.Config.Inline != nil && len(svc.Spec.Router.Scheduler.Config.Inline.Raw) < 3 /* we expect at least a few characters '{...}' */ {
		allErrs = append(allErrs, field.Invalid(
			configPath.Child("inline"),
			svc.Spec.Router.Scheduler.Config.Inline,
			"inline configuration is invalid",
		))
	}

	if svc.Spec.Router.Scheduler.Config.Ref != nil {
		if svc.Spec.Router.Scheduler.Config.Ref.Name == "" {
			allErrs = append(allErrs, field.Invalid(
				configPath.Child("ref", "name"),
				svc.Spec.Router.Scheduler.Config.Ref,
				"name is empty",
			))
		}
	}

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateScaling(llmSvc *LLMInferenceService) field.ErrorList {
	var allErrs field.ErrorList

	// Validate scaling on the main (decode) workload
	allErrs = append(allErrs, l.validateWorkloadScaling(field.NewPath("spec"), &llmSvc.Spec.WorkloadSpec)...)

	// Validate scaling on the prefill workload, if present
	if llmSvc.Spec.Prefill != nil {
		allErrs = append(allErrs, l.validateWorkloadScaling(field.NewPath("spec").Child("prefill"), llmSvc.Spec.Prefill)...)
	}

	// Validate actuator backend consistency across decode and prefill.
	// Convert to the hub (v1alpha2) type and delegate to its exported validator so
	// the consistency rules live in exactly one place.
	decode := convertWorkloadSpecToV1Alpha2(&llmSvc.Spec.WorkloadSpec)
	var prefill *v1alpha2.WorkloadSpec
	if llmSvc.Spec.Prefill != nil {
		p := convertWorkloadSpecToV1Alpha2(llmSvc.Spec.Prefill)
		prefill = &p
	}
	allErrs = append(allErrs, v1alpha2.ValidateActuatorConsistency(&decode, prefill)...)

	return allErrs
}

func (l *LLMInferenceServiceValidator) validateWorkloadScaling(basePath *field.Path, workload *WorkloadSpec) field.ErrorList {
	// Convert to the hub (v1alpha2) type and delegate to its exported validator so
	// the scaling rules live in exactly one place.
	w := convertWorkloadSpecToV1Alpha2(workload)
	return v1alpha2.ValidateWorkloadScaling(basePath, &w)
}

// immutable returns a *Error indicating "unsupported mutation".
// This is used to report unsupported mutation of values.
func immutable(path *field.Path, value interface{}, detail string) *field.Error {
	return &field.Error{Type: field.ErrorTypeNotSupported, Field: path.String(), BadValue: value, Detail: detail}
}