Skip to content

feat(extractor): add spoke-epp engine type for multi-cluster #1928

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
yehuditkerido wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat(extractor): add spoke-epp engine type for multi-cluster #1928

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
15 changes: 13 additions & 2 deletions pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,10 +198,21 @@ func (ext *Extractor) Extract(ctx context.Context, in fwkdl.PollInput[sourcemetr
return nil
}

// getEngineTypeFromEndpoint extracts the engine type from endpoint metadata labels.
// getEngineTypeFromEndpoint extracts the engine type from endpoint metadata.

@elevran elevran Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can the same be achieved by defining a custom set of metrics mapping instead of a new type?

// Cluster endpoints (identified by empty PodName) automatically use SpokeEPPEngineType.

@elevran elevran Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how are cluster endpoints named - which field is used? is the reliance on "magic" (empty PodName means its a cluster endpoint) the right long term approach or should we use an endpoint attribute instead (the file based discovery can populate the attribute)?

// For pod endpoints, the engine type is determined by labels or falls back to default.
func getEngineTypeFromEndpoint(ep fwkdl.Endpoint, labelKey string) string {
meta := ep.GetMetadata()
if meta == nil || meta.Labels == nil {
if meta == nil {
return DefaultEngineType
}

// Cluster endpoints have empty PodName (hostname-based address from file-discovery)
if meta.PodName == "" {
return SpokeEPPEngineType
}

if meta.Labels == nil {
return DefaultEngineType
}

Expand Down
114 changes: 105 additions & 9 deletions pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,8 @@ func TestExtractorMultiEngine(t *testing.T) {

// Case 1: Engine = vllm (uses default)
epVllm := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
Labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
PodName: "vllm-pod",

@elevran elevran Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the empty == cluster seems fragile. Case in point, it was not an assumption that holds for the current code base.

Labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
}, nil)
_ = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{Payload: data, Endpoint: epVllm})
if epVllm.GetMetrics().WaitingQueueSize != 10 {
Expand All @@ -258,7 +259,8 @@ func TestExtractorMultiEngine(t *testing.T) {

// Case 2: Engine = sglang (uses specific)
epSgl := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
Labels: map[string]string{DefaultEngineTypeLabelKey: "sglang"},
PodName: "sglang-pod",
Labels: map[string]string{DefaultEngineTypeLabelKey: "sglang"},
}, nil)
_ = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{Payload: data, Endpoint: epSgl})
if epSgl.GetMetrics().WaitingQueueSize != 20 {
Expand Down Expand Up @@ -287,23 +289,97 @@ func TestBackwardCompatibility(t *testing.T) {
},
}

// Case 1: No labels at all
epNone := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{Labels: nil}, nil)
// Case 1: No labels at all (but with PodName, so treated as pod)
epNone := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{PodName: "test-pod", Labels: nil}, nil)
_ = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{Payload: data, Endpoint: epNone})
if epNone.GetMetrics().WaitingQueueSize != 100 {
t.Errorf("no labels: expected 100, got %v", epNone.GetMetrics().WaitingQueueSize)
}

// Case 2: Different label key or unknown value
epUnknown := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
Labels: map[string]string{DefaultEngineTypeLabelKey: "unknown-engine"},
PodName: "unknown-pod",
Labels: map[string]string{DefaultEngineTypeLabelKey: "unknown-engine"},
}, nil)
_ = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{Payload: data, Endpoint: epUnknown})
if epUnknown.GetMetrics().WaitingQueueSize != 100 {
t.Errorf("unknown label: expected 100, got %v", epUnknown.GetMetrics().WaitingQueueSize)
}
}

