feat(standalone): use typed labels.Selector for endpoint selector

Replace the string-based --endpoint-selector flag with
Kubernetes-native labels.Selector to support set-based
label expressions (e.g., 'env in (prod,staging),tier!=frontend',
'key,!deprecated') in standalone mode.

- EndpointSelector type: string -> labels.Selector
- Complete(): parse flag string via labels.Parse()
- EndpointPool.Selector: map[string]string -> labels.Selector
- PoolLabelsMatch simplified with native Selector.Matches()
- podResyncAll passes Selector directly to K8s client
- All callers and tests updated for new type

Signed-off-by: shichaooutlook <shichao.outlook@gmail.com>

Author: shichaooutlook

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")

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
 	}
 

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,

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()
+}

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()

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")
 		}

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,

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)