From 391268776cc3d55636bdc2022294d4b137364e33 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 13:51:18 -0300 Subject: [PATCH] feat(gateway): wire engines+adapters via GatewayRuntime (write/gRPC live; subscription v2 gated on 1974) GatewayRuntime builds and owns the F4 object graph (CertProvider/pool -> WriteEngine/NatsWriteConsumer -> ActiveReadEngine -> optional GrpcServer), reusing the existing ingestor's logger/notification_handler/metrics_controller and DataManager (CertProvider shares its Mongo connection instead of opening a second one). app.py starts it after prepare_ingestor, sweeps idle OPC connections each loop iteration, and stops it on shutdown. A NATS connect failure at startup is logged/notified but never crashes the app -- the read path keeps working and idle_sweep() retries the consumer start. ingestor.py needed no changes: data_manager is already reachable via ingestor.ingestor_manager.data_manager. Also documents the new MinIO/Postgres env vars this wiring reads (via sientia_do's build_postgres_config for Postgres) in .env.example, and updates test_app.py's fixture to mock GatewayConfig/GatewayRuntime so the existing app tests keep exercising only what they mocked before. --- .env.example | 16 +++ ingestor/app.py | 21 ++++ ingestor/gateway_runtime.py | 149 +++++++++++++++++++++++++++ tests/unit/test_app.py | 29 ++++++ tests/unit/test_gateway_runtime.py | 156 +++++++++++++++++++++++++++++ 5 files changed, 371 insertions(+) create mode 100644 ingestor/gateway_runtime.py create mode 100644 tests/unit/test_gateway_runtime.py diff --git a/.env.example b/.env.example index b69ece3..27003d2 100644 --- a/.env.example +++ b/.env.example @@ -28,3 +28,19 @@ WRITE_QUALITY="true" WRITE_MIN_PAUSE_S="0.0" IDLE_CONN_MAX_S="300.0" +# OPC Gateway - MinIO (client cert storage, read by GatewayRuntime/CertProvider) +MINIO_ENDPOINT="localhost:9000" +MINIO_ACCESS_KEY="minioadmin" +MINIO_SECRET_KEY="minioadmin" +MINIO_SECURE="false" + +# OPC Gateway - Postgres (write_status updates, read by GatewayRuntime/DataSink +# via sientia_do.utils.connectors_config.build_postgres_config) +POSTGRES_HOST="localhost" +POSTGRES_PORT="5432" +POSTGRES_USER="sientia" +POSTGRES_PASSWORD="sientia" +POSTGRES_DBNAME="sientia" +POSTGRES_MIN_CONNECTIONS="5" +POSTGRES_MAX_CONNECTIONS="20" + diff --git a/ingestor/app.py b/ingestor/app.py index 36902f6..720d300 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -8,6 +8,8 @@ from time import time from prometheus_client import start_http_server import ingestor.metrics as metrics +from ingestor.config import GatewayConfig +from ingestor.gateway_runtime import GatewayRuntime from ingestor.ingestor import Ingestor exit_signal = Event() @@ -48,10 +50,27 @@ async def main(): # NOSONAR exit_signal.set() ingestor.logger.info('Ingestor prepared. Starting main loop.') + # Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor + # got far enough to have a data_manager (needed by CertProvider/DataSink) -- + # skipped entirely if prepare failed before that point. + runtime: GatewayRuntime | None = None + if ingestor.ingestor_manager is not None: + runtime = GatewayRuntime( + config=GatewayConfig.from_env(), + data_manager=ingestor.ingestor_manager.data_manager, + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + metrics_controller=ingestor.metrics_controller, + metadata=ingestor.metadata, + ) + await runtime.start() + while not exit_signal.is_set(): # NOSONAR start_time = time() # Start loop timer try: await ingestor.loop() + if runtime is not None: + await runtime.idle_sweep() metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc() # Increment loop counter # Use asyncio.sleep instead of exit_signal.wait for better async compatibility @@ -70,6 +89,8 @@ async def main(): # NOSONAR duration = time() - start_time metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration) + if runtime is not None: + await runtime.stop() await ingestor.shutdown() metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py new file mode 100644 index 0000000..22e14ec --- /dev/null +++ b/ingestor/gateway_runtime.py @@ -0,0 +1,149 @@ +"""GatewayRuntime (Task 18 F4 capstone): builds and owns the write/active-read/gRPC +object graph and wires it into the app's lifecycle. + +Reuses the existing ingestor's logger/notification_handler/metrics_controller and +its already-connected DataManager (so CertProvider shares the existing Mongo +connection instead of opening a second one -- MongoDBRepository.find_one has the +exact (collection_name, filters, metadata) signature CertProvider.fetch calls). + +# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974 +# integration; when swapping, build ONE SubscriptionEngine per OPC server (its +# non_receive_count is engine-wide -- see plan ledger F1 carry-forward). +""" + +from os import getenv + +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.notifications.models import NotificationLevel +from sientia_do.observability.logger import Logger +from sientia_do.observability.metrics_controller import MetricsController +from sientia_do.observability.sientia_monitoring import SientiaMonitoring +from sientia_do.repository.minio_repository import MinioRepository +from sientia_do.temporal.activities.postgres import Postgres +from sientia_do.utils.connectors_config import build_postgres_config + +from ingestor.adapters.grpc_server import GrpcServer +from ingestor.adapters.nats_consumer import NatsWriteConsumer +from ingestor.config import GatewayConfig, _env_bool +from ingestor.connection.certs import CertProvider +from ingestor.connection.pool import OpcConnectionPool +from ingestor.engines.active_read import ActiveReadEngine +from ingestor.engines.write import WriteEngine +from ingestor.io.data_sink import DataSink +from ingestor.managers.data_manager import DataManager + + +class GatewayRuntime(SientiaMonitoring): + """Owns the gateway write/active-read/gRPC object graph. Resilient to NATS + being unreachable at startup: the read path (existing subscription loop, + gRPC on-demand reads) must keep working even if the write path can't + connect yet -- so a NATS connect failure is logged/notified, not raised, + and retried from `idle_sweep()` on each main-loop iteration.""" + + def __init__( + self, + config: GatewayConfig, + data_manager: DataManager, + logger: Logger, + notification_handler: NotificationHandler, + metrics_controller: MetricsController, + metadata: dict | None = None, + ): + SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller) + self.config = config + self.metadata = metadata or {} + self._nats_started = False + + minio = MinioRepository( + endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'), + access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'), + secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'), + secure=_env_bool('MINIO_SECURE', True), + bucket=config.cert_bucket, + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + cert_provider = CertProvider( + mongo=data_manager.mongo_repository, + minio=minio, + cache_dir=config.cert_cache_dir, + bucket=config.cert_bucket, + metadata=self.metadata, + ) + self.pool = OpcConnectionPool(cert_provider, logger, notification_handler, metrics_controller) + + postgres = Postgres( + **build_postgres_config(), + logger=logger, + notification_handler=notification_handler, + metrics_controller=metrics_controller, + ) + self.sink = DataSink(data_manager, postgres) + + self.write_engine = WriteEngine( + self.pool, + self.sink, + logger, + notification_handler, + metrics_controller, + write_timestamp=config.write_timestamp, + write_quality=config.write_quality, + min_pause_s=config.write_min_pause_s, + ) + self.nats_consumer = NatsWriteConsumer( + config.nats_url, + self.write_engine, + logger, + subject=config.nats_write_subject, + queue=config.nats_queue, + ) + self.active_read = ActiveReadEngine( + self.pool, self.sink, logger, notification_handler, metrics_controller + ) + + # slot_lookup=None for now: address -> names resolution needs slot v2 + # (1974); until then ReadTag responses just aren't persisted. + self.grpc_server = ( + GrpcServer(self.active_read, logger, port=config.grpc_port, slot_lookup=None) + if config.grpc_enabled + else None + ) + + async def start(self) -> None: + await self._start_nats() + if self.grpc_server is not None: + await self.grpc_server.start() + self.logger.info( + f'GatewayRuntime started: nats_started={self._nats_started} ' + f'grpc_enabled={self.grpc_server is not None}' + ) + + async def stop(self) -> None: + if self.grpc_server is not None: + await self.grpc_server.stop() + if self._nats_started: + await self.nats_consumer.stop() + self._nats_started = False + await self.pool.close_all() + + async def idle_sweep(self) -> None: + await self.pool.release_idle(self.config.idle_conn_max_s) + if not self._nats_started: + await self._start_nats() + + async def _start_nats(self) -> None: + if self._nats_started: + return + try: + await self.nats_consumer.start() + self._nats_started = True + except Exception as e: + self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}') + await self.send_notification_async( + metadata=self.metadata, + notification_id='GATEWAY_NATS_CONSUMER_START_FAILED', + message=f'Failed to start NATS write consumer: {e}', + block='gateway_runtime', + level=NotificationLevel.ERROR, + ) diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 558cf1d..21454a9 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -43,6 +43,15 @@ def mock_app_env(monkeypatch): monkeypatch.setattr(app, 'Ingestor', mocks['Ingestor']) monkeypatch.setattr(asyncio, 'sleep', mocks['asyncio_sleep']) + mocks['GatewayConfig'] = MagicMock() + mocks['GatewayRuntime'] = MagicMock() + monkeypatch.setattr(app, 'GatewayConfig', mocks['GatewayConfig']) + monkeypatch.setattr(app, 'GatewayRuntime', mocks['GatewayRuntime']) + mock_runtime_instance = mocks['GatewayRuntime'].return_value + mock_runtime_instance.start = AsyncMock() + mock_runtime_instance.idle_sweep = AsyncMock() + mock_runtime_instance.stop = AsyncMock() + monkeypatch.setattr( app.metrics.APP_UP, 'labels', @@ -77,6 +86,7 @@ def mock_app_env(monkeypatch): mock_ingestor_instance = mocks['Ingestor'].return_value mock_ingestor_instance.poll_interval = 0.01 mock_ingestor_instance.logger = MagicMock() + mock_ingestor_instance.ingestor_manager = MagicMock() # Make async methods async mocks mock_ingestor_instance.prepare_ingestor = AsyncMock() @@ -124,6 +134,21 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys): mock_ingestor_instance.shutdown.assert_called_once() mock_ingestor_instance.logger.info.assert_any_call('Main loop exit_signaled.') + # GatewayRuntime (Task 18): built once prepare_ingestor succeeds, started before + # the loop, idle-swept each iteration, stopped on shutdown. + mock_app_env['GatewayRuntime'].assert_called_once_with( + config=mock_app_env['GatewayConfig'].from_env.return_value, + data_manager=mock_ingestor_instance.ingestor_manager.data_manager, + logger=mock_ingestor_instance.logger, + notification_handler=mock_ingestor_instance.notification_handler, + metrics_controller=mock_ingestor_instance.metrics_controller, + metadata=mock_ingestor_instance.metadata, + ) + mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value + mock_runtime_instance.start.assert_awaited_once() + mock_runtime_instance.idle_sweep.assert_awaited_once() + mock_runtime_instance.stop.assert_awaited_once() + captured = capsys.readouterr() assert 'Prometheus server started on port 9090.' in captured.out @@ -297,12 +322,16 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys): mock_exit_signal = mock_app_env['mock_exit_signal'] mock_ingestor_instance.prepare_ingestor.side_effect = Exception('Preparation failed') + mock_ingestor_instance.ingestor_manager = None # prepare failed before assignment mock_exit_signal.is_set.side_effect = [False, True] with pytest.raises(OsExitCalledError) as excinfo: await app.main() assert excinfo.value.code == 0 + # GatewayRuntime must not be built when prepare_ingestor never got a data_manager + mock_app_env['GatewayRuntime'].assert_not_called() + # Verify error metrics were incremented app.metrics.APP_ERRORS_TOTAL.labels.assert_called_with(pod_id='test_pod') mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once() diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py new file mode 100644 index 0000000..f4ad64b --- /dev/null +++ b/tests/unit/test_gateway_runtime.py @@ -0,0 +1,156 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from ingestor import gateway_runtime as gr +from ingestor.config import GatewayConfig + +_PATCHED_CLASSES = ( + 'MinioRepository', + 'CertProvider', + 'OpcConnectionPool', + 'Postgres', + 'build_postgres_config', + 'DataSink', + 'WriteEngine', + 'NatsWriteConsumer', + 'ActiveReadEngine', + 'GrpcServer', +) + + +@pytest.fixture +def mocks(monkeypatch): + patched = {} + for name in _PATCHED_CLASSES: + m = MagicMock() + monkeypatch.setattr(gr, name, m) + patched[name] = m + + patched['build_postgres_config'].return_value = { + 'host': 'localhost', 'port': 5432, 'user': 'u', 'password': 'p', + 'dbname': 'd', 'min_connections': 5, 'max_connections': 20, + } + patched['OpcConnectionPool'].return_value.close_all = AsyncMock() + patched['OpcConnectionPool'].return_value.release_idle = AsyncMock() + patched['NatsWriteConsumer'].return_value.start = AsyncMock() + patched['NatsWriteConsumer'].return_value.stop = AsyncMock() + patched['GrpcServer'].return_value.start = AsyncMock() + patched['GrpcServer'].return_value.stop = AsyncMock() + return patched + + +def make_runtime(**config_overrides): + return gr.GatewayRuntime( + config=GatewayConfig(**config_overrides), + data_manager=MagicMock(), + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=MagicMock(), + ) + + +def test_builds_object_graph_with_config_flags(mocks): + runtime = make_runtime( + write_timestamp=False, write_quality=False, write_min_pause_s=0.25, + cert_bucket='my-bucket', cert_cache_dir='certs-cache', grpc_enabled=False, + ) + + mocks['OpcConnectionPool'].assert_called_once() + mocks['DataSink'].assert_called_once() + + mocks['WriteEngine'].assert_called_once() + _, kwargs = mocks['WriteEngine'].call_args + assert kwargs['write_timestamp'] is False + assert kwargs['write_quality'] is False + assert kwargs['min_pause_s'] == 0.25 + + _, cert_kwargs = mocks['CertProvider'].call_args + assert cert_kwargs['bucket'] == 'my-bucket' + assert cert_kwargs['cache_dir'] == 'certs-cache' + + mocks['GrpcServer'].assert_not_called() + assert runtime.grpc_server is None + + +def test_builds_grpc_server_when_enabled(mocks): + runtime = make_runtime(grpc_enabled=True, grpc_port=12345) + + mocks['GrpcServer'].assert_called_once() + assert runtime.grpc_server is mocks['GrpcServer'].return_value + + +@pytest.mark.asyncio +async def test_start_starts_nats_consumer(mocks): + runtime = make_runtime(grpc_enabled=False) + + await runtime.start() + + mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once() + mocks['GrpcServer'].return_value.start.assert_not_called() + assert runtime._nats_started is True + + +@pytest.mark.asyncio +async def test_start_also_starts_grpc_when_enabled(mocks): + runtime = make_runtime(grpc_enabled=True) + + await runtime.start() + + mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once() + mocks['GrpcServer'].return_value.start.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks): + runtime = make_runtime() + runtime.send_notification_async = AsyncMock() + nats_mock = mocks['NatsWriteConsumer'].return_value + nats_mock.start = AsyncMock(side_effect=[Exception('boom'), None]) + + await runtime.start() # must not raise + + assert runtime._nats_started is False + runtime.logger.error.assert_called_once() + runtime.send_notification_async.assert_awaited_once() + + await runtime.idle_sweep() + + assert runtime._nats_started is True + assert nats_mock.start.await_count == 2 + + +@pytest.mark.asyncio +async def test_stop_stops_started_components_and_closes_pool(mocks): + runtime = make_runtime(grpc_enabled=True) + await runtime.start() + + await runtime.stop() + + mocks['GrpcServer'].return_value.stop.assert_awaited_once() + mocks['NatsWriteConsumer'].return_value.stop.assert_awaited_once() + mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stop_skips_nats_stop_if_never_started(mocks): + runtime = make_runtime() + runtime.send_notification_async = AsyncMock() + nats_mock = mocks['NatsWriteConsumer'].return_value + nats_mock.start = AsyncMock(side_effect=Exception('boom')) + + await runtime.start() # nats fails to start + await runtime.stop() + + nats_mock.stop.assert_not_called() + mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_idle_sweep_releases_idle_connections(mocks): + runtime = make_runtime(idle_conn_max_s=42.0) + runtime._nats_started = True # isolate from the retry path + + await runtime.idle_sweep() + + mocks['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0)