feat(prometheus-write): add stream-aware sampling before conversion

[pood1e]

What changed

• add opt-in message-sampling.by-subscription rules to the Prometheus write
  output
• configure each rule with an interval and optional minimum-bytes threshold
• sample managed gNMI STREAM Update responses by source and subscription before
  protobuf-to-event conversion
• preserve every Update response before the first successful SyncResponse
• treat each reconnect as a new stream instance with a new initial snapshot
• let narrow messages, control responses, ONCE/POLL subscriptions, and unmanaged
  messages bypass sampling without consuming the interval
• align later matching messages to stable per-stream phases when spread is
  enabled
• bound sampler state with an LRU cache and roll back accepted state when
  enqueueing is canceled
• retain sampler state across unrelated output configuration updates
• expose gnmic_prometheus_write_output_messages_skipped_total

Why

Event processors run after protobuf conversion. High-frequency wide snapshots
therefore consume CPU and allocations even when a later processor discards them.
Some telemetry streams mix a multi-response initial snapshot with narrow
incremental notifications, so sampling must preserve the complete initial sync
and must not consume cadence for narrow deltas.

After initial sync, every matching response at or above minimum-bytes must be
independently discardable. Sampling is not safe for a later logical snapshot
that is split across multiple responses because the output cannot infer that
grouping.

Configuration

message-sampling:
  by-subscription:
    port-counters:
      interval: 60s
      minimum-bytes: 262144
  cache-size: 100000
  spread: true

Validation

• go test -race ./...
• cd pkg/api && go test -race ./target
• git diff --check

Fixes #953.

[pood1e]

Follow-up hardening is in 90775bb133cc954c8b9d58507ab02bdae2aed463:

• use structured, trusted source/subscription identity instead of mutable event metadata;
• isolate sampling state by source, subscription, and stream attempt so delayed responses from reconnects cannot reset the active window;
• reserve sampling admission and commit it only after the output queue accepts the message, so rollback cannot evict unrelated LRU state;
• reuse one subscription context across outputs and remove the hot-path concatenated-key allocation;
• add regressions for interleaved reconnects, transactional rollback at LRU capacity, output-manager propagation, queue timeout rollback, and metadata overrides.

Validation completed on this head:

• CGO_ENABLED=0 ./tests/run_tests.sh
• go test -race ./...
• (cd pkg/api && go test -race ./...)
• (cd pkg/cache && go test -race ./...)
• go vet ./... in the root, pkg/api, and pkg/cache modules
• go test -race -count=100 -run 'TestSubscriptionSampler|TestPromWriteOutputSampling' ./pkg/outputs/prometheus_output/prometheus_write_output
• synthetic merge with current origin/main (fa4eb451b28ed15f6d6e69a281f8cbbe5a781073), followed by full race tests in all three modules

The rejected-message benchmark reports 0 B/op and 0 allocs/op on the sampling hot path.

[karimra]

While I get the problem you are trying to solve, I have to say I'm not a fan of the solution.
The sampler filters by source and subscription and doesn't consider wildcard keys that might be present in the subscription. If responses get sampled, it can result in entire keys being sampled out.

I think the solution you are looking for is running 2 gNMIc instances where the first one subscribes to the target at a fast sample interval (say 10s) while the second one subscribes the first gNMIc instance at the desired sampling rate (say 60s).
The second instance writes to prometheus and the first instance write to whichever other output is fine with frequent writes.
If you don't need a frequent subscription, you might as well just subscribe with a 60s sample interval from the target.