feat(metrics): Fix metrics naming and add log metrics backend

[fpacifici]

Adjust metrics naming and introduce a log based metric backend.

1. We have some issues with the metrics produced via datadog.
   1a. These are the metrics we produce sentry_streams..sentry_streams.streams.pipeline.input.message. WE are turning them into streams.pipeline.input.messages .
   1b. For the rust ones we adding streams.pipeline in front.

2. Add a backend that prints metric values in logs periodically. This is meant to allow troubleshooting metrics production. At present we do not see any metric in prod.

Metrics naming

• Python: use streams. prefix and suffix-only enum values (e.g. streams.input.messages) with pipeline tag.
• Arroyo (Rust): prefix keys with streams. so arroyo.consumer.run.count becomes streams.arroyo.consumer.run.count; pipeline tag unchanged.

Log metrics backend

• Add METRICS_PREFIX constant and LogMetricsBackend (YAML type: log, configurable period_sec) that buffers and periodically logs accumulated metrics.
• Introduce MetricsFlusher protocol, DatadogFlusher, LogFlusher, and BufferedMetricsBackend to remove duplication between Datadog and Log backends.
• Config schema and runner wiring for type: log; tests mock module-level logger and consolidate buffer tests.

Made with Cursor

[github-actions]

Semver Impact of This PR

🟡 Minor (new features)

📋 Changelog Preview

This is how your changes will appear in the changelog.
Entries from this PR are highlighted with a left border (blockquote style).

---

New Features ✨

• (metrics) Fix metrics naming and add log metrics backend by fpacifici in #270

• Update PyO3 0.24 -> 0.28.2 for Python 3.14 support by joshuarli in #266

---

_{🤖 This preview updates automatically when you update the PR.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: LogFlusher "no metrics" branch is unreachable dead code
  • Changed condition from 'if parts:' to 'if len(parts) > 1:' to properly detect when only the prefix is present without any metrics.

Or push these changes by commenting:

@cursor push e035a18b47

Preview (e035a18b47)

diff --git a/sentry_streams/sentry_streams/metrics/metrics.py b/sentry_streams/sentry_streams/metrics/metrics.py
--- a/sentry_streams/sentry_streams/metrics/metrics.py
+++ b/sentry_streams/sentry_streams/metrics/metrics.py
@@ -189,7 +189,7 @@
         for name, value, tags in gauges:
             tags_str = " ".join(tags) if tags else ""
             parts.append(f"gauge {name.value}={value} {tags_str}".strip())
-        if parts:
+        if len(parts) > 1:
             logger.info(" | ".join(parts))
         else:
             logger.info("No metrics to flush")

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}