address comments from @shmuelk

Signed-off-by: Guangya Liu <gyliu513@gmail.com>

Author: gyliu513

cmd/pd-sidecar/main.go

@@ -17,45 +17,16 @@ package main
 
 import (
 	"context"
-	"errors"
-	"fmt"
-	"net/http"
-	"time"
 
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/pflag"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log"
-	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	"github.com/llm-d/llm-d-router/pkg/common/observability/tracing"
-	sidecarmetrics "github.com/llm-d/llm-d-router/pkg/sidecar/metrics"
 	"github.com/llm-d/llm-d-router/pkg/sidecar/proxy"
 	"github.com/llm-d/llm-d-router/pkg/sidecar/version"
 )
 
-// metricsShutdownTimeout bounds graceful shutdown of the metrics server so a
-// scraper holding a connection at process exit cannot block termination.
-const metricsShutdownTimeout = 5 * time.Second
-
-// serveMetrics starts a standalone Prometheus metrics HTTP server serving
-// controller-runtime's registry, where sidecar metrics are registered.
-func serveMetrics(ctx context.Context, port int) error {
-	mux := http.NewServeMux()
-	mux.Handle("/metrics", promhttp.HandlerFor(ctrlmetrics.Registry, promhttp.HandlerOpts{EnableOpenMetrics: true}))
-	srv := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: mux}
-	go func() {
-		<-ctx.Done()
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), metricsShutdownTimeout)
-		defer cancel()
-		_ = srv.Shutdown(shutdownCtx)
-	}()
-	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-		return fmt.Errorf("metrics server: %w", err)
-	}
-	return nil
-}
-
 func main() {
 	// Initialize options with defaults
 	opts := proxy.NewOptions()
@@ -97,16 +68,6 @@ func main() {
 		}
 	}
 
-	if opts.MetricsPort > 0 {
-		sidecarmetrics.Register()
-		go func() {
-			logger.Info("starting metrics server", "port", opts.MetricsPort)
-			if err := serveMetrics(ctx, opts.MetricsPort); err != nil {
-				logger.Error(err, "metrics server failed")
-			}
-		}()
-	}
-
 	logger.Info("Proxy starting", "Built on", version.BuildRef, "From Git SHA", version.CommitSHA)
 	logger.Info("Proxy configuration", "config", opts.Config)
 

pkg/sidecar/metrics/metrics.go

@@ -14,10 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package metrics defines the Prometheus metrics exposed by the P/D sidecar
-// proxy. Metrics are registered with controller-runtime's registry so the
-// sidecar's /metrics endpoint serves them alongside Go runtime metrics, matching
-// EPP. The subsystem mirrors the sidecar span namespace (llm_d.pd_proxy.*).
+// Package metrics defines the Prometheus metrics exposed by the disaggregation
+// sidecar proxy. Metrics are registered with controller-runtime's registry and
+// served on the sidecar's /metrics route alongside Go runtime metrics, matching
+// EPP. The sidecar handles encode, prefill, and decode disaggregation.
 package metrics
 
 import (
@@ -31,18 +31,19 @@ import (
 	metricsutil "github.com/llm-d/llm-d-router/pkg/common/observability/metrics"
 )
 
-// subsystem is the Prometheus subsystem for sidecar metrics, aligned with the
-// llm_d.pd_proxy.* span namespace.
-const subsystem = "llm_d_pd_proxy"
+// subsystem is the Prometheus subsystem for the disaggregation sidecar metrics.
+const subsystem = "llm_d_disagg_sidecar"
 
-// Stage labels for prefill/decode error attribution.
+// Stage labels for encode/prefill/decode error attribution.
 const (
+	StageEncode  = "encode"
 	StagePrefill = "prefill"
 	StageDecode  = "decode"
 )
 
-// latencyBuckets covers prefill and decode wall-clock latency in seconds, from a
-// few milliseconds to several minutes (decode can stream for a long time).
+// latencyBuckets covers encode, prefill, and decode wall-clock latency in
+// seconds, from a few milliseconds to several minutes (decode can stream for a
+// long time).
 var latencyBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300}
 
 var (
@@ -64,6 +65,16 @@ var (
 		[]string{"connector"},
 	)
 
+	encodeDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Subsystem: subsystem,
+			Name:      "encode_duration_seconds",
+			Help:      metricsutil.HelpMsgWithStability("Encode stage latency in seconds, by connector.", compbasemetrics.ALPHA),
+			Buckets:   latencyBuckets,
+		},
+		[]string{"connector"},
+	)
+
 	prefillDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Subsystem: subsystem,
