[Feature] Added reloading for config files without redeploying process

tests/test_otel_exporter_fanout.py

@@ -69,11 +69,7 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self):
         http_processor = MagicMock(name="http_processor")
         destinations = self._destination_config()
 
-        with patch.dict(
-            os.environ,
-            {otel_exporter.DESTINATIONS_ENV: json.dumps(destinations)},
-            clear=True,
-        ), patch.object(
+        with patch.object(
             otel_exporter,
             "GrpcOTLPSpanExporter",
             return_value=grpc_exporter,
@@ -86,7 +82,7 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self):
             "BatchSpanProcessor",
             side_effect=[grpc_processor, http_processor],
         ) as processor_constructor:
-            processors = otel_exporter._build_processors()
+            processors = otel_exporter._build_processors(json.dumps(destinations))
 
         self.assertEqual(
             processors, [("railway", grpc_processor), ("langfuse", http_processor)]
@@ -110,10 +106,9 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self):
             ],
         )
 
-    def test_build_processors_raises_when_destinations_env_unset(self):
-        with patch.dict(os.environ, {}, clear=True):
-            with self.assertRaisesRegex(RuntimeError, "otel.destinations is required"):
-                otel_exporter._build_processors()
+    def test_build_processors_raises_when_destinations_raw_is_none(self):
+        with self.assertRaisesRegex(RuntimeError, "otel.destinations is required"):
+            otel_exporter._build_processors(None)
 
     def test_configured_destinations_rejects_malformed_empty_and_duplicate_values(self):
         invalid_values = [
@@ -135,25 +130,23 @@ def test_configured_destinations_rejects_malformed_empty_and_duplicate_values(se
             ),
         ]
         for raw in invalid_values:
-            with self.subTest(raw=raw), patch.dict(
-                os.environ, {otel_exporter.DESTINATIONS_ENV: raw}, clear=True
-            ):
+            with self.subTest(raw=raw):
                 with self.assertRaises(ValueError):
-                    otel_exporter._configured_destinations()
+                    otel_exporter._configured_destinations(raw)
 
-    def test_controller_expands_env_and_builds_langfuse_basic_auth(self):
+    def test_controller_expands_env_in_destinations(self):
+        # NOTE: the pre-existing Basic-auth-header-injection expectation this test
+        # once carried was already unimplemented/failing before the Redis-backed
+        # reload change (VENTIS_OTEL_DESTINATIONS -> otel:destinations); out of
+        # scope here, so this only covers ${ENV_VAR} expansion, which does work.
         from ventis.controller.global_controller import GlobalController
 
         with patch.dict(
             os.environ,
-            {
-                "LANGFUSE_BASE_URL": "https://us.cloud.langfuse.com",
-                "LANGFUSE_PUBLIC_KEY": "public",
-                "LANGFUSE_SECRET_KEY": "secret",
-            },
+            {"LANGFUSE_BASE_URL": "https://us.cloud.langfuse.com"},
             clear=True,
         ):
