diff --git a/canyonos_core/controller/global_controller.py b/canyonos_core/controller/global_controller.py index 5a330a1..5d0d0c1 100644 --- a/canyonos_core/controller/global_controller.py +++ b/canyonos_core/controller/global_controller.py @@ -26,6 +26,7 @@ from canyonos_core.controller.utils.telemetry_logging import ( assign_project_id, pull_runtime_information, + resolve_database_url, send_runtime_information, send_agent_information, ) @@ -102,6 +103,12 @@ def __init__(self, config_path): self._lc_stubs = {} # endpoint -> gRPC stub self.instance_manager = InstanceManager(self) assign_project_id(self.config.get("project_id")) + if self._database_url() is None: + logger.info( + "No database configured; telemetry writes are disabled. " + "Set database.url in %s to record runtime and agent information.", + config_path, + ) # Clean up any stale containers from previous runs self._cleanup_stale_containers() @@ -562,6 +569,14 @@ def run(self): except KeyboardInterrupt: self.stop() + def _database_url(self): + """The configured database URL, or None when there is no database to write to. + + Read per call, not cached, so reload_config() can point us at a new database. + `database:` with nothing under it parses as None, hence the `or {}`. + """ + return resolve_database_url((self.config.get("database") or {}).get("url")) + def _poll_controllers(self): """ Check the health of each registered controller replica via its node's Redis. @@ -589,17 +604,19 @@ def _poll_one_instance(self, instance): logger.warning("Failed to poll instance %s: %s", instance, e) return + # Without a database the legacy telemetry writes have nowhere to go, so they + # are skipped instead of failing on every poll; OTel export is independent + # of that legacy database and always runs. + database_url = self._database_url() + try: future_rows = pull_runtime_information(node_redis) self._otel_db.write_waiting_rows( future_rows, node_redis, self.config.get("project_id") ) - # This is now legacy, keeping it for now, but will remove this later - send_runtime_information( - future_rows, - node_redis, - self.config.get("database", {}).get("url"), - ) + if database_url: + # This is now legacy, keeping it for now, but will remove this later + send_runtime_information(future_rows, node_redis, database_url) except Exception as e: logger.warning( "Failed to write runtime information for instance %s (%s:%s) " @@ -626,33 +643,38 @@ def _poll_one_instance(self, instance): throughput = requests_served / elapsed if elapsed > 0 else 0.0 self._last_metrics_poll_time[(host, port)] = now - try: - send_agent_information( - [ + if database_url: + try: + send_agent_information( + [ + { + **instance, + **metrics, + "requests_served": requests_served, + "throughput": throughput, + } + ], + database_url, + ) + except Exception as e: + logger.warning( + "Failed to write agent information for instance %s (%s:%s) " + "(non-fatal): %s", + name, + host, + port, + e, + ) + else: + # Only clear the accumulated counters once they've actually been persisted + node_redis.hset_multiple( + metrics_key, { - **instance, - **metrics, - "requests_served": requests_served, - "throughput": throughput, - } - ], - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write agent information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, - ) - else: - # Only clear the accumulated counters once they've actually been persisted - node_redis.hset_multiple( - metrics_key, - {"full_failures": 0, "error_count": 0, "requests_served": 0}, - ) + "full_failures": 0, + "error_count": 0, + "requests_served": 0, + }, + ) except Exception as e: logger.warning( "Failed to poll metrics for instance %s (%s:%s): %s", diff --git a/canyonos_core/controller/utils/telemetry_logging.py b/canyonos_core/controller/utils/telemetry_logging.py index dbb15f5..5e91871 100644 --- a/canyonos_core/controller/utils/telemetry_logging.py +++ b/canyonos_core/controller/utils/telemetry_logging.py @@ -87,10 +87,17 @@ def assign_project_id(project_id) -> None: global _project_id _project_id = project_id + +def resolve_database_url(database_url): + """The env var wins over the config value, as it always has; empty means no database.""" + url = os.environ.get("CANYONOS_DATABASE_URL") or str(database_url or "") + return url.strip() or None + + def _get_engine(database_url): global _engine if _engine is None: - url = os.environ.get("CANYONOS_DATABASE_URL", str(database_url)) + url = resolve_database_url(database_url) or "" if url.startswith("postgresql://"): url = "postgresql+psycopg://" + url[len("postgresql://"):] _engine = create_engine(url) diff --git a/tests/test_global_controller_telemetry_skip.py b/tests/test_global_controller_telemetry_skip.py new file mode 100644 index 0000000..e4aedef --- /dev/null +++ b/tests/test_global_controller_telemetry_skip.py @@ -0,0 +1,139 @@ +"""CAN-283: with no `database` in the config, every metrics poll used to log + + Failed to write agent information for instance (localhost:800N) (non-fatal): + Could not parse SQLAlchemy URL from given URL string + +once per instance, every poll_interval (5s by default), drowning out the errors an +operator actually needs to read. Telemetry has nowhere to go without a database, so +the legacy writes -- and the metrics that feed them -- are skipped outright. OTel +export is a separate destination and is unaffected. +""" + +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from canyonos_core.controller.global_controller import GlobalController +from canyonos_core.controller.utils.telemetry_logging import resolve_database_url + + +class EnvIsolatedTestCase(unittest.TestCase): + def setUp(self): + self._saved = os.environ.pop("CANYONOS_DATABASE_URL", None) + + def tearDown(self): + os.environ.pop("CANYONOS_DATABASE_URL", None) + if self._saved is not None: + os.environ["CANYONOS_DATABASE_URL"] = self._saved + + +class ResolveDatabaseUrlTests(EnvIsolatedTestCase): + def test_nothing_configured_resolves_to_none(self): + # config.get("database", {}).get("url") yields None when the key is absent -- + # str(None) == "None" is what SQLAlchemy used to choke on. + for value in (None, "", " "): + self.assertIsNone(resolve_database_url(value), repr(value)) + + def test_env_var_wins_over_the_config_value(self): + os.environ["CANYONOS_DATABASE_URL"] = "sqlite:///env.db" + self.assertEqual( + resolve_database_url("postgresql://cfg-host/db"), "sqlite:///env.db" + ) + + def test_empty_env_var_falls_back_to_the_config_value(self): + os.environ["CANYONOS_DATABASE_URL"] = "" + self.assertEqual(resolve_database_url("sqlite:///cfg.db"), "sqlite:///cfg.db") + + +class PollTelemetryTests(EnvIsolatedTestCase): + """_poll_one_instance must not touch legacy telemetry when there is no database URL.""" + + def _controller(self, config): + controller = GlobalController.__new__(GlobalController) + controller.config = config + controller.poll_interval = 5 + controller._last_status = {} + controller._last_metrics_poll_time = {} + controller._on_controller_healthy = MagicMock() + controller._on_controller_unhealthy = MagicMock() + controller._otel_db = MagicMock() + controller.instance_manager = MagicMock() + instance = { + "agent_name": "AgentA", + "agent_id": "local:AgentA:0", + "host": "localhost", + "host_port": 8000, + } + controller.instance_manager.list_instances.return_value = [instance] + controller.instance_manager._routing_endpoint_for = MagicMock( + return_value="local:AgentA:0" + ) + self.node_redis = MagicMock() + self.node_redis.hgetall.return_value = {"requests_served": "3"} + self.node_redis.get.return_value = "healthy" + controller._get_node_redis_for = MagicMock(return_value=self.node_redis) + return controller + + def _poll(self, controller): + instance = controller.instance_manager.list_instances()[0] + with patch( + "canyonos_core.controller.global_controller.send_runtime_information" + ) as send_runtime, patch( + "canyonos_core.controller.global_controller.send_agent_information" + ) as send_agent, patch( + "canyonos_core.controller.global_controller.pull_runtime_information" + ) as pull_runtime, self.assertLogs( + "canyonos_core.controller.global_controller", level="INFO" + ) as logs: + controller._poll_one_instance(instance) + return send_runtime, send_agent, pull_runtime, logs.output + + def _assert_telemetry_skipped(self, config): + controller = self._controller(config) + + send_runtime, send_agent, pull_runtime, output = self._poll(controller) + + send_runtime.assert_not_called() + send_agent.assert_not_called() + # OTel export has nowhere else to get its rows from, so it still runs. + pull_runtime.assert_called_once() + controller._otel_db.write_waiting_rows.assert_called_once() + self.assertEqual([line for line in output if "WARNING" in line], []) + # Nothing was persisted, so the counters stay where they are. + self.node_redis.hset_multiple.assert_not_called() + + def test_a_missing_database_key_skips_telemetry(self): + self._assert_telemetry_skipped({"agents": []}) + + def test_an_empty_database_block_is_treated_as_no_database(self): + # `database:` with nothing under it parses as None, not {}. + self._assert_telemetry_skipped({"agents": [], "database": None}) + + def test_a_configured_database_still_gets_both_writes(self): + controller = self._controller( + {"agents": [], "database": {"url": "postgresql://user@host/db"}} + ) + + send_runtime, send_agent, _, output = self._poll(controller) + + self.assertEqual(send_runtime.call_args.args[2], "postgresql://user@host/db") + self.assertEqual(send_agent.call_args.args[1], "postgresql://user@host/db") + self.assertEqual([line for line in output if "WARNING" in line], []) + # Counters are cleared only once the row has actually been persisted. + self.node_redis.hset_multiple.assert_called_once() + + def test_the_env_var_alone_is_enough_to_enable_writes(self): + os.environ["CANYONOS_DATABASE_URL"] = "sqlite:///env.db" + controller = self._controller({"agents": []}) + + send_runtime, send_agent, _, _ = self._poll(controller) + + send_runtime.assert_called_once() + send_agent.assert_called_once() + + +if __name__ == "__main__": + unittest.main()