diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index 96ca1b5..59d4345 100644 --- a/tests/test_otel_exporter_fanout.py +++ b/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() diff --git a/ventis/OTLP_Exporter/otel_exporter.py b/ventis/OTLP_Exporter/otel_exporter.py index eafb786..7e365b0 100644 --- a/ventis/OTLP_Exporter/otel_exporter.py +++ b/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) diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index 6bdbd1a..fc5dd38 100644 --- a/ventis/controller/global_controller.py +++ b/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: diff --git a/ventis/controller/utils/process_supervisor.py b/ventis/controller/utils/process_supervisor.py index 8c061bc..f5336e6 100644 --- a/ventis/controller/utils/process_supervisor.py +++ b/ventis/controller/utils/process_supervisor.py @@ -19,6 +19,9 @@ def __init__(self): self._specs = {} # name -> (argv, env) tuple self._procs = {} # name -> subprocess.Popen + def is_registered(self, name): + return name in self._specs + def register(self, name, argv, env=None): """Declare a process to manage. Does not start it -- call start_all() once everything is registered. `env`, if given, is merged on top of (not a