diff --git a/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor.go b/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor.go index c9fbcc1f58..9ab476875a 100644 --- a/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor.go +++ b/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor.go @@ -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. +// Cluster endpoints (identified by empty PodName) automatically use SpokeEPPEngineType. +// 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 } diff --git a/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor_test.go b/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor_test.go index 6305a3572e..f5bf2a3a0f 100644 --- a/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor_test.go +++ b/pkg/epp/framework/plugins/datalayer/extractor/metrics/extractor_test.go @@ -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", + Labels: map[string]string{DefaultEngineTypeLabelKey: "vllm"}, }, nil) _ = extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{Payload: data, Endpoint: epVllm}) if epVllm.GetMetrics().WaitingQueueSize != 10 { @@ -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 { @@ -287,8 +289,8 @@ 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) @@ -296,7 +298,8 @@ func TestBackwardCompatibility(t *testing.T) { // 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 { @@ -304,6 +307,79 @@ func TestBackwardCompatibility(t *testing.T) { } } +func TestSpokeEPPClusterEndpointExtraction(t *testing.T) { + ctx := context.Background() + + registry := NewMappingRegistry() + + // Register spoke-epp mapping for cluster endpoints + spokeMapping, err := NewMapping( + "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() @@ -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", @@ -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) diff --git a/pkg/epp/framework/plugins/datalayer/extractor/metrics/factories.go b/pkg/epp/framework/plugins/datalayer/extractor/metrics/factories.go index 1f783b7cbf..a15c28fa27 100644 --- a/pkg/epp/framework/plugins/datalayer/extractor/metrics/factories.go +++ b/pkg/epp/framework/plugins/datalayer/extractor/metrics/factories.go @@ -143,6 +143,17 @@ var defaultEngineConfigs = []engineConfigParams{ CacheBlockSizeSpec: "", CacheNumBlocksSpec: "", }, + // "spoke-epp" defines metrics for multi-cluster Hub-and-Spoke deployments where the Hub EPP + // 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. diff --git a/pkg/epp/framework/plugins/datalayer/extractor/metrics/mapping_registry.go b/pkg/epp/framework/plugins/datalayer/extractor/metrics/mapping_registry.go index fbe2a52191..eac3d7ccbd 100644 --- a/pkg/epp/framework/plugins/datalayer/extractor/metrics/mapping_registry.go +++ b/pkg/epp/framework/plugins/datalayer/extractor/metrics/mapping_registry.go @@ -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" diff --git a/pkg/epp/framework/plugins/datalayer/extractor/metrics/metrics_extraction_from_config_test.go b/pkg/epp/framework/plugins/datalayer/extractor/metrics/metrics_extraction_from_config_test.go index cd4bb694f1..dde40b5131 100644 --- a/pkg/epp/framework/plugins/datalayer/extractor/metrics/metrics_extraction_from_config_test.go +++ b/pkg/epp/framework/plugins/datalayer/extractor/metrics/metrics_extraction_from_config_test.go @@ -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()) @@ -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,