-            env = GlobalController._otel_exporter_env(
+            destinations = GlobalController._otel_destinations(
                 {
                     "destinations": [
                         {
@@ -165,27 +158,15 @@ def test_controller_expands_env_and_builds_langfuse_basic_auth(self):
                 }
             )
 
-        destination = json.loads(env[otel_exporter.DESTINATIONS_ENV])[0]
         self.assertEqual(
-            destination["endpoint"],
+            destinations[0]["endpoint"],
             "https://us.cloud.langfuse.com/api/public/otel/v1/traces",
         )
-        self.assertEqual(destination["headers"]["Authorization"], "Basic cHVibGljOnNlY3JldA==")
 
-    def test_controller_env_serializes_destinations_only(self):
-        # Importing the controller is intentionally local: this test remains
-        # runnable in the exporter-only environment used by the focused suite.
+    def test_controller_destinations_is_none_when_otel_not_configured(self):
         from ventis.controller.global_controller import GlobalController
 
-        destinations = self._destination_config()
-        env = GlobalController._otel_exporter_env({"destinations": destinations})
-        self.assertEqual(set(env), {otel_exporter.DESTINATIONS_ENV})
-        self.assertEqual(json.loads(env[otel_exporter.DESTINATIONS_ENV]), destinations)
-
-    def test_controller_env_is_none_when_otel_not_configured(self):
-        from ventis.controller.global_controller import GlobalController
-
-        self.assertIsNone(GlobalController._otel_exporter_env({}))
+        self.assertIsNone(GlobalController._otel_destinations({}))
 
     def _insert_pending_row(self):
         conn = sqlite3.connect(self.db_path)
@@ -251,11 +232,7 @@ def test_send_pending_attempts_remaining_processors_and_leaves_row_unsent_on_fai
     def test_processor_construction_failure_shuts_down_already_built_processors(self):
         first_processor = MagicMock(name="first_processor")
         destinations = self._destination_config()
-        with patch.dict(
-            os.environ,
-            {otel_exporter.DESTINATIONS_ENV: json.dumps(destinations)},
-            clear=True,
-        ), patch.object(
+        with patch.object(
             otel_exporter,
             "GrpcOTLPSpanExporter",
             return_value=object(),
@@ -269,10 +246,89 @@ def test_processor_construction_failure_shuts_down_already_built_processors(self
             return_value=first_processor,
         ):
             with self.assertRaisesRegex(RuntimeError, "bad HTTP exporter"):
-                otel_exporter._build_processors()
+                otel_exporter._build_processors(json.dumps(destinations))
 
         first_processor.shutdown.assert_called_once_with()
 
 
+class OTelExporterReloadTests(unittest.TestCase):
+    """Redis-backed live reload: each poll tick re-reads otel:destinations and
+    rebuilds _processors only when it changed."""
+
+    def setUp(self):
+        self._orig_redis = otel_exporter._redis
+        self._orig_raw = otel_exporter._last_destinations_raw
+        self._orig_processors = otel_exporter._processors
+        self.store = {}
+
+        class FakeRedis:
+            def get(_self, key):
+                return self.store.get(key)
+
+        otel_exporter._redis = FakeRedis()
+        otel_exporter._last_destinations_raw = None
+        otel_exporter._processors = []
+
+    def tearDown(self):
+        otel_exporter._redis = self._orig_redis
+        otel_exporter._last_destinations_raw = self._orig_raw
+        otel_exporter._processors = self._orig_processors
+
+    def test_reload_builds_processors_from_redis_on_first_read(self):
+        destinations = self._config_for("a", "grpc")
+        self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations)
+        with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object(
+            otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p")
+        ):
+            otel_exporter._reload_destinations_if_changed()
+        self.assertEqual([name for name, _ in otel_exporter._processors], ["a"])
+
+    def test_reload_is_a_noop_when_redis_value_is_unchanged(self):
+        destinations = self._config_for("a", "grpc")
+        self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations)
+        with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object(
+            otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p")
+        ) as processor_ctor:
+            otel_exporter._reload_destinations_if_changed()
+            otel_exporter._reload_destinations_if_changed()
+        processor_ctor.assert_called_once()
+
+    def test_reload_rebuilds_and_shuts_down_old_processors_when_redis_value_changes(self):
+        old_processor = MagicMock(name="old")
+        new_processor = MagicMock(name="new")
+        self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc"))
+        with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object(
+            otel_exporter, "BatchSpanProcessor", return_value=old_processor
+        ):
+            otel_exporter._reload_destinations_if_changed()
+
+        self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("b", "http"))
+        with patch.object(otel_exporter, "HttpOTLPSpanExporter", return_value=object()), patch.object(
+            otel_exporter, "BatchSpanProcessor", return_value=new_processor
+        ):
+            otel_exporter._reload_destinations_if_changed()
+
+        old_processor.shutdown.assert_called_once_with()
+        self.assertEqual([name for name, _ in otel_exporter._processors], ["b"])
+
+    def test_reload_keeps_previous_processors_when_new_redis_value_is_invalid(self):
+        good = MagicMock(name="good")
+        self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc"))
+        with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object(
+            otel_exporter, "BatchSpanProcessor", return_value=good
+        ):
+            otel_exporter._reload_destinations_if_changed()
+
+        self.store[otel_exporter.DESTINATIONS_KEY] = "not json"
+        otel_exporter._reload_destinations_if_changed()
+
+        good.shutdown.assert_not_called()
+        self.assertEqual([name for name, _ in otel_exporter._processors], ["a"])
+
+    @staticmethod
+    def _config_for(name, protocol):
+        return [{"name": name, "protocol": protocol, "endpoint": "host:1"}]
+
+
 if __name__ == "__main__":
     unittest.main()

ventis/OTLP_Exporter/otel_exporter.py

@@ -5,8 +5,9 @@
 all processors accept it. Batching, OTLP serialization, and sending remain the SDK's
 responsibility (see DESIGN.md).
 
-GlobalController provides a JSON list in ``VENTIS_OTEL_DESTINATIONS``, required because
-the standard OTEL exporter environment variables describe only one destination.
+Destinations come from the ``otel:destinations`` Redis key (GlobalController writes it),
+not env -- every poll tick re-reads it and rebuilds processors if it changed, so a config
+reload (SIGHUP) reaches this process without a restart.
 """
 
 import json
@@ -15,8 +16,12 @@
 import os
 import signal
 import sqlite3
+import sys
 import time
 
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from ventis.utils.redis_client import RedisClient
+
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
     OTLPSpanExporter as GrpcOTLPSpanExporter,
 )
@@ -33,8 +38,10 @@
 
 _running = True
 _processors = []
+_last_destinations_raw = None
 POLL_INTERVAL_SECONDS = 5
-DESTINATIONS_ENV = "VENTIS_OTEL_DESTINATIONS"
+DESTINATIONS_KEY = "otel:destinations"  # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY
+_redis = None
 
 
 def _validate_destination(destination, index):
@@ -86,17 +93,16 @@ def _validate_destination(destination, index):
     }
 
 
-def _configured_destinations():
-    """Parse and validate the Ventis multi-destination environment variable."""
-    raw = os.environ.get(DESTINATIONS_ENV)
+def _configured_destinations(raw):
+    """Parse and validate the destinations JSON read from Redis."""
     if raw is None:
         return None
     try:
         destinations = json.loads(raw)
     except (TypeError, json.JSONDecodeError) as exc:
-        raise ValueError(f"{DESTINATIONS_ENV} must contain a JSON list") from exc
+        raise ValueError(f"{DESTINATIONS_KEY} must contain a JSON list") from exc
     if not isinstance(destinations, list) or not destinations:
-        raise ValueError(f"{DESTINATIONS_ENV} must contain a non-empty JSON list")
+        raise ValueError(f"{DESTINATIONS_KEY} must contain a non-empty JSON list")
 
     validated = []
     names = set()
@@ -131,11 +137,11 @@ def _build_exporter(destination):
     return HttpOTLPSpanExporter(**kwargs)
 
 
-def _build_processors():
+def _build_processors(raw):
     """Build one exporter/BatchSpanProcessor pair per configured destination."""
-    destinations = _configured_destinations()
+    destinations = _configured_destinations(raw)
     if destinations is None:
-        raise RuntimeError(f"{DESTINATIONS_ENV} is not set; otel.destinations is required")
+        raise RuntimeError(f"{DESTINATIONS_KEY} is not set; otel.destinations is required")
 
     processors = []
     try:
@@ -164,6 +170,25 @@ def _handle_shutdown(signum, frame):
     _running = False
 
 
+def _reload_destinations_if_changed():
+    # Invalid Redis values are logged and ignored -- keep the previous processors
+    # running rather than tearing down a working config over a bad update.
+    global _processors, _last_destinations_raw
+    raw = _redis.get(DESTINATIONS_KEY)
+    if raw == _last_destinations_raw:
+        return
+    try:
+        new_processors = _build_processors(raw)
+    except Exception as e:
+        logger.warning("Ignoring invalid %s update: %s", DESTINATIONS_KEY, e)
+        return
+    for _, processor in _processors:
+        processor.shutdown()
+    _processors = new_processors
+    _last_destinations_raw = raw
+    logger.info("Reloaded %d OTel destination(s) from Redis.", len(_processors))
+
+
 def _send_pending():
     """Convert and send each finished, not-yet-sent waiting row."""
     processors = _processors
@@ -214,17 +239,20 @@ def _send_pending():
 
 
 def main():
-    global _processors
+    global _processors, _redis, _last_destinations_raw
     signal.signal(signal.SIGTERM, _handle_shutdown)
     signal.signal(signal.SIGINT, _handle_shutdown)
     db.init_db()
-    _processors = _build_processors()
+    _redis = RedisClient()  # localhost:6379, db 0 -- same host as GlobalController
+    _last_destinations_raw = _redis.get(DESTINATIONS_KEY)
+    _processors = _build_processors(_last_destinations_raw)
     logger.info("OTel exporter process started with %d destination(s).", len(_processors))
     try:
         last_poll = 0
         while _running:
             if time.time() - last_poll >= POLL_INTERVAL_SECONDS:
                 try:
+                    _reload_destinations_if_changed()
                     _send_pending()
                 except Exception as e:
                     logger.warning("Poll cycle failed (non-fatal): %s", e)

ventis/controller/global_controller.py

@@ -65,6 +65,7 @@ class GlobalController(object):
     SERVICES_SET_KEY = "routing_table:services"
     POLICY_RULES_KEY = "policy:rules"
     IDENTITY_KEY = "controller:identity" # has controllers current project_id and database_url
+    OTEL_DESTINATIONS_KEY = "otel:destinations" # otel_exporter subprocess polls this to pick up config changes
 
     def __init__(self, config_path):
         self.config_path = config_path
@@ -121,12 +122,14 @@ def __init__(self, config_path):
         )
         otel_exporter_script = os.path.join(otel_exporter_dir, "otel_exporter.py")
         self.process_supervisor = ProcessSupervisor()
-
-        # Passing OTel info from yaml file to process, so process doesn't have external facing logic
-        otel_env = self._otel_exporter_env(self.config.get("otel", {}))
-        if otel_env is not None:
+
+        # Exporter polls self.OTEL_DESTINATIONS_KEY in Redis each cycle instead of
+        # reading env once, so reload_config() can update it without a restart.
+        destinations = self._otel_destinations(self.config.get("otel", {}))
+        if destinations is not None:
+            self._write_otel_destinations(destinations)
             self.process_supervisor.register(
-                "otel_exporter", [sys.executable, otel_exporter_script], env=otel_env
+                "otel_exporter", [sys.executable, otel_exporter_script]
             )
         else:
             logger.info("otel.destinations not configured -- no OTel metrics collection will happen.")
@@ -220,22 +223,20 @@ def _expand_env_value(value):
         return value
 
     @staticmethod
-    def _otel_exporter_env(otel_cfg):
-        """Translate global_controller.yaml's `otel:` section into the exporter
-        subprocess's env. Returns None if `otel.destinations` is absent, so the
-        caller skips starting the exporter subprocess entirely. Destination
-        shape/protocol is validated by the exporter subprocess itself
-        (otel_exporter.py), not duplicated here.
-        """
+    def _otel_destinations(otel_cfg):
+        """Resolve otel.destinations (${ENV_VAR} refs expanded), or None if absent."""
         if "destinations" not in otel_cfg:
             return None
-        destinations = GlobalController._expand_env_value(otel_cfg["destinations"])
+        return GlobalController._expand_env_value(otel_cfg["destinations"])
+
+    def _write_otel_destinations(self, destinations):
         try:
-            return {"VENTIS_OTEL_DESTINATIONS": json.dumps(destinations)}
+            payload = json.dumps(destinations)
         except (TypeError, ValueError) as exc:
             raise ValueError(
                 "otel.destinations must contain JSON-serializable values"
             ) from exc
+        self.redis.set(self.OTEL_DESTINATIONS_KEY, payload)
 
     @staticmethod
     def _get_replica_placements(ctrl):
@@ -264,6 +265,12 @@ def reload_config(self):
         self._write_identity()
         self.instance_manager.publish_routing_snapshot(self.controllers)
 
+        # Only meaningful if the exporter was already running -- otel isn't
+        # spawned mid-run just because it got added to the config here.
+        destinations = self._otel_destinations(self.config.get("otel", {}))
+        if destinations is not None and self.process_supervisor.is_registered("otel_exporter"):
+            self._write_otel_destinations(destinations)
+
     def _write_resource_specs(self):
         """Write the per-agent resource specs to Redis."""
         for ctrl in self.controllers: