Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 8 additions & 18 deletions cmd/epp/runner/runner.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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")
Expand Down
15 changes: 14 additions & 1 deletion pkg/epp/controller/inferencepool_reconciler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"}
Expand Down Expand Up @@ -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
}

Expand Down
5 changes: 3 additions & 2 deletions pkg/epp/datalayer/endpoint_pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down
20 changes: 14 additions & 6 deletions pkg/epp/datastore/datastore.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 ///
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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()
}
17 changes: 15 additions & 2 deletions pkg/epp/datastore/datastore_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 {
Expand Down Expand Up @@ -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}
Expand Down Expand Up @@ -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()
Expand Down
30 changes: 19 additions & 11 deletions pkg/epp/server/options.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
//
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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")
}
Expand Down
7 changes: 4 additions & 3 deletions pkg/epp/util/pool/pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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,
Expand Down
3 changes: 2 additions & 1 deletion test/integration/epp/harness.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
Loading