feat(coordinator): expose Prometheus metrics

[dmitripikus]

What type of PR is this?
/kind feature

What this PR does / why we need it:
Adds Prometheus metrics to the coordinator. With EPP metrics alone, operators see per-leg sub-requests, not client requests, so they cannot answer request rate, latency, error mix, per-step wall time, upstream fan-out, or disaggregation path from the coordinator process itself.

Three commits map to the three families in the spec (#2277):

1. Endpoint + request family — new pkg/coordinator/metrics package, /metrics on its own port (default 9090, --metrics-port), and request_total / request_error_total / request_duration_seconds / request_size_bytes / request_running.
2. Step + upstream families — the pipeline executor brackets every step with step_running / step_duration_seconds / step_errors_total and classifies failures (bad_request, upstream_4xx, upstream_5xx, internal); each outbound call site emits upstream_request_total / upstream_request_duration_seconds.
3. Per-request outcomes — execution_path_total, conditional_decode_probes_total, request_input_tokens.

Every model_name label is capped at 1000 distinct values with an other overflow, matching EPP's cardinality guard.

Deviations from the WIP doc in #2277, per review feedback that has not yet landed in the doc:

• Drops response_size_bytes
• Renames disagg_decision_total to execution_path_total (the coordinator observes rather than decides).
• Adds request_input_tokens (token IDs are available on RequestContext.TokenIDs after render).

Which issue(s) this PR fixes:
Part of #2276

Release note:

Coordinator now exposes Prometheus metrics on a dedicated /metrics endpoint (default port 9090, configurable via --metrics-port or server.metrics_port). Metrics cover request rate/latency/error mix, per-pipeline-step wall time and error classification, per-upstream call counts and latencies, execution path (decode-only / prefill-decode / encode-prefill-decode), conditional-decode probe outcomes, and render-derived input token count. Every model_name label is capped at 1000 distinct values with an "other" overflow.

[roytman]

Thank you, @dmitripikus. I'm still working on the review, but I have 2 initial comments.

[roytman]

we have some inconsistency here (can be resolved in a separate PR)

• pkg/sidecar/proxy: MetricsPort > 0 → enabled; 0 (default) or unset → disabled, falls back to MORIIO_METRICS_ADDR env var. Documented in the flag help text and README.
• pkg/coordinator (this PR): same rule, explicitly copied: cfg.MetricsPort > 0 starts the goroutine, non-positive logs "metrics endpoint disabled" and skips it. Its own comment states it's "matching pkg/sidecar/proxy semantics."
• cmd/epp/runner: No disable path at all. serveMetrics is unconditionally wired into the run group; --metrics-port defaults to 9090 and has no "0 disables" semantics in its flag help, its Options.Validate(), or the server code. Setting --metrics-port=0 binds ":0" (OS picks a random ephemeral port) rather than disabling anything; a negative value would fail ListenAndServe and — since EPP's run-group cancels all tasks when one fails — take down the whole process.

[dmitripikus]

For now I replaced wrong statement "matching pkg/sidecar/proxy semantics": coordinator matches the > 0 gate but not the MORIIO_METRICS_ADDR env-var fallback. Now the comment states the rule directly.

The broader inconsistency with cmd/epp/runner (no disable gate; --metrics-port=0 binds a random ephemeral port, negative crashes the whole errgroup) is a real bug worth its own issue + PR. I have not filed either yet — I'll open the issue if you agree with the semantics, i.e. align EPP with coordinator: <= 0 disables, log "metrics endpoint disabled", update --metrics-port help text; leave sidecar's env-var fallback alone as legacy.

[roytman]

yes, it is identical to pkg/epp/metrics/metrics.go:60-64

[roytman]

identical to pkg/epp/metrics/llm_d_router_metrics.go:73-77

[roytman]

Almost identical to EPP, fixes the typo 32778→32768

[roytman]

Register/Reset: are not a duplicate of EPP's Register/Reset - actually structurally here is better. EPP hand-repeats its ~60-metric list once in Register and again in Reset (llm_d_router_metrics.go:424-503, 506-578), and its label composition uses raw append(modelLabels, "error_code") with no aliasing guard - exactly the hazard withLabel was built to prevent.
The coordinator's single allCollectors() list feeding both functions, plus withLabel, is arguably a pattern EPP should adopt, not the reverse.

[roytman]

Should we mention here that we exceed the 412 error for conditional-decode

[roytman]

The current code is correct, but the idiomatic Go fix for "I need a defer scoped to one loop iteration" is to extract a named function rather than an inline IIFE — it's the same defer-scoping trick, just without the closure-capturing-an-outer-variable smell:

func (p *Pipeline) runStep(ctx context.Context, reqCtx *RequestContext, step Step, idx int, timings []stepTiming) error {
	name := step.Name()
	coordmetrics.IncStepRunning(name)
	start := time.Now()
	defer func() {
		d := time.Since(start)
		coordmetrics.RecordStepDuration(name, d)
		coordmetrics.DecStepRunning(name)
		timings[idx] = stepTiming{name: name, duration: d}
		if r := recover(); r != nil {
			coordmetrics.IncStepErrorTotal(name, coordmetrics.ErrorCodeInternal)
			panic(r)
		}
	}()
	return step.Execute(ctx, reqCtx)
}

and the loop becomes:

for idx, step := range p.steps {
	if err := ctx.Err(); err != nil {
		return fmt.Errorf("pipeline cancelled: %w", err)
	}
	logger.V(logutil.TRACE).Info("step starting", "step", step.Name())
	if err := p.runStep(ctx, reqCtx, step, idx, timings); err != nil {
		if errors.Is(err, ErrPipelineDone) {
			executed[step.Name()] = true
			return nil
		}
		coordmetrics.IncStepErrorTotal(step.Name(), coordmetrics.ClassifyErrorCode(err, ClassifyOpts))
		return fmt.Errorf("step %q failed: %w", step.Name(), err)
	}
	executed[step.Name()] = true
	logger.V(logutil.TRACE).Info("step complete", "step", step.Name())
}

Why this is better, not just different:

• No outer-variable mutation through a closure. The current code does var err error; func() { ...; err = step.Execute(...) }() — assigning into a variable declared outside the closure. A real function that just returns is the more direct idiom and is easier to read at a glance.
• Independently testable. runStep can be unit-tested directly (call it, panic a stub step, assert the gauge/duration/error recording) without going through the whole Execute loop.
• Same defer-scoping guarantee. A method's own body is exactly as valid a function boundary for defer as an anonymous IIFE — nothing is lost.

This is a pure readability refactor, not a correctness fix — the anonymous-function version in the PR is not buggy, this is just the more idiomatic shape for "defer must run once per loop iteration."