Fix three critical metrics backend bugs

Bug 1 (high): Fix constructor tags being dropped when per-call tags are empty dict
- In DatadogMetricsBackend, the condition 'if tags else None' was falsy for empty dict {}
- BufferedMetricsBackend always passes a dict (possibly empty), causing constructor tags to be lost
- Fixed by calling _combine_tags first, then checking if the result is truthy
- Added test to verify constructor tags are preserved with empty dict

Bug 2 (medium): Fix ignored period_sec parameter in LogMetricsBackend
- LogMetricsBackend accepted period_sec but never used it
- configure_metrics always used hardcoded METRICS_FREQUENCY_SEC (10s)
- Fixed by removing unused parameter from LogMetricsBackend constructor
- Added throttle_interval_sec parameter to configure_metrics
- Updated runner.py to pass period_sec to configure_metrics
- Updated all test calls to remove period_sec parameter

Bug 3 (high): Fix unpicklable Datadog backend in multiprocessing initializer
- get_inner_metrics() returned DatadogMetricsBackend with unpicklable thread/socket
- Multiprocessing with spawn context cannot pickle threads or sockets
- Fixed by creating serializable MetricsConfig dataclass
- Pass config to initializer instead of backend instance
- Initializer recreates backend from config in each subprocess
- Updated runner.py to create and store MetricsConfig
- Updated rust_arroyo.py to use get_metrics_config() and pass config to initializer

Author: cursoragent

sentry_streams/sentry_streams/adapters/arroyo/rust_arroyo.py

@@ -41,7 +41,11 @@
     get_metrics,
     get_size,
 )
