[Python][RunInference] Add a capacity-based autoscaling signal for model-server handlers (#40907)

* [RunInference] Add a capacity-based autoscaling signal for model-server handlers

Dataflow's GPU autoscaling reads num_loaded_models_<tag> in the
BeamML_ModelManager namespace as the number of messages each worker can hold.
A model handler that serves inference through an out-of-process server (vLLM,
Triton, ...) always has exactly one loaded copy per worker, so it needs to
publish a capacity estimate under that metric instead.

Add a generic library for that in apache_beam.ml.inference.autoscaling_metrics:
- NumLoadedModelsPublisher publishes a capacity signal on a background thread,
  resetting the process-wide distribution so the latest value (not a lifetime
  mean) is reported, with optional aggregation policies on top.
- PrometheusCounterGaugeSampler reads a server's completed-requests counter and
  in-flight gauge from its Prometheus /metrics endpoint; the metric names are
  the only server-specific input.
- CompletionRateCapacityEstimator turns those samples into a per-worker
  capacity estimate: per-slot throughput times p90 concurrency over a trailing
  window. It does no I/O.
- AutoscalingMetricConfig holds the user-facing knobs, resolves the metric tag
  (explicit tag, then the RunInference model identifier verbatim, then a
  sanitized model name), and builds the estimator and publisher.

* Clean up docstrings and readability in autoscaling_metrics

- Docstrings: Google-style Args/Attributes/Raises throughout; module
  docstring shows how a handler wires the pieces together; the publisher
  docstring now reflects that 'last' is the default policy; the estimator
  docstring states which servers it suits (latencies long enough to be seen
  in flight) and shows the formula in one block.
- Parse Prometheus sample and label lines with regexes instead of a
  hand-rolled index parser. Quoted label values may now contain commas,
  braces and escapes.
- Validate the publisher's aggregation policy up front (unknown policies
  raise ValueError instead of silently behaving like 'max') and split
  ingestion into _record(), which also stops 'last' from retaining samples.
- Store estimator samples as a NamedTuple instead of positional tuples.

Outputs are unchanged for the estimator, every valid aggregation policy, and
valid Prometheus exposition (checked against the previous commit on
randomized inputs).

* Make NumLoadedModelsPublisher a plain sink

Each component now has one job: the sampler reads, the estimator computes
(the only window and the only multiplier), the publisher publishes, and the
config wires them together.

Remove the publisher's aggregation policies, its aggregation window and its
multiplier, and the matching AutoscalingMetricConfig fields. The capacity
estimate is already smoothed and its multiplier already applied, so for the
estimator these were a no-op ('last', 1.0) that read like a second smoothing
layer. A handler whose signal needs extra smoothing can apply it in its
signal_fn.

The estimator, sampler and Prometheus parsing are unchanged.

* Package the model-server capacity signal behind one call

Add AutoscalingMetricConfig.make_model_server_publisher(metrics_port_fn,
completed_metric, running_metric). A handler for a model server that exposes
a completed-requests counter and an in-flight gauge now only supplies those
two metric names and how to find the server's port; the library builds the
sampler and estimator, wires them together, honors load_signal_fn, skips
ticks while the server is down, and returns a ready publisher.

The sampler, estimator and make_publisher stay public for handlers that need
a different signal.

* Fold the sampler and signal glue into make_model_server_publisher

Producing one capacity value crossed five classes (publisher -> signal ->
sampler -> estimator -> publisher), two of which were only glue. Write the
poll as a single function inside make_model_server_publisher (port, scrape,
parse the two metrics, estimate) and remove PrometheusCounterGaugeSampler,
_ModelServerCapacitySignal and AutoscalingMetricConfig.make_capacity_estimator.

What remains is three classes with one job each (estimator computes,
publisher publishes, config configures) plus the scrape/parse functions.
Published values are unchanged.

* Make AutoscalingMetricConfig plain data

