diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go index 9f210bf66f..0ed438cd41 100644 --- a/cmd/epp/runner/runner.go +++ b/cmd/epp/runner/runner.go @@ -287,7 +287,7 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op 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. + startCrdReconcilers := opts.EndpointSelector == nil // If endpointSelector is nil, 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") @@ -433,38 +433,31 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op func NewEndpointPoolFromOptions( namespace string, name string, - endpointSelector string, + endpointSelector labels.Selector, endpointTargetPorts []int, ) (*datalayer.EndpointPool, error) { - // namespace is from epp namespace in standalone mode without inference api support if namespace == "" { return nil, errors.New("namespace must not be empty") } - // name is from epp name in standalone mode without inference api support if name == "" { return nil, errors.New("name must not be empty") } - if endpointSelector == "" { - return nil, errors.New("endpoint selector must not be empty") + if endpointSelector == nil { + return nil, errors.New("endpoint selector must not be nil") } 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.Selector = endpointSelector 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) { + startCrdReconcilers bool, namespace, name string, endpointSelector labels.Selector, endpointTargetPorts []int) (datastore.Datastore, error) { if startCrdReconcilers { return datastore.NewDatastore(ctx, epFactory, modelServerMetricsPort), nil @@ -750,9 +743,8 @@ func extractDeploymentName(podName string) (string, error) { return "", fmt.Errorf("failed to parse deployment name from pod name %s", podName) } -func extractGKNN(poolName, poolGroup, poolNamespace, endpointSelector string) (*common.GKNN, error) { +func extractGKNN(poolName, poolGroup, poolNamespace string, endpointSelector labels.Selector) (*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, @@ -768,10 +760,8 @@ func extractGKNN(poolName, poolGroup, poolNamespace, endpointSelector string) (* }, nil } - if endpointSelector != "" { - // Determine EPP namespace: NAMESPACE env var; else default + if endpointSelector != nil { 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") diff --git a/pkg/epp/controller/inferencepool_reconciler_test.go b/pkg/epp/controller/inferencepool_reconciler_test.go index 88186da138..6fb1020dda 100644 --- a/pkg/epp/controller/inferencepool_reconciler_test.go +++ b/pkg/epp/controller/inferencepool_reconciler_test.go @@ -24,6 +24,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" @@ -40,6 +41,18 @@ import ( testutil "github.com/llm-d/llm-d-router/pkg/epp/util/testing" ) +var endpointPoolCmpOpts = []cmp.Option{ + cmp.Comparer(func(a, b labels.Selector) bool { + if a == nil && b == nil { + return true + } + if a == nil || b == nil { + return false + } + return a.String() == b.String() + }), +} + var ( selectorV1 = map[string]string{"app": "vllm_v1"} selectorV2 = map[string]string{"app": "vllm_v2"} @@ -181,7 +194,7 @@ type diffStoreParams struct { func diffStore(store datastore.Datastore, params diffStoreParams) string { gotPool, _ := store.PoolGet() - if diff := cmp.Diff(params.wantPool, gotPool); diff != "" { + if diff := cmp.Diff(params.wantPool, gotPool, endpointPoolCmpOpts...); diff != "" { return "inferencePool:" + diff } diff --git a/pkg/epp/datalayer/endpoint_pool.go b/pkg/epp/datalayer/endpoint_pool.go index 59f7cc99d9..94afa3f01c 100644 --- a/pkg/epp/datalayer/endpoint_pool.go +++ b/pkg/epp/datalayer/endpoint_pool.go @@ -17,11 +17,12 @@ limitations under the License. package datalayer import ( + "k8s.io/apimachinery/pkg/labels" v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1" ) type EndpointPool struct { - Selector map[string]string + Selector labels.Selector TargetPorts []int Namespace string Name string @@ -31,7 +32,7 @@ type EndpointPool struct { // NewEndpointPool creates and returns a new empty instance of EndpointPool. func NewEndpointPool(namespace string, name string) *EndpointPool { return &EndpointPool{ - Selector: make(map[string]string), + Selector: labels.Everything(), TargetPorts: []int{}, Namespace: namespace, Name: name, diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go index 9418c0a620..09b62dd6bd 100644 --- a/pkg/epp/datastore/datastore.go +++ b/pkg/epp/datastore/datastore.go @@ -168,7 +168,7 @@ func (ds *datastore) PoolSet(ctx context.Context, reader client.Reader, endpoint oldEndpointPool := ds.pool ds.pool = endpointPool - selectorChanged := oldEndpointPool == nil || !labels.Equals(oldEndpointPool.Selector, endpointPool.Selector) + selectorChanged := oldEndpointPool == nil || !selectorEqual(oldEndpointPool.Selector, endpointPool.Selector) targetPortsChanged := oldEndpointPool != nil && !slices.Equal(oldEndpointPool.TargetPorts, endpointPool.TargetPorts) if selectorChanged || targetPortsChanged { @@ -207,12 +207,10 @@ func (ds *datastore) PoolHasSynced() bool { func (ds *datastore) PoolLabelsMatch(podLabels map[string]string) bool { ds.mu.RLock() defer ds.mu.RUnlock() - if ds.pool == nil { + if ds.pool == nil || ds.pool.Selector == nil { return false } - poolSelector := labels.SelectorFromSet(ds.pool.Selector) - podSet := labels.Set(podLabels) - return poolSelector.Matches(podSet) + return ds.pool.Selector.Matches(labels.Set(podLabels)) } // /// InferenceObjective APIs /// @@ -415,7 +413,7 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err logger := log.FromContext(ctx) podList := &corev1.PodList{} if err := reader.List(ctx, podList, &client.ListOptions{ - LabelSelector: labels.SelectorFromSet(ds.pool.Selector), + LabelSelector: ds.pool.Selector, Namespace: ds.pool.Namespace, }); err != nil { return fmt.Errorf("failed to list pods - %w", err) @@ -487,3 +485,13 @@ func createEndpointNamespacedName(pod *corev1.Pod, idx int) types.NamespacedName Namespace: pod.Namespace, } } + +func selectorEqual(a, b labels.Selector) bool { + if a == nil && b == nil { + return true + } + if a == nil || b == nil { + return false + } + return a.String() == b.String() +} diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go index 16e8346519..c9c23930e2 100644 --- a/pkg/epp/datastore/datastore_test.go +++ b/pkg/epp/datastore/datastore_test.go @@ -31,6 +31,7 @@ import ( "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" @@ -46,6 +47,18 @@ import ( testutil "github.com/llm-d/llm-d-router/pkg/epp/util/testing" ) +var endpointPoolCmpOpts = []cmp.Option{ + cmp.Comparer(func(a, b labels.Selector) bool { + if a == nil && b == nil { + return true + } + if a == nil || b == nil { + return false + } + return a.String() == b.String() + }), +} + // mockEndpointFactory is a minimal EndpointFactory for EndpointUpsert/Delete tests. // When returnNil is true, NewEndpoint returns nil (simulating a duplicate-start race). type mockEndpointFactory struct { @@ -78,7 +91,7 @@ func (f *mockEndpointFactory) updateEvents() []fwkdl.Endpoint { func TestPoolGet_NoDeadlockWithConcurrentWrite(t *testing.T) { pool := &datalayer.EndpointPool{ Namespace: "default", - Selector: map[string]string{"app": "vllm"}, + Selector: labels.SelectorFromSet(labels.Set{"app": "vllm"}), TargetPorts: []int{8000}, } ds := &datastore{pool: pool} @@ -161,7 +174,7 @@ func TestPool(t *testing.T) { if diff := cmp.Diff(tt.wantErr, gotErr, cmpopts.EquateErrors()); diff != "" { t.Errorf("Unexpected error diff (+got/-want): %s", diff) } - if diff := cmp.Diff(poolutil.InferencePoolToEndpointPool(tt.wantPool), gotPool); diff != "" { + if diff := cmp.Diff(poolutil.InferencePoolToEndpointPool(tt.wantPool), gotPool, endpointPoolCmpOpts...); diff != "" { t.Errorf("Unexpected pool diff (+got/-want): %s", diff) } gotSynced := ds.PoolHasSynced() diff --git a/pkg/epp/server/options.go b/pkg/epp/server/options.go index dfc7a57cfe..b3da2459f5 100644 --- a/pkg/epp/server/options.go +++ b/pkg/epp/server/options.go @@ -25,6 +25,7 @@ import ( "github.com/spf13/pflag" "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/sets" "github.com/llm-d/llm-d-router/pkg/common/observability/logging" @@ -72,9 +73,9 @@ type Options struct { // // Endpoints (in lieu of using an InferencePool for service discovery). // - EndpointSelector string // Selector to filter model server pods on, only 'key=value' pairs are supported. (TODO: k8s.Selector, pflag.StringSlice?) - EndpointTargetPorts []int // Target ports of model server pods. - DisableEndpointSubsetFilter bool // Disables respecting destination endpoint subset metadata in EPP. + EndpointSelector labels.Selector // Parsed selector to filter model server pods on. Set via --endpoint-selector flag and parsed in Complete(). + EndpointTargetPorts []int // Target ports of model server pods. + DisableEndpointSubsetFilter bool // Disables respecting destination endpoint subset metadata in EPP. // // MSP metrics scraping. // @@ -111,7 +112,8 @@ type Options struct { ConfigText string // The configuration specified as text, in lieu of a file. // internal - fs *pflag.FlagSet // FlagSet used in AddFlags() and consulted in Validate() + fs *pflag.FlagSet // FlagSet used in AddFlags() and consulted in Validate() + endpointSelectorStr string // Raw string from --endpoint-selector flag, parsed to EndpointSelector in Complete() } // NewOptions returns a new Options struct initialized with the default values. @@ -158,9 +160,10 @@ func (opts *Options) AddFlags(fs *pflag.FlagSet) { fs.StringVar(&opts.PoolNamespace, "pool-namespace", opts.PoolNamespace, "Namespace of the InferencePool this Endpoint Picker is associated with.") fs.StringVar(&opts.PoolName, "pool-name", opts.PoolName, "Name of the InferencePool this Endpoint Picker is associated with.") - fs.StringVar(&opts.EndpointSelector, "endpoint-selector", opts.EndpointSelector, - "Selector to filter model server pods on, only 'key=value' pairs are supported. "+ - "Format: a comma-separated list of key=value pairs without whitespace (e.g., 'app=vllm-qwen3-32b,env=prod').") + fs.StringVar(&opts.endpointSelectorStr, "endpoint-selector", opts.endpointSelectorStr, + "Selector to filter model server pods on. "+ + "Supports Kubernetes label selector syntax: equality-based (e.g., 'app=vllm,env=prod'), "+ + "set-based (e.g., 'env in (prod,staging),tier!=frontend'), and existence (e.g., 'key,!deprecated').") fs.IntSliceVar(&opts.EndpointTargetPorts, "endpoint-target-ports", opts.EndpointTargetPorts, "Target ports of model server pods. "+ "Format: a comma-separated list of numbers without whitespace (e.g., '3000,3001,3002').") fs.BoolVar(&opts.DisableEndpointSubsetFilter, "disable-endpoint-subset-filter", opts.DisableEndpointSubsetFilter, @@ -222,8 +225,13 @@ func (opts *Options) AddFlags(fs *pflag.FlagSet) { } func (opts *Options) Complete() error { - // TODO: postprocessing or command line arguments. For example, convert EndpointSelector - // from raw string to k8s.LabelSelector, load ConfigFile into ConfigText, etc. + if opts.endpointSelectorStr != "" { + selector, err := labels.Parse(opts.endpointSelectorStr) + if err != nil { + return fmt.Errorf("invalid endpoint-selector %q: %w", opts.endpointSelectorStr, err) + } + opts.EndpointSelector = selector + } opts.EndpointTargetPorts = removeDuplicatePorts(opts.EndpointTargetPorts) @@ -269,10 +277,10 @@ func (opts *Options) Complete() error { } func (opts *Options) Validate() error { - if (opts.PoolName != "" && opts.EndpointSelector != "") || (opts.PoolName == "" && opts.EndpointSelector == "") { + if (opts.PoolName != "" && opts.EndpointSelector != nil) || (opts.PoolName == "" && opts.EndpointSelector == nil) { return errors.New("either pool-name or endpoint-selector must be set") } - if opts.EndpointSelector != "" { + if opts.EndpointSelector != nil { if len(opts.EndpointTargetPorts) == 0 || len(opts.EndpointTargetPorts) > 8 { return fmt.Errorf("flag %q should have length from 1 to 8", "endpoint-target-ports") } diff --git a/pkg/epp/util/pool/pool.go b/pkg/epp/util/pool/pool.go index 706e711f09..3f20cfa732 100644 --- a/pkg/epp/util/pool/pool.go +++ b/pkg/epp/util/pool/pool.go @@ -17,6 +17,7 @@ limitations under the License. package pool import ( + "k8s.io/apimachinery/pkg/labels" v1 "sigs.k8s.io/gateway-api-inference-extension/api/v1" "github.com/llm-d/llm-d-router/pkg/epp/datalayer" @@ -31,12 +32,12 @@ func InferencePoolToEndpointPool(inferencePool *v1.InferencePool) *datalayer.End targetPorts = append(targetPorts, int(p.Number)) } - selector := make(map[string]string, len(inferencePool.Spec.Selector.MatchLabels)) + selectorMap := make(map[string]string, len(inferencePool.Spec.Selector.MatchLabels)) for k, v := range inferencePool.Spec.Selector.MatchLabels { - selector[string(k)] = string(v) + selectorMap[string(k)] = string(v) } endpointPool := &datalayer.EndpointPool{ - Selector: selector, + Selector: labels.SelectorFromSet(labels.Set(selectorMap)), TargetPorts: targetPorts, Namespace: inferencePool.Namespace, Name: inferencePool.Name, diff --git a/test/integration/epp/harness.go b/test/integration/epp/harness.go index 547aaed034..f044b05236 100644 --- a/test/integration/epp/harness.go +++ b/test/integration/epp/harness.go @@ -37,6 +37,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" metricsutils "k8s.io/component-base/metrics/testutil" @@ -204,7 +205,7 @@ func NewTestHarness(ctx context.Context, t *testing.T, opts ...HarnessOption) *T eppOptions := defaultEppServerOptions(t, testNamespaceName, configText) if config.runMode == modeStandalone && config.standaloneStrategy == strategyNoCRD { // Only standalone EPP without crd need to set the EndpointSelector. - eppOptions.EndpointSelector = "app=" + testPoolName + eppOptions.EndpointSelector = labels.SelectorFromSet(labels.Set{"app": testPoolName}) } // Shorten the Prometheus refresh interval so WaitForReadyPodsMetric (10s timeout)