-from sentry_streams.metrics.metrics import get_inner_metrics
+from sentry_streams.metrics.metrics import (
+    MetricsConfig,
+    build_metrics_backend_from_config,
+    get_metrics_config,
+)
 from sentry_streams.pipeline.function_template import (
     InputType,
     OutputType,
@@ -79,9 +83,10 @@
 logger = logging.getLogger(__name__)
 
 
-def initializer(metrics: MetricsBackend) -> None:
-    # Reinitialize the metrics backend for each process
-    configure_metrics(metrics)
+def initializer(config: MetricsConfig) -> None:
+    # Reinitialize the metrics backend for each process from serializable config
+    metrics = build_metrics_backend_from_config(config)
+    configure_metrics(metrics, config=config, throttle_interval_sec=config.throttle_interval_sec)
 
 
 def _metrics_wrapped_function(
@@ -194,6 +199,7 @@ def finalize_chain(chains: TransformChains, route: Route) -> RuntimeOperator:
             f"batch_time={config['batch_time']}"
         )
 
+        metrics_config = get_metrics_config()
         return RuntimeOperator.PythonAdapter(
             rust_route,
             MultiprocessDelegateFactory(
@@ -202,7 +208,7 @@ def finalize_chain(chains: TransformChains, route: Route) -> RuntimeOperator:
                 config["batch_time"],
                 MultiprocessingPool(
                     num_processes=config["processes"],
-                    initializer=functools.partial(initializer, get_inner_metrics()),
+                    initializer=functools.partial(initializer, metrics_config),
                 ),
                 input_block_size=config.get("input_block_size"),
                 output_block_size=config.get("output_block_size"),

sentry_streams/sentry_streams/metrics/__init__.py

@@ -6,6 +6,7 @@
     Metric,
     Metrics,
     MetricsBackend,
+    MetricsConfig,
     configure_metrics,
     get_metrics,
     get_size,
@@ -21,5 +22,6 @@
     "Metric",
     "Metrics",
     "MetricsBackend",
+    "MetricsConfig",
     "get_size",
 ]

sentry_streams/sentry_streams/metrics/metrics.py

@@ -3,6 +3,7 @@
 import logging
 import time
 from abc import abstractmethod
+from dataclasses import dataclass
 from enum import Enum
 from typing import (
     Any,
@@ -159,24 +160,27 @@ def __normalize_tags(self, tags: Tags) -> list[str]:
     def increment(
         self, name: str, value: Union[int, float] = 1, tags: Optional[Tags] = None
     ) -> None:
+        combined_tags = _combine_tags(self.__tags, tags)
         self.datadog_client.increment(
             name,
             value,
-            tags=self.__normalize_tags(_combine_tags(self.__tags, tags)) if tags else None,
+            tags=self.__normalize_tags(combined_tags) if combined_tags else None,
         )
 
     def gauge(self, name: str, value: Union[int, float], tags: Optional[Tags] = None) -> None:
+        combined_tags = _combine_tags(self.__tags, tags)
         self.datadog_client.gauge(
             name,
             value,
-            tags=self.__normalize_tags(_combine_tags(self.__tags, tags)) if tags else None,
+            tags=self.__normalize_tags(combined_tags) if combined_tags else None,
         )
 
     def timing(self, name: str, value: Union[int, float], tags: Optional[Tags] = None) -> None:
+        combined_tags = _combine_tags(self.__tags, tags)
         self.datadog_client.timing(
             name,
             value,
-            tags=self.__normalize_tags(_combine_tags(self.__tags, tags)) if tags else None,
+            tags=self.__normalize_tags(combined_tags) if combined_tags else None,
         )
 
 
@@ -186,10 +190,9 @@ class LogMetricsBackend(MetricsBackend):
     as LogFlusher: ``prefix | counter|gauge|timing name=value tag1:val1 ...``.
     """
 
-    def __init__(self, period_sec: float, tags: Optional[Tags] = None) -> None:
+    def __init__(self, tags: Optional[Tags] = None) -> None:
         self.__prefix = METRICS_PREFIX.strip(".")
         self.__base_tags: Tags = tags if tags is not None else {}
-        self._period_sec = period_sec
 
     @staticmethod
     def __normalize_tags(tags: Tags) -> list[str]:
@@ -384,12 +387,52 @@ def timing(
         self.__backend.timing(name, value, tags=_tags_from_mapping(tags))
 
 
+@dataclass
+class MetricsConfig:
+    """
+    Serializable configuration for metrics backends.
+    Used to pass metrics configuration to multiprocessing workers.
+    """
+
+    backend_type: str
+    host: Optional[str] = None
+    port: Optional[int] = None
+    tags: Optional[Tags] = None
+    udp_queue_size: Optional[int] = None
+    throttle_interval_sec: Optional[float] = None
+
+
+def build_metrics_backend_from_config(config: MetricsConfig) -> MetricsBackend:
+    """
+    Build a metrics backend from a serializable configuration.
+    This is used in multiprocessing workers to recreate the backend.
+    """
+    if config.backend_type == "datadog":
+        assert config.host is not None and config.port is not None
+        return DatadogMetricsBackend(
+            host=config.host,
+            port=config.port,
+            tags=config.tags,
+            udp_queue_size=config.udp_queue_size,
+        )
+    elif config.backend_type == "log":
+        return LogMetricsBackend(tags=config.tags)
+    else:
+        return DummyMetricsBackend()
+
+
 _inner_metrics_backend: Optional[MetricsBackend] = None
 _metrics_backend: Optional[MetricsBackend] = None
+_metrics_config: Optional[MetricsConfig] = None
 _dummy_metrics_backend = DummyMetricsBackend()
 
 
-def configure_metrics(metrics: MetricsBackend, force: bool = False) -> None:
+def configure_metrics(
+    metrics: MetricsBackend,
+    force: bool = False,
+    throttle_interval_sec: Optional[float] = None,
+    config: Optional[MetricsConfig] = None,
+) -> None:
     """
     Metrics can generally only be configured once, unless force is passed
     on subsequent initializations.
@@ -398,14 +441,17 @@ def configure_metrics(metrics: MetricsBackend, force: bool = False) -> None:
     """
     global _metrics_backend
     global _inner_metrics_backend
+    global _metrics_config
     if not force:
         assert _metrics_backend is None, "Metrics is already set"
 
     # Runtime-check the metrics implementation so misconfiguration fails early.
     assert isinstance(metrics, MetricsBackend)
 
     _inner_metrics_backend = metrics
-    _metrics_backend = BufferedMetricsBackend(metrics, throttle_interval_sec=METRICS_FREQUENCY_SEC)
+    interval = throttle_interval_sec if throttle_interval_sec is not None else METRICS_FREQUENCY_SEC
+    _metrics_backend = BufferedMetricsBackend(metrics, throttle_interval_sec=interval)
+    _metrics_config = config
     arroyo_configure_metrics(ArroyoMetricsBackend(_metrics_backend))
 
 
@@ -415,6 +461,14 @@ def get_inner_metrics() -> MetricsBackend:
     return _inner_metrics_backend
 
 
+def get_metrics_config() -> Optional[MetricsConfig]:
+    """
+    Get the serializable metrics configuration.
+    This is used to pass metrics configuration to multiprocessing workers.
+    """
+    return _metrics_config
+
+
 def get_metrics() -> Metrics:
     if _metrics_backend is None:
         return Metrics(_dummy_metrics_backend)

sentry_streams/sentry_streams/runner.py

@@ -17,6 +17,7 @@
     DummyMetricsBackend,
     LogMetricsBackend,
     MetricsBackend,
+    MetricsConfig,
     configure_metrics,
 )
 from sentry_streams.pipeline.config import load_config
@@ -106,6 +107,7 @@ def load_runtime(
 
     metric_config = environment_config.get("metrics", {})
     metrics_backend: MetricsBackend
+    serializable_config: MetricsConfig
     if metric_config.get("type") == "datadog":
         default_tags = metric_config.get("tags", {})
         default_tags["pipeline"] = name
@@ -116,7 +118,14 @@ def load_runtime(
             tags=default_tags,
             udp_queue_size=metric_config.get("udp_queue_size"),
         )
-        configure_metrics(metrics_backend)
+        serializable_config = MetricsConfig(
+            backend_type="datadog",
+            host=metric_config["host"],
+            port=metric_config["port"],
+            tags=default_tags,
+            udp_queue_size=metric_config.get("udp_queue_size"),
+        )
+        configure_metrics(metrics_backend, config=serializable_config)
         metric_config = {
             "host": metric_config["host"],
             "port": metric_config["port"],
@@ -129,14 +138,23 @@ def load_runtime(
         default_tags["pipeline"] = name
 
         metrics_backend = LogMetricsBackend(
-            period_sec=metric_config["period_sec"],
             tags=default_tags,
         )
-        configure_metrics(metrics_backend)
+        serializable_config = MetricsConfig(
+            backend_type="log",
+            tags=default_tags,
+            throttle_interval_sec=metric_config["period_sec"],
+        )
+        configure_metrics(
+            metrics_backend,
+            throttle_interval_sec=metric_config["period_sec"],
+            config=serializable_config,
+        )
         metric_config = {}
     else:
         metrics_backend = DummyMetricsBackend()
-        configure_metrics(metrics_backend)
+        serializable_config = MetricsConfig(backend_type="dummy")
+        configure_metrics(metrics_backend, config=serializable_config)
         metric_config = {}
 
     assigned_segment_id = int(segment_id) if segment_id else None

sentry_streams/tests/pipeline/test_metrics.py

@@ -138,6 +138,27 @@ def test_datadog_increment_merges_constructor_tags(mock_dogstatsd: Any) -> None:
     assert set(mock_client.timing.call_args[1]["tags"]) == expected
 
 
+@patch("sentry_streams.metrics.metrics.DogStatsd")
+def test_datadog_constructor_tags_preserved_with_empty_dict(mock_dogstatsd: Any) -> None:
+    backend = DatadogMetricsBackend("localhost", 8125, tags={"service": "streams"})
+    mock_client = mock_dogstatsd.return_value
+
+    backend.increment(_metric(Metric.INPUT_MESSAGES), 1, tags={})
+    backend.gauge(_metric(Metric.INPUT_BYTES), 100, tags={})
+    backend.timing(_metric(Metric.DURATION), 1500, tags={})
+
+    expected = ["service:streams"]
+    mock_client.increment.assert_called_once()
+    assert mock_client.increment.call_args[0] == ("input.messages", 1)
+    assert mock_client.increment.call_args[1]["tags"] == expected
+    mock_client.gauge.assert_called_once()
+    assert mock_client.gauge.call_args[0] == ("input.bytes", 100)
+    assert mock_client.gauge.call_args[1]["tags"] == expected
+    mock_client.timing.assert_called_once()
+    assert mock_client.timing.call_args[0] == ("duration", 1500)
+    assert mock_client.timing.call_args[1]["tags"] == expected
+
+
 @patch("sentry_streams.metrics.metrics.DogStatsd")
 @patch("time.time")
 def test_buffered_increment_with_throttling(mock_time: Any, mock_dogstatsd: Any) -> None:
@@ -248,7 +269,7 @@ def test_arroyo_methods_without_tags_pass_empty_dict() -> None:
 
 @patch("sentry_streams.metrics.metrics.logger")
 def test_log_increment_logs_immediately(mock_logger: Any) -> None:
-    backend = LogMetricsBackend(period_sec=15.0, tags={"env": "test"})
+    backend = LogMetricsBackend(tags={"env": "test"})
     mock_info = mock_logger.info
 
     backend.increment(_metric(Metric.INPUT_MESSAGES), 1)
@@ -261,7 +282,7 @@ def test_log_increment_logs_immediately(mock_logger: Any) -> None:
 
 @patch("sentry_streams.metrics.metrics.logger")
 def test_log_each_call_emits_separate_log_line(mock_logger: Any) -> None:
-    backend = LogMetricsBackend(period_sec=60.0)
+    backend = LogMetricsBackend()
     mock_info = mock_logger.info
 
     backend.increment(_metric(Metric.INPUT_MESSAGES), 1)
@@ -274,7 +295,7 @@ def test_log_each_call_emits_separate_log_line(mock_logger: Any) -> None:
 @patch("time.time")
 def test_buffered_log_accumulation_and_flush(mock_time: Any, mock_logger: Any) -> None:
     mock_time.return_value = 0.0
-    inner = LogMetricsBackend(period_sec=60.0)
+    inner = LogMetricsBackend()
     backend = BufferedMetricsBackend(inner, throttle_interval_sec=60.0)
     mock_info = mock_logger.info
 
@@ -297,7 +318,7 @@ def test_buffered_log_accumulation_and_flush(mock_time: Any, mock_logger: Any) -
 @patch("time.time")
 def test_buffered_log_flush_logs_and_clears(mock_time: Any, mock_logger: Any) -> None:
     mock_time.return_value = 0.0
-    inner = LogMetricsBackend(period_sec=60.0)
+    inner = LogMetricsBackend()
     backend = BufferedMetricsBackend(inner, throttle_interval_sec=60.0)
     mock_info = mock_logger.info
 
@@ -321,7 +342,7 @@ def test_buffered_log_flush_logs_and_clears(mock_time: Any, mock_logger: Any) ->
 @patch("time.time")
 def test_buffered_log_throttled_flush(mock_time: Any, mock_logger: Any) -> None:
     mock_time.return_value = 0.0
-    inner = LogMetricsBackend(period_sec=60.0)
+    inner = LogMetricsBackend()
     backend = BufferedMetricsBackend(inner, throttle_interval_sec=10.0)
     mock_info = mock_logger.info
 
@@ -337,7 +358,7 @@ def test_buffered_log_throttled_flush(mock_time: Any, mock_logger: Any) -> None:
 @patch("time.time")
 def test_buffered_log_global_tags_from_inner(mock_time: Any, mock_logger: Any) -> None:
     mock_time.return_value = 0.0
-    inner = LogMetricsBackend(period_sec=60.0, tags={"service": "streams"})
+    inner = LogMetricsBackend(tags={"service": "streams"})
     backend = BufferedMetricsBackend(inner, throttle_interval_sec=60.0)
     mock_info = mock_logger.info