feat: make model metrics endpoints configurable

Author: nayihz

cmd/epp/runner/runner.go

@@ -110,6 +110,9 @@ var (
 		"vllm:lora_requests_info",
 		"Prometheus metric for the LoRA info metrics (must be in vLLM label format).")
 
+	modelServerMetricsPort = flag.Int("modelServerMetricsPort", 0, "Port to scrape metrics from pods")
+	modelServerMetricsPath = flag.String("modelServerMetricsPath", "/metrics", "Path to scrape metrics from pods")
+
 	setupLog = ctrl.Log.WithName("setup")
 
 	// Environment variables
@@ -183,7 +186,11 @@ func (r *Runner) Run(ctx context.Context) error {
 		return err
 	}
 	verifyMetricMapping(*mapping, setupLog)
-	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.PodMetricsClientImpl{MetricMapping: mapping}, *refreshMetricsInterval)
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.PodMetricsClientImpl{
+		MetricMapping:          mapping,
+		ModelServerMetricsPort: int32(*modelServerMetricsPort),
+		ModelServerMetricsPath: *modelServerMetricsPath,
+	}, *refreshMetricsInterval)
 
 	datastore := datastore.NewDatastore(ctx, pmf)
 

pkg/epp/backend/metrics/fake.go

@@ -56,7 +56,7 @@ type FakePodMetricsClient struct {
 	Res   map[types.NamespacedName]*MetricsState
 }
 
-func (f *FakePodMetricsClient) FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, port int32) (*MetricsState, error) {
+func (f *FakePodMetricsClient) FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, url string) (*MetricsState, error) {
 	f.errMu.RLock()
 	err, ok := f.Err[pod.NamespacedName]
 	f.errMu.RUnlock()
@@ -73,6 +73,10 @@ func (f *FakePodMetricsClient) FetchMetrics(ctx context.Context, pod *backend.Po
 	return res.Clone(), nil
 }
 
+func (f *FakePodMetricsClient) GetMetricEndpoint(_ *backend.Pod, _ int32) string {
+	return "fake-metric-endpoint"
+}
+
 func (f *FakePodMetricsClient) SetRes(new map[types.NamespacedName]*MetricsState) {
 	f.resMu.Lock()
 	defer f.resMu.Unlock()

pkg/epp/backend/metrics/metrics.go

@@ -37,15 +37,13 @@ const (
 )
 
 type PodMetricsClientImpl struct {
-	MetricMapping *MetricMapping
+	MetricMapping          *MetricMapping
+	ModelServerMetricsPort int32
+	ModelServerMetricsPath string
 }
 
 // FetchMetrics fetches metrics from a given pod, clones the existing metrics object and returns an updated one.
-func (p *PodMetricsClientImpl) FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, port int32) (*MetricsState, error) {
-	// Currently the metrics endpoint is hard-coded, which works with vLLM.
-	// TODO(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/16): Consume this from InferencePool config.
-	url := "http://" + pod.Address + ":" + strconv.Itoa(int(port)) + "/metrics"
-
+func (p *PodMetricsClientImpl) FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, url string) (*MetricsState, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create request: %v", err)
@@ -70,6 +68,13 @@ func (p *PodMetricsClientImpl) FetchMetrics(ctx context.Context, pod *backend.Po
 	return p.promToPodMetrics(metricFamilies, existing)
 }
 
+func (p *PodMetricsClientImpl) GetMetricEndpoint(pod *backend.Pod, targetPortNumber int32) string {
+	if p.ModelServerMetricsPort == 0 {
+		p.ModelServerMetricsPort = targetPortNumber
+	}
+	return fmt.Sprintf("http://%s:%d%s", pod.Address, p.ModelServerMetricsPort, p.ModelServerMetricsPath)
+}
+
 // promToPodMetrics updates internal pod metrics with scraped Prometheus metrics.
 func (p *PodMetricsClientImpl) promToPodMetrics(
 	metricFamilies map[string]*dto.MetricFamily,

pkg/epp/backend/metrics/metrics_test.go

@@ -495,9 +495,10 @@ func TestFetchMetrics(t *testing.T) {
 		},
 	}
 	existing := &MetricsState{}
-	p := &PodMetricsClientImpl{} // No MetricMapping needed for this basic test
+	p := &PodMetricsClientImpl{ModelServerMetricsPort: 9999, ModelServerMetricsPath: "/metrics"} // No MetricMapping needed for this basic test
 
-	_, err := p.FetchMetrics(ctx, pod, existing, 9999) // Use a port that's unlikely to be in use.
+	url := p.GetMetricEndpoint(pod, 9999) // Use a port that's unlikely to be in use
+	_, err := p.FetchMetrics(ctx, pod, existing, url)
 	if err == nil {
 		t.Errorf("FetchMetrics() expected error, got nil")
 	}

pkg/epp/backend/metrics/pod_metrics.go

@@ -50,7 +50,8 @@ type podMetrics struct {
 }
 
 type PodMetricsClient interface {
-	FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, port int32) (*MetricsState, error)
+	FetchMetrics(ctx context.Context, pod *backend.Pod, existing *MetricsState, url string) (*MetricsState, error)
+	GetMetricEndpoint(pod *backend.Pod, poolTargetPortNumber int32) string
 }
 
 func (pm *podMetrics) String() string {
@@ -116,9 +117,11 @@ func (pm *podMetrics) refreshMetrics() error {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), fetchMetricsTimeout)
 	defer cancel()
-	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetPod(), pm.GetMetrics(), pool.Spec.TargetPortNumber)
+	url := pm.pmc.GetMetricEndpoint(pm.GetPod(), pool.Spec.TargetPortNumber)
+	updated, err := pm.pmc.FetchMetrics(ctx, pm.GetPod(), pm.GetMetrics(), url)
 	if err != nil {
 		pm.logger.V(logutil.TRACE).Info("Failed to refreshed metrics:", "err", err)
+		return err
 	}
 	// Optimistically update metrics even if there was an error.
 	// The FetchMetrics can return an error for the following reasons: