From b4e45e8bcf4ea516123b8a7eab68c936637c42ad Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 3 Sep 2026 12:32:00 -0700 Subject: [PATCH 1/3] Redis-backed otel destination reload: config change no longer needs full redeploy Moves otel.destinations from a one-shot VENTIS_OTEL_DESTINATIONS env var (frozen at exporter subprocess spawn) to a Redis key (otel:destinations), mirroring the existing routing-table live-reload pattern. GlobalController writes it at startup and again in reload_config() (SIGHUP); otel_exporter.py's existing 5s poll tick re-reads it each cycle and rebuilds its BatchSpanProcessors only when it changed. No signal-forwarding, no subprocess restart, no ProcessSupervisor.restart -- just a small ProcessSupervisor.is_registered() so reload_config knows whether the exporter is even running. Kept in scope: exporter start-gating at boot is unchanged (still skipped entirely if otel.destinations is absent at startup); destinations added after boot only take effect if the exporter was already running. --- tests/test_otel_exporter_fanout.py | 164 +++++++++++++----- ventis/OTLP_Exporter/otel_exporter.py | 69 ++++++-- ventis/controller/global_controller.py | 57 ++++-- ventis/controller/utils/process_supervisor.py | 5 + 4 files changed, 230 insertions(+), 65 deletions(-) diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index 96ca1b5..5231115 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,45 @@ 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) + self.assertIsNone(GlobalController._otel_destinations({})) - def test_controller_env_is_none_when_otel_not_configured(self): + def test_controller_exporter_env_carries_redis_connection_only(self): + # Destinations travel via Redis (otel:destinations), not env, so this + # is now just the fixed connection info the subprocess needs to reach it. from ventis.controller.global_controller import GlobalController - self.assertIsNone(GlobalController._otel_exporter_env({})) + env = GlobalController._otel_exporter_env( + {"host": "redis-host", "port": 6380, "db": 2} + ) + self.assertEqual( + env, + { + "VENTIS_REDIS_HOST": "redis-host", + "VENTIS_REDIS_PORT": "6380", + "VENTIS_REDIS_DB": "2", + }, + ) + + def test_controller_exporter_env_defaults(self): + from ventis.controller.global_controller import GlobalController + + env = GlobalController._otel_exporter_env({}) + self.assertEqual( + env, + { + "VENTIS_REDIS_HOST": "localhost", + "VENTIS_REDIS_PORT": "6379", + "VENTIS_REDIS_DB": "0", + }, + ) def _insert_pending_row(self): conn = sqlite3.connect(self.db_path) @@ -251,11 +262,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 +276,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..94c195e 100644 --- a/ventis/OTLP_Exporter/otel_exporter.py +++ b/ventis/OTLP_Exporter/otel_exporter.py @@ -5,8 +5,12 @@ 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. +GlobalController writes the resolved destination list to the ``otel:destinations`` Redis +key (required because the standard OTEL exporter environment variables describe only one +destination). Every poll tick also re-reads that key and rebuilds the configured +processors if it changed, so a config reload (SIGHUP -> GlobalController.reload_config) +reaches this process without a restart. Redis connection info itself, unlike +destinations, is fixed for the process's lifetime and passed once via env. """ import json @@ -15,8 +19,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 +41,20 @@ _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 + + +def _redis_client(): + return RedisClient( + host=os.environ.get("VENTIS_REDIS_HOST", "localhost"), + port=int(os.environ.get("VENTIS_REDIS_PORT", 6379)), + db=int(os.environ.get("VENTIS_REDIS_DB", 0)), + ) + + +_redis = None def _validate_destination(destination, index): @@ -86,17 +106,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 +150,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 +183,27 @@ def _handle_shutdown(signum, frame): _running = False +def _reload_destinations_if_changed(): + """Re-read otel:destinations from Redis; rebuild _processors if it changed. + Invalid or missing values are logged and the previous processors are kept + running, matching the poll loop's existing non-fatal error handling. + """ + 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 +254,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 = _redis_client() + _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..7bd1a5d 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,17 @@ 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: + + # Destinations live in Redis (self.OTEL_DESTINATIONS_KEY), not env: the exporter + # subprocess polls that key each cycle, so config changes (SIGHUP -> reload_config) + # reach it without a restart. Only the fixed Redis connection info is passed as env. + 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], + env=self._otel_exporter_env(redis_cfg), ) else: logger.info("otel.destinations not configured -- no OTel metrics collection will happen.") @@ -220,22 +226,40 @@ 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 global_controller.yaml's `otel.destinations` (expanding any + ${ENV_VAR} refs). Returns None if 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. """ if "destinations" not in otel_cfg: return None - destinations = GlobalController._expand_env_value(otel_cfg["destinations"]) + return GlobalController._expand_env_value(otel_cfg["destinations"]) + + @staticmethod + def _otel_exporter_env(redis_cfg): + """Env for the exporter subprocess: just enough to reach the same Redis + as this GlobalController. Fixed for the process's lifetime -- unlike + destinations, the Redis location itself isn't something a running + deploy can be reconfigured onto. + """ + return { + "VENTIS_REDIS_HOST": str(redis_cfg.get("host", "localhost")), + "VENTIS_REDIS_PORT": str(redis_cfg.get("port", 6379)), + "VENTIS_REDIS_DB": str(redis_cfg.get("db", 0)), + } + + def _write_otel_destinations(self, destinations): + """Push the resolved destination list to Redis. Raises if it can't be + JSON-serialized -- same validation the old env-var path had.""" 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 +288,13 @@ def reload_config(self): self._write_identity() self.instance_manager.publish_routing_snapshot(self.controllers) + # Refresh otel destinations too, same as the routing table above. Only + # meaningful if the exporter subprocess is already running (started at + # boot) -- it isn't spawned mid-run just because otel got added 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..44233c1 100644 --- a/ventis/controller/utils/process_supervisor.py +++ b/ventis/controller/utils/process_supervisor.py @@ -19,6 +19,11 @@ def __init__(self): self._specs = {} # name -> (argv, env) tuple self._procs = {} # name -> subprocess.Popen + def is_registered(self, name): + """Whether `name` was ever registered (regardless of whether it's still + running -- see check_and_respawn for restarts).""" + 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 From 9f630fe163cedcaba54828d2839ea5876797a5db Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 3 Sep 2026 12:54:28 -0700 Subject: [PATCH 2/3] Simplify: assume otel_exporter's Redis is always localhost:6379 Drops the redis-connection-info env plumbing (VENTIS_REDIS_HOST/PORT/DB, GlobalController._otel_exporter_env) added in the previous commit -- otel_exporter and GlobalController always run on the same host, and RedisClient's own defaults already are localhost:6379/db0, so passing them through was dead flexibility for a case that doesn't exist yet. --- tests/test_otel_exporter_fanout.py | 30 -------------------------- ventis/OTLP_Exporter/otel_exporter.py | 16 +++----------- ventis/controller/global_controller.py | 20 +++-------------- 3 files changed, 6 insertions(+), 60 deletions(-) diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index 5231115..59d4345 100644 --- a/tests/test_otel_exporter_fanout.py +++ b/tests/test_otel_exporter_fanout.py @@ -168,36 +168,6 @@ def test_controller_destinations_is_none_when_otel_not_configured(self): self.assertIsNone(GlobalController._otel_destinations({})) - def test_controller_exporter_env_carries_redis_connection_only(self): - # Destinations travel via Redis (otel:destinations), not env, so this - # is now just the fixed connection info the subprocess needs to reach it. - from ventis.controller.global_controller import GlobalController - - env = GlobalController._otel_exporter_env( - {"host": "redis-host", "port": 6380, "db": 2} - ) - self.assertEqual( - env, - { - "VENTIS_REDIS_HOST": "redis-host", - "VENTIS_REDIS_PORT": "6380", - "VENTIS_REDIS_DB": "2", - }, - ) - - def test_controller_exporter_env_defaults(self): - from ventis.controller.global_controller import GlobalController - - env = GlobalController._otel_exporter_env({}) - self.assertEqual( - env, - { - "VENTIS_REDIS_HOST": "localhost", - "VENTIS_REDIS_PORT": "6379", - "VENTIS_REDIS_DB": "0", - }, - ) - def _insert_pending_row(self): conn = sqlite3.connect(self.db_path) try: diff --git a/ventis/OTLP_Exporter/otel_exporter.py b/ventis/OTLP_Exporter/otel_exporter.py index 94c195e..58704b0 100644 --- a/ventis/OTLP_Exporter/otel_exporter.py +++ b/ventis/OTLP_Exporter/otel_exporter.py @@ -9,8 +9,8 @@ key (required because the standard OTEL exporter environment variables describe only one destination). Every poll tick also re-reads that key and rebuilds the configured processors if it changed, so a config reload (SIGHUP -> GlobalController.reload_config) -reaches this process without a restart. Redis connection info itself, unlike -destinations, is fixed for the process's lifetime and passed once via env. +reaches this process without a restart. Redis itself is assumed to be on localhost:6379, +same as GlobalController's own default -- both run on the same host. """ import json @@ -44,16 +44,6 @@ _last_destinations_raw = None POLL_INTERVAL_SECONDS = 5 DESTINATIONS_KEY = "otel:destinations" # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY - - -def _redis_client(): - return RedisClient( - host=os.environ.get("VENTIS_REDIS_HOST", "localhost"), - port=int(os.environ.get("VENTIS_REDIS_PORT", 6379)), - db=int(os.environ.get("VENTIS_REDIS_DB", 0)), - ) - - _redis = None @@ -258,7 +248,7 @@ def main(): signal.signal(signal.SIGTERM, _handle_shutdown) signal.signal(signal.SIGINT, _handle_shutdown) db.init_db() - _redis = _redis_client() + _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)) diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index 7bd1a5d..85d2e23 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -125,14 +125,13 @@ def __init__(self, config_path): # Destinations live in Redis (self.OTEL_DESTINATIONS_KEY), not env: the exporter # subprocess polls that key each cycle, so config changes (SIGHUP -> reload_config) - # reach it without a restart. Only the fixed Redis connection info is passed as env. + # reach it without a restart. Redis itself is assumed to be localhost:6379 (the + # exporter's own RedisClient default) -- no connection env needed. 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=self._otel_exporter_env(redis_cfg), + "otel_exporter", [sys.executable, otel_exporter_script] ) else: logger.info("otel.destinations not configured -- no OTel metrics collection will happen.") @@ -237,19 +236,6 @@ def _otel_destinations(otel_cfg): return None return GlobalController._expand_env_value(otel_cfg["destinations"]) - @staticmethod - def _otel_exporter_env(redis_cfg): - """Env for the exporter subprocess: just enough to reach the same Redis - as this GlobalController. Fixed for the process's lifetime -- unlike - destinations, the Redis location itself isn't something a running - deploy can be reconfigured onto. - """ - return { - "VENTIS_REDIS_HOST": str(redis_cfg.get("host", "localhost")), - "VENTIS_REDIS_PORT": str(redis_cfg.get("port", 6379)), - "VENTIS_REDIS_DB": str(redis_cfg.get("db", 0)), - } - def _write_otel_destinations(self, destinations): """Push the resolved destination list to Redis. Raises if it can't be JSON-serialized -- same validation the old env-var path had.""" From c085dbc86649d14f06eaa4cc787485aa7b71bcd5 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 3 Sep 2026 12:57:17 -0700 Subject: [PATCH 3/3] Trim explanatory comments off simple/obvious functions Kept comments only where behavior is genuinely non-obvious (why the exporter polls Redis instead of restarting, why reload_config gates on is_registered, the invalid-update-keeps-old-processors fallback). Dropped comments/docstrings that just narrated 'this was added' on trivial pass-through code. --- ventis/OTLP_Exporter/otel_exporter.py | 15 +++++--------- ventis/controller/global_controller.py | 20 +++++-------------- ventis/controller/utils/process_supervisor.py | 2 -- 3 files changed, 10 insertions(+), 27 deletions(-) diff --git a/ventis/OTLP_Exporter/otel_exporter.py b/ventis/OTLP_Exporter/otel_exporter.py index 58704b0..7e365b0 100644 --- a/ventis/OTLP_Exporter/otel_exporter.py +++ b/ventis/OTLP_Exporter/otel_exporter.py @@ -5,12 +5,9 @@ all processors accept it. Batching, OTLP serialization, and sending remain the SDK's responsibility (see DESIGN.md). -GlobalController writes the resolved destination list to the ``otel:destinations`` Redis -key (required because the standard OTEL exporter environment variables describe only one -destination). Every poll tick also re-reads that key and rebuilds the configured -processors if it changed, so a config reload (SIGHUP -> GlobalController.reload_config) -reaches this process without a restart. Redis itself is assumed to be on localhost:6379, -same as GlobalController's own default -- both run on the same host. +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 @@ -174,10 +171,8 @@ def _handle_shutdown(signum, frame): def _reload_destinations_if_changed(): - """Re-read otel:destinations from Redis; rebuild _processors if it changed. - Invalid or missing values are logged and the previous processors are kept - running, matching the poll loop's existing non-fatal error handling. - """ + # 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: diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index 85d2e23..fc5dd38 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -123,10 +123,8 @@ def __init__(self, config_path): otel_exporter_script = os.path.join(otel_exporter_dir, "otel_exporter.py") self.process_supervisor = ProcessSupervisor() - # Destinations live in Redis (self.OTEL_DESTINATIONS_KEY), not env: the exporter - # subprocess polls that key each cycle, so config changes (SIGHUP -> reload_config) - # reach it without a restart. Redis itself is assumed to be localhost:6379 (the - # exporter's own RedisClient default) -- no connection env needed. + # 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) @@ -226,19 +224,12 @@ def _expand_env_value(value): @staticmethod def _otel_destinations(otel_cfg): - """Resolve global_controller.yaml's `otel.destinations` (expanding any - ${ENV_VAR} refs). Returns None if 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. - """ + """Resolve otel.destinations (${ENV_VAR} refs expanded), or None if absent.""" if "destinations" not in otel_cfg: return None return GlobalController._expand_env_value(otel_cfg["destinations"]) def _write_otel_destinations(self, destinations): - """Push the resolved destination list to Redis. Raises if it can't be - JSON-serialized -- same validation the old env-var path had.""" try: payload = json.dumps(destinations) except (TypeError, ValueError) as exc: @@ -274,9 +265,8 @@ def reload_config(self): self._write_identity() self.instance_manager.publish_routing_snapshot(self.controllers) - # Refresh otel destinations too, same as the routing table above. Only - # meaningful if the exporter subprocess is already running (started at - # boot) -- it isn't spawned mid-run just because otel got added here. + # 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) diff --git a/ventis/controller/utils/process_supervisor.py b/ventis/controller/utils/process_supervisor.py index 44233c1..f5336e6 100644 --- a/ventis/controller/utils/process_supervisor.py +++ b/ventis/controller/utils/process_supervisor.py @@ -20,8 +20,6 @@ def __init__(self): self._procs = {} # name -> subprocess.Popen def is_registered(self, name): - """Whether `name` was ever registered (regardless of whether it's still - running -- see check_and_respawn for restarts).""" return name in self._specs def register(self, name, argv, env=None):