drpc: add per-connection stream multiplexing metrics

[shubhamdhama]

Add client-side metrics for DRPC stream multiplexing so on-call can
diagnose stream-lifecycle issues (e.g. stream leaks). This wires three
per-connection counters through the dial path:

• streams opened (every stream the manager creates)
• streams closed (terminated gracefully)
• streams failed (error, cancellation, or connection drop)

drpcmetrics gains a label-free MuxMetrics bundle whose handles are bound
to their peer labels by the caller, so the hot path passes no labels and
allocates nothing. WithDefaults makes a nil bundle or any nil field a
no-op and gates all collection behind ShouldRecord.

drpcstream classifies termination via Stream.Succeeded: a distinct
termRemoteClosed sentinel lets a graceful remote close be told apart
from a connection drop (both otherwise surface as ClosedError). This is
a v1 heuristic; any cause that is not a recognized graceful sentinel
counts as a failure.

drpcmanager normalizes the bundle once per connection, counts opened
streams in newStream, and classifies the outcome in manageStream's
teardown. drpcclient exposes WithMuxMetrics to supply the bundle.

[suj-krishnan]

nit: The code reads better if this is renamed to muxMetrics

[suj-krishnan]

Any reason to set this to true by default?

[suj-krishnan]

Also, I think it would be better to use a single ShouldRecord knob for ClientMetrics and MuxMetrics since both are at a connection-level. Do you see a good reason to have separate knobs?

[suj-krishnan]

We are publishing these metrics for client streams. I was thinking if it would be useful to have these for server-side streams as well - but those would be aggregated across all streams and as a result, not too useful. Your thoughts?

[suj-krishnan]

This is heap-allocated and will always be allocated even on the server path (where we don't record metrics) - the allocation itself is tiny (~56 bytes) but is incurred on every connection on the server.
We can do this instead:

  type Manager struct {
        ...
        mux drpcmetrics.MuxMetrics   // was *drpcmetrics.MuxMetrics
  }

  // drpcmetrics/metrics.go
  func (m *MuxMetrics) WithDefaults() MuxMetrics {   // was *MuxMetrics
        out := MuxMetrics{}
        if m != nil {
                out = *m
        }
        if out.StreamsOpened == nil {
                out.StreamsOpened = NoOpCounter{}
        }
        if out.StreamsClosed == nil {
                out.StreamsClosed = NoOpCounter{}
        }
        if out.StreamsFailed == nil {
                out.StreamsFailed = NoOpCounter{}
        }
        if out.ShouldRecord == nil {
                out.ShouldRecord = func() bool { return true }
        }
        return out   // was &out — now copied into the caller, out stays on stack
  }

[suj-krishnan]

For a given stream, I think ShouldRecord should be evaluated only once (if stream creation is recorded but stream end is not, the counts can be misleading). We can save the value of ShouldRecord here and use the same in manageStream to record stream end.