The config class mixed knobs with construction: resolve_tag() returned a copy
with the tag filled in, and make_model_server_publisher() held the polling
logic. Move construction to module-level functions:

- make_model_server_publisher(...): the one call a model-server handler makes
  (metric names, port, model identifier, optional config).
- make_publisher(...): for a handler's own capacity signal.

Both resolve the metric tag internally, so handlers no longer call a separate
resolve_tag step, and both take keyword-only arguments so the identifier,
model name and metric names can't be swapped. AutoscalingMetricConfig is now
just the knobs. Published values are unchanged.

* Let NumLoadedModelsPublisher read the server itself

Remove the make_publisher and make_model_server_publisher wrappers. A
handler now constructs NumLoadedModelsPublisher directly with its server's
metric names, a metrics_port_fn, the model identifier and an optional
AutoscalingMetricConfig; the publisher resolves the metric tag, owns its
CompletionRateCapacityEstimator, and on each poll reads the port, scrapes
/metrics (or calls config.load_signal_fn), updates the estimate and
publishes it. Behavior is unchanged.

* Describe the capacity signal in requests and state its units

* Add completion-rate aggregation modes to the capacity estimator

CompletionRateCapacityEstimator and AutoscalingMetricConfig take an
aggregation: 'concurrency' (the default and the existing estimate),
'window_mean', 'window_max' or 'lifetime_max'. The last three use the raw
completion rate and ignore concurrency, so they also work for servers whose
in-flight gauge is almost always 0. The default is unchanged.

* Tidy the estimator docstring and fill test gaps

Group the estimator docstring by mode, and make start() a no-op once a
publisher is stopped: previously a superseded publisher that was started
again would take its metric back from the newer one and then publish
nothing. Add tests for scrape failures, non-numeric samples, negative gauge
readings, running_percentile, window_secs=0, errors in the polling loop,
and repeated start() calls; the module now has full line and branch
coverage.

* Publish the autoscaling metric atomically and time samples monotonically

The SDK harness reads process-wide metrics from another thread, and
reset() followed by update() briefly left the distribution empty
(count=0), which a concurrent report could pick up. Swap the cell's data
in one step under its lock instead. Time estimator samples with
time.monotonic() so wall-clock adjustments can't distort intervals.

* Publish the autoscaling metric without touching cell internals

The previous atomic swap took DistributionCell._lock and assigned
cell.data, neither of which is reachable from Python when
apache_beam.metrics is compiled with Cython (as in CI and on workers), so
_publish raised AttributeError. Install a fresh DistributionCell holding
(sum=value, count=1) in the process-wide container with a single dict
store instead; that is still atomic under the GIL, so a concurrent reader
never sees an empty cell. Tests re-fetch the cell for every read since it
is now replaced rather than mutated.

* Add DistributionCell.reset_to and skip estimates with no completions

- Add DistributionCell.reset_to(value), which replaces the distribution
  with a single sample under the cell's lock. It works when
  apache_beam.metrics is compiled with Cython, where _lock and data are
  not reachable from outside the class. NumLoadedModelsPublisher uses it
  instead of swapping the cell in the container.
- CompletionRateCapacityEstimator returns None when requests are in flight
  but none completed within the window (slow first requests, first poll
  after an idle period), keeping the previously published capacity
  instead of reporting the floor. lifetime_max keeps reporting its peak.

* Narrow the autoscaling_metrics public API and state its requirements

Address review feedback:

- Rename NumLoadedModelsPublisher to PrometheusNumLoadedModelsPublisher and
  document what the model server must provide: a local Prometheus
  text-format /metrics endpoint with a completed-requests counter and an
  in-flight gauge (or AutoscalingMetricConfig.load_signal_fn otherwise).
- Export only AutoscalingMetricConfig and PrometheusNumLoadedModelsPublisher.
  The capacity estimator and the Prometheus scrape/parse helpers are now
  private; the aggregation modes are documented on AutoscalingMetricConfig.