compute: retry metric sink collector registration

[mtabebe]

Problem:

A curated metric sink keys its gauges on its stable definition
name, so every incarnation shares one prometheus Desc id. When a
restarted environmentd reconciles against a still-running replica, the
old incarnation and the new one live in the same process, and nothing
sequences the old dataflow's teardown (which drops its registration)
before the new one renders. Registration collided and
register_collector_with_dropper soft-panicked.

Solution:

Add MetricsRegistry::try_register_collector_with_dropper, a
fallible variant that returns the prometheus error, and re-base the
soft-panicking one on it so its callers keep the backstop. The sink
operator owns a PendingRegistration and retries once a second,
self-scheduling via its activator so a quiescent input still gets
another try. The guard lives in the operator closure, so it drops with
the dataflow, freeing the Desc id for the retrying incarnation.

A new counter, mz_compute_metric_sink_registration_retries_total,
makes retries visible.

[def-]

QA LLM Review

1. MEDIUM -- Moving the registration guard into the operator closure unregisters the collector when the operator shuts down, not when the sink is dropped

src/compute/src/sink/metric_sink.rs:215

render_sink now returns None and keeps the registration guard inside the operator's logic closure. Timely drops that closure as soon as the operator shuts down, which happens whenever both input frontiers reach the empty antichain, not only when the dataflow is torn down. A metric sink whose source collection finishes (a constant-folding source) therefore unregisters its collector moments after it is created and publishes nothing, silently and permanently; before this change the guard lived in sink_token and survived operator shutdown.

Details

The operator is built with OperatorBuilder::build, not build_reschedule, so its logic always reports "not incomplete" (timely wraps it as move |frontier| { logic(frontier); false }). Timely's Subgraph::activate_child then does: if the child reported not-incomplete and frontiers_empty && no_capabilities, call child.shut_down(), which sets operator = None and drops the logic closure. The metric sink holds no capabilities (op.build(move |_capabilities| ...) drops the minted vec; the operator has no outputs), so the only condition is empty input frontiers.

Concretely reachable: CREATE VIEW v AS SELECT 'm'::text AS metric_name, 'gauge'::text AS metric_type, '{}'::map[text=>text] AS labels, 1.0::double AS value, 'h'::text AS help (no FROM, or a FROM whose predicate folds to false), then CREATE METRIC SINK s FROM v. The optimizer folds the shaped source to a Constant, so the dataflow has no imports and src/compute/src/render.rs:1330 renders it via to_stream, whose frontier reaches the empty antichain after the initial batch. Both ok_input and err_input close, the operator shuts down, and the guard drops. expire_collection_at would pin a capability at the expiration time and mask this, but it is only applied when dataflow_expiration is set, which it is not for a dataflow with no time dependence.

The same mechanism also means a registration that is still retrying when the input closes never completes: the operator is gone before the next activate_after fires.

Suggested fix: keep the guard's lifetime tied to the collection rather than the operator. Share the PendingRegistration as an Rc<RefCell<_>> between the operator (which fills in handle on a successful retry) and the value returned from render_sink, so it still lands in collection.sink_token. The retry path is unaffected, and the drop point goes back to drop_collection, which is strictly earlier than dataflow drain and so also shortens the window a successor has to retry through.

[antiguru]

If we don't have a registration, do we not have to store the data locally so that we can publish it subsequently? Our inputs should be changes to metrics, not the metrics on each tick, so here we'd only start recording metrics once they change, right?

[mtabebe]

I'm not sure I 100% understand your concern.

I think there are a few things going on:

1. The sink reads its source with with_snapshot: true. So my understanding is that it means that it gets the entire current contents at the as_of (as diffs), and then subsequent changes. So there isn't only recording of new changes.

2. If we don't have a registration, that means we are a none active worker (only one worker receives data in the model). So a non active worker has nothing to store regardless of whether the registration is sucessful or not.

3. The active worker accumulates regardless of registration outcome. After try_register, we don't have early return. Failed attempts schedules a retry, but execution continues. So the diffs arriving while registration is pending are accumulated in theSinkState. And then a scrape after registration exposes the snapshot

So I don't quite follow where we would miss metrics.

[antiguru]

Left a comment; I think we can miss metrics with this approach. Do you mind checking this?