@@ -103,6 +114,7 @@ func Register() {
 		ctrlmetrics.Registry.MustRegister(
 			requestsTotal,
 			disaggRequestsTotal,
+			encodeDuration,
 			prefillDuration,
 			decodeDuration,
 			errorsTotal,
@@ -120,6 +132,11 @@ func RecordDisagg(connector string) {
 	disaggRequestsTotal.WithLabelValues(connector).Inc()
 }
 
+// RecordEncodeDuration records encode stage latency for the given connector.
+func RecordEncodeDuration(connector string, d time.Duration) {
+	encodeDuration.WithLabelValues(connector).Observe(d.Seconds())
+}
+
 // RecordPrefillDuration records prefill stage latency for the given connector.
 func RecordPrefillDuration(connector string, d time.Duration) {
 	prefillDuration.WithLabelValues(connector).Observe(d.Seconds())

pkg/sidecar/metrics/metrics_test.go

@@ -45,12 +45,15 @@ func TestRecordDisagg(t *testing.T) {
 }
 
 func TestRecordDurations(t *testing.T) {
+	encodeDuration.Reset()
 	prefillDuration.Reset()
 	decodeDuration.Reset()
 
+	RecordEncodeDuration("nixl", 50*time.Millisecond)
 	RecordPrefillDuration("nixlv2", 100*time.Millisecond)
 	RecordDecodeDuration("nixlv2", 250*time.Millisecond)
 
+	require.Equal(t, 1, promtestutil.CollectAndCount(encodeDuration))
 	require.Equal(t, 1, promtestutil.CollectAndCount(prefillDuration))
 	require.Equal(t, 1, promtestutil.CollectAndCount(decodeDuration))
 }
@@ -61,9 +64,11 @@ func TestRecordError(t *testing.T) {
 	RecordError("sglang", StagePrefill)
 	RecordError("sglang", StagePrefill)
 	RecordError("sglang", StageDecode)
+	RecordError("nixl", StageEncode)
 
 	require.Equal(t, 2.0, promtestutil.ToFloat64(errorsTotal.WithLabelValues("sglang", StagePrefill)))
 	require.Equal(t, 1.0, promtestutil.ToFloat64(errorsTotal.WithLabelValues("sglang", StageDecode)))
+	require.Equal(t, 1.0, promtestutil.ToFloat64(errorsTotal.WithLabelValues("nixl", StageEncode)))
 }
 
 // Register must be idempotent so repeated calls do not panic on duplicate

pkg/sidecar/proxy/connector_ec_nixl.go

@@ -6,9 +6,11 @@ import (
 	"fmt"
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
 	logging "github.com/llm-d/llm-d-router/pkg/common/observability/logging"
+	"github.com/llm-d/llm-d-router/pkg/sidecar/metrics"
 )
 
 // fanoutEncoderCollect fans out per-image encoder requests and merges
@@ -102,14 +104,17 @@ func (s *Server) handleECNIXL(w http.ResponseWriter, r *http.Request, prefillEnd
 
 	// Step 1: fan out to encoders, collect per-image ec_transfer_params.
 	if len(encodeEndPoints) > 0 {
+		encodeStart := time.Now()
 		params, contributed, total, err := s.fanoutEncoderCollect(r.Context(), completionRequest, encodeEndPoints, requestID)
 		if err != nil {
+			metrics.RecordError(s.config.ECConnector, metrics.StageEncode)
 			s.logger.Error(err, "encoder processing failed", "requestID", requestID)
 			if err := errorBadGateway(err, w); err != nil {
 				s.logger.Error(err, "failed to send error response to client")
 			}
 			return
 		}
+		metrics.RecordEncodeDuration(s.config.ECConnector, time.Since(encodeStart))
 		if total > 0 {
 			// All-missing degrades silently to primer-mode; warn so the
 			// operator sees the regression.

pkg/sidecar/proxy/options.go

@@ -73,7 +73,6 @@ const (
 	inlineConfiguration       = "configuration"
 	configurationFile         = "configuration-file"
 	tracingFlag               = "tracing"
-	metricsPortFlag           = "metrics-port"
 
 	// Deprecated flags
 	connector                      = "connector"
@@ -93,7 +92,6 @@ const (
 	defaultVLLMPort              = "8001"
 	defaultDataParallelSize      = 1
 	defaultMooncakeBootstrapPort = 8998
-	defaultMetricsPort           = 9090
 
 	// TLS stages
 	prefillStage = "prefiller"
@@ -127,7 +125,6 @@ type yamlConfiguration struct {
 	PrefillRetryBackoff            string   `json:"prefill-retry-backoff,omitempty"`
 	DecodeChunkSize                int      `json:"decode-chunk-size,omitempty"`
 	Tracing                        *bool    `json:"tracing,omitempty"`
-	MetricsPort                    *int     `json:"metrics-port,omitempty"`
 }
 
 // Options holds the CLI-facing configuration for the pd-sidecar proxy.
@@ -216,7 +213,6 @@ func NewOptions() *Options {
 			PoolGroup:               routing.InferencePoolAPIGroup,
 			DecodeChunkSize:         0,
 			Tracing:                 false,
-			MetricsPort:             defaultMetricsPort,
 			// MoRI-IO defaults: off, preserving existing NIXLv2 behaviour.
 			// Port defaults match vLLM's MoRI-IO connector defaults.
 			MoRIIOWriteMode:            false,
@@ -266,7 +262,6 @@ func (opts *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&opts.PoolGroup, poolGroup, opts.PoolGroup, "group of the InferencePool this Endpoint Picker is associated with.")
 	fs.IntVar(&opts.DecodeChunkSize, decodeChunkSize, opts.DecodeChunkSize, "enables chunked decode mode when > 0; value is the token budget per chunk. For best performance should be a multiple of the block size.")
 	fs.BoolVar(&opts.Tracing, tracingFlag, opts.Tracing, "Enable OpenTelemetry tracing")
-	fs.IntVar(&opts.MetricsPort, metricsPortFlag, opts.MetricsPort, "the port the Prometheus /metrics endpoint listens on (0 disables it)")
 
 	// MoRI-IO WRITE-mode flags. Only meaningful with --kv-connector=nixlv2
 	// against vLLM engines running MoRI-IO in WRITE mode.
@@ -551,11 +546,6 @@ func (opts *Options) Validate() error {
 		return fmt.Errorf("--mooncake-bootstrap-port must be between 1 and 65535, got %d", opts.MooncakeBootstrapPort)
 	}
 
-	// Validate metrics port (0 disables the metrics server)
-	if opts.MetricsPort < 0 || opts.MetricsPort > 65535 {
-		return fmt.Errorf("--metrics-port must be between 0 and 65535 (0 disables it), got %d", opts.MetricsPort)
-	}
-
 	// Validate SSRF protection requirements
 	if opts.EnableSSRFProtection {
 		if opts.InferencePoolNamespace == "" || opts.InferencePoolName == "" {
@@ -647,9 +637,6 @@ func (opts *Options) mergeYAMLConfiguration(cfg yamlConfiguration) {
 	if cfg.MooncakeBootstrapPort != 0 && !opts.isFlagSet(mooncakeBootstrapPortFlag) {
 		opts.MooncakeBootstrapPort = cfg.MooncakeBootstrapPort
 	}
-	if cfg.MetricsPort != nil && !opts.isFlagSet(metricsPortFlag) {
-		opts.MetricsPort = *cfg.MetricsPort
-	}
 	if cfg.DataParallelSize != 0 && !opts.isFlagSet(dataParallelSize) {
 		opts.DataParallelSize = cfg.DataParallelSize
 	}

pkg/sidecar/proxy/options_test.go

@@ -621,32 +621,6 @@ func TestValidateConnector(t *testing.T) {
 	}
 }
 
-func TestValidateMetricsPort(t *testing.T) {
-	tests := []struct {
-		name        string
-		metricsPort int
-		wantErr     bool
-	}{
-		{"default", defaultMetricsPort, false},
-		{"disabled", 0, false},
-		{"max", 65535, false},
-		{"negative", -1, true},
-		{"too large", 65536, true},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			opts := NewOptions()
-			opts.MetricsPort = tt.metricsPort
-			_ = opts.Complete() // Complete must be called before Validate
-			err := opts.Validate()
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
 func TestValidateTLSStages(t *testing.T) {
 	tests := []struct {
 		name      string

pkg/sidecar/proxy/proxy.go

@@ -31,11 +31,14 @@ import (
 
 	"github.com/go-logr/logr"
 	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"golang.org/x/sync/errgroup"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	"github.com/llm-d/llm-d-router/pkg/sidecar/constants"
+	"github.com/llm-d/llm-d-router/pkg/sidecar/metrics"
 )
 
 const (
@@ -189,10 +192,6 @@ type Config struct {
 	// InsecureSkipVerifyForDecoder configures the proxy to skip TLS verification for requests to the decoder.
 	InsecureSkipVerifyForDecoder bool
 
-	// MetricsPort is the port the Prometheus /metrics endpoint listens on.
-	// 0 disables the metrics server.
-	MetricsPort int
-
 	// SecureServing enables TLS for the sidecar server itself.
 	SecureServing bool
 	// CertPath is the path to TLS certificates for the sidecar server.
@@ -490,6 +489,9 @@ func (s *Server) createRoutes() *http.ServeMux {
 	mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusOK)
 	})
+
+	metrics.Register()
+	mux.Handle("GET /metrics", promhttp.HandlerFor(ctrlmetrics.Registry, promhttp.HandlerOpts{EnableOpenMetrics: true}))
 	mux.HandleFunc("POST "+ChatCompletionsPath, s.disaggregatedPrefillHandler(APITypeChatCompletions))
 	mux.HandleFunc("POST "+CompletionsPath, s.disaggregatedPrefillHandler(APITypeChatCompletions))
 	mux.HandleFunc("POST "+MessagesPath, s.disaggregatedPrefillHandler(APITypeChatCompletions))