func TestSpokeEPPClusterEndpointExtraction(t *testing.T) {
ctx := context.Background()

registry := NewMappingRegistry()

// Register spoke-epp mapping for cluster endpoints
spokeMapping, err := NewMapping(

@elevran elevran Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the empty mappings indicate that EPP metrics and endpoint metrics are the same shape. I would strongly recommend a new Extractor for EPP endpoints.
It can fill in the existing metrics fields or create a new endpoint attribute with only the relevant fields.
I would suspect that Cluster endpoints have different attributes than inference server endpoints (not just metrics).

"llm_d_epp_average_queue_size",
"llm_d_epp_average_running_requests",
"llm_d_epp_average_kv_cache_utilization",
"", "")
if err != nil {
t.Fatalf("failed to create spoke-epp mapping: %v", err)
}
if err := registry.Register(SpokeEPPEngineType, spokeMapping); err != nil {
t.Fatalf("failed to register spoke-epp mapping: %v", err)
}

// Register default vLLM mapping for pods
vllmMapping, _ := NewMapping("vllm:num_requests_waiting", "", "", "", "")
_ = registry.Register(DefaultEngineType, vllmMapping)

extractor, _ := NewCoreMetricsExtractor(registry, "")

// Spoke EPP metrics (aggregated pool averages)
clusterData := sourcemetrics.PrometheusMetricMap{
"llm_d_epp_average_queue_size": &dto.MetricFamily{
Type: dto.MetricType_GAUGE.Enum(),
Metric: []*dto.Metric{
{Gauge: &dto.Gauge{Value: ptr.To(15.0)}},
},
},
"llm_d_epp_average_running_requests": &dto.MetricFamily{
Type: dto.MetricType_GAUGE.Enum(),
Metric: []*dto.Metric{
{Gauge: &dto.Gauge{Value: ptr.To(8.0)}},
},
},
"llm_d_epp_average_kv_cache_utilization": &dto.MetricFamily{
Type: dto.MetricType_GAUGE.Enum(),
Metric: []*dto.Metric{
{Gauge: &dto.Gauge{Value: ptr.To(0.72)}},
},
},
}

// Cluster endpoint: empty PodName (hostname-based address from file-discovery)
clusterEp := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
PodName: "", // Empty PodName identifies cluster endpoints
Address: "spoke-us-east.example.com",
Labels: map[string]string{"region": "us-east"},
}, nil)

err = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{
Payload: clusterData,
Endpoint: clusterEp,
})
if err != nil {
t.Fatalf("unexpected error extracting cluster metrics: %v", err)
}

m := clusterEp.GetMetrics()
if m.WaitingQueueSize != 15 {
t.Errorf("cluster WaitingQueueSize: expected 15, got %d", m.WaitingQueueSize)
}
if m.RunningRequestsSize != 8 {
t.Errorf("cluster RunningRequestsSize: expected 8, got %d", m.RunningRequestsSize)
}
if m.KVCacheUsagePercent != 0.72 {
t.Errorf("cluster KVCacheUsagePercent: expected 0.72, got %f", m.KVCacheUsagePercent)
}
}

func TestCacheInfoLabelAliasing(t *testing.T) {
ctx := context.Background()

Expand Down Expand Up @@ -720,24 +796,28 @@ func TestCoreMetricsExtractorFactoryDefaultEngine(t *testing.T) {
func TestGetEngineTypeFromEndpoint(t *testing.T) {
tests := []struct {
name string
podName string
labels map[string]string
labelKey string
want string
}{
{
name: "new label key",
podName: "test-pod",
labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
labelKey: DefaultEngineTypeLabelKey,
want: "vllm",
},
{
name: "legacy GAIE label key fallback",
podName: "test-pod",
labels: map[string]string{legacyGAIEEngineTypeLabelKey: "sglang"},
labelKey: DefaultEngineTypeLabelKey,
want: "sglang",
},
{
name: "new label key takes precedence over legacy GAIE key",
name: "new label key takes precedence over legacy GAIE key",
podName: "test-pod",
labels: map[string]string{
DefaultEngineTypeLabelKey: "vllm",
legacyGAIEEngineTypeLabelKey: "sglang",
Expand All @@ -746,22 +826,38 @@ func TestGetEngineTypeFromEndpoint(t *testing.T) {
want: "vllm",
},
{
name: "no labels returns default",
name: "no labels returns default for pod",
podName: "test-pod",
labels: map[string]string{},
labelKey: DefaultEngineTypeLabelKey,
want: DefaultEngineType,
},
{
name: "nil labels returns default",
name: "nil labels returns default for pod",
podName: "test-pod",
labels: nil,
labelKey: DefaultEngineTypeLabelKey,
want: DefaultEngineType,
},
{
name: "empty PodName returns spoke-epp (cluster endpoint)",
podName: "",
labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
labelKey: DefaultEngineTypeLabelKey,
want: SpokeEPPEngineType,
},
{
name: "empty PodName with nil labels returns spoke-epp",
podName: "",
labels: nil,
labelKey: DefaultEngineTypeLabelKey,
want: SpokeEPPEngineType,
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ep := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{Labels: tt.labels}, nil)
ep := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{PodName: tt.podName, Labels: tt.labels}, nil)
got := getEngineTypeFromEndpoint(ep, tt.labelKey)
if got != tt.want {
t.Errorf("getEngineTypeFromEndpoint() = %q, want %q", got, tt.want)
Expand Down
11 changes: 11 additions & 0 deletions pkg/epp/framework/plugins/datalayer/extractor/metrics/factories.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,17 @@ var defaultEngineConfigs = []engineConfigParams{
CacheBlockSizeSpec: "",
CacheNumBlocksSpec: "",
},
// "spoke-epp" defines metrics for multi-cluster Hub-and-Spoke deployments where the Hub EPP

@elevran elevran Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please reconsider the overloading of the current metrics struct and mechanisms for carrying EPP information. While doable, feels hackish and there should be cleaner alterntiaves using core datalayer concepts for new plugins.

// scrapes pool-average metrics from Spoke EPPs. These are aggregated metrics exposed by each
// Spoke EPP representing the average state of its local pod pool.
{
Name: SpokeEPPEngineType,
QueuedRequestsSpec: "llm_d_epp_average_queue_size",
RunningRequestsSpec: "llm_d_epp_average_running_requests",
KVUsageSpec: "llm_d_epp_average_kv_cache_utilization",
LoRASpec: "",
CacheInfoSpec: "",
},
}

// defaultEngineName is the default engine used when defaultEngine is not specified.
Expand Down
4 changes: 4 additions & 0 deletions pkg/epp/framework/plugins/datalayer/extractor/metrics/mapping_registry.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,10 @@ const (
// DefaultEngineType is the key for the fallback mapping.
DefaultEngineType = "default"

// SpokeEPPEngineType is the engine type for multi-cluster Hub-and-Spoke deployments.
// Cluster endpoints (identified by empty PodName) automatically use this engine type.
SpokeEPPEngineType = "spoke-epp"

// DefaultEngineTypeLabelKey is the default label on Pods that indicates the inference engine type.
DefaultEngineTypeLabelKey = "llm-d.ai/engine-type"

Expand Down
4 changes: 3 additions & 1 deletion pkg/epp/framework/plugins/datalayer/extractor/metrics/metrics_extraction_from_config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ func buildPipeline(t *testing.T, serverURL string, params *modelServerExtractorP
// newEndpointAt creates a fwkdl.Endpoint with the given host (host:port) and optional labels.
func newEndpointAt(host string, labels map[string]string) fwkdl.Endpoint {
return fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
PodName: "test-pod",
MetricsHost: host,
Labels: labels,
}, fwkdl.NewMetrics())
Expand Down Expand Up @@ -404,7 +405,8 @@ func TestMetricsExtractionCustomCounterFromConfig(t *testing.T) {
})

ep := fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
Labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
PodName: "test-pod",
Labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"},
}, nil)
err := ext.Extract(context.Background(), fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{
Endpoint: ep,
Expand Down
Loading