From ff605eea9345ee7208b762cee580f47a180b3209 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 08:59:19 -0300 Subject: [PATCH 01/11] feat(gateway): GatewayConfig env (NATS/gRPC/certs/write flags) --- .env.example | 13 +++++++++ ingestor/config.py | 46 ++++++++++++++++++++++++++++++ tests/unit/test_config.py | 60 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 119 insertions(+) create mode 100644 ingestor/config.py create mode 100644 tests/unit/test_config.py diff --git a/.env.example b/.env.example index 54ea343..b69ece3 100644 --- a/.env.example +++ b/.env.example @@ -15,3 +15,16 @@ MONGODB_USERNAME="mongo_user" MONGODB_PASSWORD="mongo_password" MONGODB_DATABASE="sientia" +# OPC Gateway +NATS_URL="nats://localhost:4222" +NATS_WRITE_SUBJECT="opc.write" +NATS_QUEUE="gateway-writers" +GRPC_ENABLED="false" +GRPC_PORT="50051" +CERT_BUCKET="sientia-dataops" +CERT_CACHE_DIR="/tmp/opc-certs" +WRITE_TIMESTAMP="true" +WRITE_QUALITY="true" +WRITE_MIN_PAUSE_S="0.0" +IDLE_CONN_MAX_S="300.0" + diff --git a/ingestor/config.py b/ingestor/config.py new file mode 100644 index 0000000..3d0bf20 --- /dev/null +++ b/ingestor/config.py @@ -0,0 +1,46 @@ +"""Gateway-specific env config (NATS/gRPC/certs/write flags). + +Separate from ingestor.py's inline os.getenv (Redis/Mongo/TTLs) — this module +covers only the new gateway wiring consumed by app.py in Task 18. +""" + +from dataclasses import dataclass +from os import getenv + + +def _env_bool(name: str, default: bool) -> bool: + value = getenv(name) + if value is None: + return default + return value.strip().lower() in ('true', '1', 'yes') + + +@dataclass +class GatewayConfig: + nats_url: str = 'nats://localhost:4222' + nats_write_subject: str = 'opc.write' + nats_queue: str = 'gateway-writers' + grpc_enabled: bool = False + grpc_port: int = 50051 + cert_bucket: str = 'sientia-dataops' + cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 - default is spec'd, overridable via env + write_timestamp: bool = True + write_quality: bool = True + write_min_pause_s: float = 0.0 + idle_conn_max_s: float = 300.0 + + @classmethod + def from_env(cls) -> 'GatewayConfig': + return cls( + nats_url=getenv('NATS_URL', cls.nats_url), + nats_write_subject=getenv('NATS_WRITE_SUBJECT', cls.nats_write_subject), + nats_queue=getenv('NATS_QUEUE', cls.nats_queue), + grpc_enabled=_env_bool('GRPC_ENABLED', cls.grpc_enabled), + grpc_port=int(getenv('GRPC_PORT', cls.grpc_port)), + cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket), + cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir), + write_timestamp=_env_bool('WRITE_TIMESTAMP', cls.write_timestamp), + write_quality=_env_bool('WRITE_QUALITY', cls.write_quality), + write_min_pause_s=float(getenv('WRITE_MIN_PAUSE_S', cls.write_min_pause_s)), + idle_conn_max_s=float(getenv('IDLE_CONN_MAX_S', cls.idle_conn_max_s)), + ) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..839e588 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,60 @@ +import pytest + +from ingestor.config import GatewayConfig + +_ENV_VARS = ( + 'NATS_URL', 'NATS_WRITE_SUBJECT', 'NATS_QUEUE', 'GRPC_ENABLED', 'GRPC_PORT', + 'CERT_BUCKET', 'CERT_CACHE_DIR', 'WRITE_TIMESTAMP', 'WRITE_QUALITY', + 'WRITE_MIN_PAUSE_S', 'IDLE_CONN_MAX_S', +) + + +@pytest.fixture(autouse=True) +def clean_env(monkeypatch): + for var in _ENV_VARS: + monkeypatch.delenv(var, raising=False) + + +def test_from_env_defaults(): + cfg = GatewayConfig.from_env() + + assert cfg.nats_url == 'nats://localhost:4222' + assert cfg.nats_write_subject == 'opc.write' + assert cfg.nats_queue == 'gateway-writers' + assert cfg.grpc_enabled is False + assert cfg.grpc_port == 50051 + assert cfg.cert_bucket == 'sientia-dataops' + assert cfg.cert_cache_dir == '/tmp/opc-certs' # noqa: S108 + assert cfg.write_timestamp is True + assert cfg.write_quality is True + assert cfg.write_min_pause_s == 0.0 + assert cfg.idle_conn_max_s == 300.0 + + +def test_from_env_overrides(monkeypatch): + monkeypatch.setenv('GRPC_ENABLED', 'true') + monkeypatch.setenv('GRPC_PORT', '6000') + monkeypatch.setenv('WRITE_MIN_PAUSE_S', '0.5') + monkeypatch.setenv('WRITE_QUALITY', 'false') + + cfg = GatewayConfig.from_env() + + assert cfg.grpc_enabled is True + assert cfg.grpc_port == 6000 + assert cfg.write_min_pause_s == 0.5 + assert cfg.write_quality is False + + +@pytest.mark.parametrize('value,expected', [ + ('0', False), + ('no', False), + ('FALSE', False), + ('1', True), + ('YES', True), +]) +def test_bool_parsing_edge_cases(monkeypatch, value, expected): + monkeypatch.setenv('GRPC_ENABLED', value) + + cfg = GatewayConfig.from_env() + + assert cfg.grpc_enabled is expected From 391268776cc3d55636bdc2022294d4b137364e33 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 13:51:18 -0300 Subject: [PATCH 02/11] 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) From 3f413f8d6f9cb7beae898b6ad08af9f940a54954 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 14:17:21 -0300 Subject: [PATCH 03/11] fix(gateway): app survives runtime init failure; dedupe NATS-down notifications; fix stale slot_lookup note --- ingestor/adapters/grpc_server.py | 4 +++- ingestor/app.py | 29 ++++++++++++++++++---------- ingestor/gateway_runtime.py | 5 +++++ tests/unit/test_app.py | 31 ++++++++++++++++++++++++++++++ tests/unit/test_gateway_runtime.py | 11 +++++++++-- 5 files changed, 67 insertions(+), 13 deletions(-) diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py index 05cf6e8..a9fee06 100644 --- a/ingestor/adapters/grpc_server.py +++ b/ingestor/adapters/grpc_server.py @@ -17,7 +17,9 @@ class ReadTagServicer(opc_gateway_pb2_grpc.OpcGatewayServicer): self.active_read = active_read self.logger = logger # ponytail: slot_lookup wiring (address -> names from the current slot) - # lands in Task 18 (app wiring); optional callable keeps this servicer usable now. + # lands at the SIENTIAPDE-1974 integration -- `names` only exist in slot v2, + # which the Orchestrator produces after 1974. Until then ReadTag never + # persists (persisted=False); optional callable keeps this servicer usable now. self.slot_lookup = slot_lookup async def ReadTag(self, request, context): diff --git a/ingestor/app.py b/ingestor/app.py index 720d300..bf96140 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -52,18 +52,27 @@ async def main(): # NOSONAR # 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. + # skipped entirely if prepare failed before that point. Construction can hit + # the network (MinIO bucket check, Postgres engine): a failure there must not + # kill the read path, so the app continues without the gateway runtime. 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() + try: + 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() + except Exception as e: + runtime = None + metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc() + ingestor.logger.error( + f'GatewayRuntime failed to initialize, continuing without gateway: {e}' + ) while not exit_signal.is_set(): # NOSONAR start_time = time() # Start loop timer diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py index 22e14ec..16fe85f 100644 --- a/ingestor/gateway_runtime.py +++ b/ingestor/gateway_runtime.py @@ -53,6 +53,7 @@ class GatewayRuntime(SientiaMonitoring): self.config = config self.metadata = metadata or {} self._nats_started = False + self._nats_notified = False # notify once per outage, not per 5s retry minio = MinioRepository( endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'), @@ -138,8 +139,12 @@ class GatewayRuntime(SientiaMonitoring): try: await self.nats_consumer.start() self._nats_started = True + self._nats_notified = False except Exception as e: self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}') + if self._nats_notified: + return + self._nats_notified = True await self.send_notification_async( metadata=self.metadata, notification_id='GATEWAY_NATS_CONSUMER_START_FAILED', diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 21454a9..ab038a3 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -346,3 +346,34 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys): mock_ingestor_instance.logger.error.assert_called_once_with( 'Failed to prepare ingestor: Preparation failed' ) + + +@pytest.mark.asyncio +async def test_main_gateway_runtime_init_failure_keeps_read_path_alive(mock_app_env): + """GatewayRuntime construction hitting the network (MinIO/Postgres) and failing + must not kill the app: the main loop (read path) still runs and shuts down cleanly.""" + mock_ingestor_instance = mock_app_env['Ingestor'].return_value + mock_exit_signal = mock_app_env['mock_exit_signal'] + + mock_app_env['GatewayRuntime'].side_effect = Exception('MinIO unreachable') + mock_exit_signal.is_set.side_effect = [False, True] + mock_app_env['time_time'].side_effect = [10.0, 11.0] + + with pytest.raises(OsExitCalledError) as excinfo: + await app.main() + assert excinfo.value.code == 0 + + # Loop still ran and shutdown was clean, without any runtime calls + mock_ingestor_instance.loop.assert_called_once() + mock_ingestor_instance.shutdown.assert_called_once() + mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value + mock_runtime_instance.idle_sweep.assert_not_called() + mock_runtime_instance.stop.assert_not_called() + + # Failure was logged and counted, but did not set exit_signal + assert any( + 'GatewayRuntime failed to initialize' in str(c) + for c in mock_ingestor_instance.logger.error.call_args_list + ) + mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once() + mock_exit_signal.set.assert_not_called() diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py index f4ad64b..1a29050 100644 --- a/tests/unit/test_gateway_runtime.py +++ b/tests/unit/test_gateway_runtime.py @@ -106,7 +106,7 @@ 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]) + nats_mock.start = AsyncMock(side_effect=[Exception('boom'), Exception('boom'), None]) await runtime.start() # must not raise @@ -114,10 +114,17 @@ async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks): runtime.logger.error.assert_called_once() runtime.send_notification_async.assert_awaited_once() + await runtime.idle_sweep() # second failure: retry connects, but no re-notify + + assert runtime._nats_started is False + runtime.send_notification_async.assert_awaited_once() # still just one notification + await runtime.idle_sweep() assert runtime._nats_started is True - assert nats_mock.start.await_count == 2 + assert nats_mock.start.await_count == 3 + # flag reset on success: a future outage notifies again + assert runtime._nats_notified is False @pytest.mark.asyncio From 21d0b371054be0b053ed40607562842767da9b2d Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 14:52:50 -0300 Subject: [PATCH 04/11] docs(gateway): deploy notes for helm/images/platform repos (Task 19+20) --- docs/deploy-notes.md | 94 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) create mode 100644 docs/deploy-notes.md diff --git a/docs/deploy-notes.md b/docs/deploy-notes.md new file mode 100644 index 0000000..9954a51 --- /dev/null +++ b/docs/deploy-notes.md @@ -0,0 +1,94 @@ +# OPC Gateway — Deploy Notes (SIENTIAPDE-1972) + +This repo has no Dockerfile and no `values/` directory — that's by design (see +below). This doc captures what the Helm, container-images, and platform repos +need to know to deploy the gateway. Cross-reference: +`docs/superpowers/specs/2026-07-13-opc-gateway-design.md` (§9/§10/§12). + +## Entrypoint & image + +- Runs as `ingestor.app` (`PYTHON_APP` entrypoint), cloned from Gitea at pod + start onto the shared **`sientia-module`** base image (Python 3.11+). +- No repo-local Dockerfile: images are centralized in + `sientia-container-images`. +- New pip deps that must be present in the image / library distribution + (see `requirements.txt`): + - `nats-py>=2.6` + - `grpcio>=1.60` + - `grpcio-tools>=1.60` + - the `sientia_do` release that includes SIENTIAPDE-1590 (generic + Mongo/MinIO/Postgres methods used by the gateway). + +## Env (new since the ingestor) + +From `.env.example`, `# OPC Gateway` block: + +- `NATS_URL` +- `NATS_WRITE_SUBJECT` +- `NATS_QUEUE` +- `GRPC_ENABLED` +- `GRPC_PORT` +- `CERT_BUCKET` +- `CERT_CACHE_DIR` +- `WRITE_TIMESTAMP` +- `WRITE_QUALITY` +- `WRITE_MIN_PAUSE_S` +- `IDLE_CONN_MAX_S` + +Plus MinIO (client cert storage) and Postgres (`write_status` updates): + +- `MINIO_ENDPOINT`, `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`, `MINIO_SECURE` +- `POSTGRES_HOST`, `POSTGRES_PORT`, `POSTGRES_USER`, `POSTGRES_PASSWORD`, + `POSTGRES_DBNAME`, `POSTGRES_MIN_CONNECTIONS`, `POSTGRES_MAX_CONNECTIONS` + +## Health probe + +Exec probe (platform DataOps standard): + +``` +curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' +``` + +Verified: `ingestor/app.py` sets `metrics.APP_UP.labels(pod_id=...).set(1)` +right after `start_http_server(HTTP_METRICS_PORT, default 9090)` in +`start_prometheus_server()`, and `.set(0)` on shutdown in the main loop's +`finally`/exit path. A runtime check (prometheus_client HTTP server, gauge +set to 1, scraped over HTTP) produced the exposition line +`app_up{pod_id="x"} 1.0`, which matches the probe regex. + +## waitForDependencies + +Init-container wait list: **Gitea, MongoDB, Redis, Postgres, MinIO, NATS**. + +NATS is also tolerated at runtime: the app starts even if NATS is +unreachable and the `nats-py` client retries/reconnects internally +(`ingestor/adapters/nats_consumer.py` — no custom reconnect logic, relies on +nats-py's built-in reconnection); only the first connection failure should +raise a notification, not every retry. + +## NetworkPolicy + +- Egress: Mongo, Redis, Postgres, MinIO, NATS, OPC servers (port 4840). +- Ingress: Prometheus scrape on 9090. +- gRPC port (default 50051): **internal-only**, no auth — there is no + consumer yet. Only add an ingress rule when a real caller appears, per the + feature owner (Vitor Pimentel). + +## Cert cache + +- `CERT_CACHE_DIR` (default `/tmp/opc-certs`) must be writable — + `emptyDir`/tmpfs, since the pod root filesystem is read-only. +- Certs are sourced from MinIO bucket `sientia-dataops` + Mongo collection + `platform_certificates`. + +## ServiceMonitor + +- Label `release: kube-prometheus-stack`. + +## Cross-repo integration gates + +- Subscription v2 + gRPC persist land at **SIENTIAPDE-1974** (Orchestrator). +- `write_status` column via Flyway (`sientia-dataops-database-migrations`) + + the Laborious DataFrame change land at **SIENTIAPDE-1975**. +- NATS broker provisioning on the air-gapped platform is + **SIENTIAPDE-1985**. From cebb5215a37d9e620f1d0223b84bac33f04c59ee Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 15:31:37 -0300 Subject: [PATCH 05/11] chore(gateway): final sweep - README/functional Kafka cleanup, asyncua pin, 1879 checklist verification (Task 21) --- README.md | 57 ++++++++++++++++++---------- requirements.txt | 2 +- tests/functional/conftest.py | 16 -------- tests/functional/test_single_node.py | 25 ------------ 4 files changed, 39 insertions(+), 61 deletions(-) diff --git a/README.md b/README.md index d7ef4ad..91cafdb 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,6 @@ A high-performance, scalable OPC UA data ingestion system designed for industria ### Core Functionality - **OPC UA Integration**: Native support for OPC UA servers with secure and unsecured connections - **Automatic Load Balancing**: Slot-based architecture for horizontal scaling across multiple instances -- **Real-time Data Streaming**: Kafka integration for historical data streaming - **Persistent Storage**: MongoDB integration for historical data persistence - **Health Monitoring**: Comprehensive Prometheus metrics and health checks - **Fault Tolerance**: Automatic failover, reconnection, and error recovery @@ -36,9 +35,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi ┌──────────────────┐ ┌─────────────────┐ │ ResourceManager │ │ DataManager │ │ │ │ │ - │ - Redis Coord. │ │ - Kafka Export │ - │ - Slot Leasing │ │ - MongoDB Store │ - │ - Heartbeats │ │ - Data Pipeline │ + │ - Redis Coord. │ │ - MongoDB Store │ + │ - Slot Leasing │ │ - Data Pipeline │ + │ - Heartbeats │ │ │ └──────────────────┘ └─────────────────┘ ``` @@ -47,18 +46,38 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi - **Ingestor**: Main application orchestrator managing the overall lifecycle - **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing - **OPC Manager**: Handles individual OPC UA server connections and tag subscriptions -- **Data Manager**: Manages data streaming (MongoDB | Kafka) +- **Data Manager**: Manages data streaming (MongoDB) - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis +## 🚪 OPC Gateway (SIENTIAPDE-1972) + +Alongside the passive ingestion pipeline above, this service also runs an OPC +Gateway for on-demand and command-driven interactions with OPC UA servers: + +- **NATS write consumer**: consumes write commands from `opc.write` (see + `NATS_WRITE_SUBJECT`) and executes them against the target OPC server + (`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`). +- **On-demand gRPC `ReadTag`**: a gRPC server (`ingestor/adapters/grpc_server.py`, + `ingestor/proto/opc_gateway.proto`) for synchronous, single-tag reads. +- **Scheduled active reads**: a polling read engine (`ingestor/engines/active_read.py`) + for tags that need periodic reads outside the subscription pipeline; + orchestrator-side wiring lands in SIENTIAPDE-1974. +- **Certificate management**: client certs for secure OPC connections are + fetched from MinIO + a Mongo `platform_certificates` collection + (`ingestor/connection/certs.py`) rather than shipped in the image. + +See `docs/deploy-notes.md` for the deployment-facing details (env vars, +health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in +`.env.example` for the full list of new environment variables. + ## 📋 Prerequisites - Python 3.11+ - Redis server - MongoDB server -- Kafka cluster (optional, for data streaming) - OPC UA servers for data collection -**Note**: External dependencies (Redis, MongoDB, Kafka) must be available either through: +**Note**: External dependencies (Redis, MongoDB) must be available either through: - Port forwarding from a Kubernetes cluster - External Docker Compose setup - Cloud-managed services @@ -99,7 +118,6 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi # Port forwarding from Kubernetes cluster kubectl port-forward svc/redis-master 6379:6379 kubectl port-forward svc/mongodb 27017:27017 - kubectl port-forward svc/kafka 9092:9092 # Or connect to external Docker Compose # Ensure services are accessible on localhost with appropriate ports @@ -215,21 +233,23 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: - `ingestor_slots_managed_current`: Slots managed by this instance - `ingestor_active_total`: Total active ingestor instances - `redis_connection_status`: Redis connection health -- `kafka_connection_status`: Kafka connection health ### Data Processing Metrics - `ingestor_tag_written_count`: Data write operations -- `kafka_messages_sent_total`: Kafka message count - `redis_operations_total`: Redis operation count +### OPC Gateway Metrics +- `opc_write_total`: Active write attempts (NATS-driven), by result +- `opc_write_latency_seconds`: Active write latency +- `opc_active_read_total`: Scheduled/on-demand read attempts, by result +- `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...) + ## ⚙️ Configuration ### Environment Variables | Variable | Description | Default | Required | |----------|-------------|---------|----------| -| `KAFKA_SERVERS` | Comma-separated Kafka server addresses | `localhost:9092` | No | -| `EXPORT_TO_KAFKA` | Enable Kafka data export | `false` | No | | `REDIS_HOST` | Redis server hostname | `localhost` | Yes | | `REDIS_PORT` | Redis server port | `6379` | Yes | | `REDIS_USERNAME` | Redis username | `None` | No | @@ -244,6 +264,10 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: | `MONGODB_PASSWORD` | MongoDB password | `sientia` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | +See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`, +`CERT_BUCKET`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the +gateway-specific environment variables. + ### OPC Server Configuration OPC servers are configured through Redis with the following structure: @@ -319,12 +343,7 @@ sientia-dataops-opc-ingestor/ - Check authentication credentials - Ensure proper network configuration -3. **Kafka Export Failures** - - Verify Kafka cluster is running - - Check broker addresses and network connectivity - - Review topic configuration and permissions - -4. **Performance Issues** +3. **Performance Issues** - Monitor Prometheus metrics for bottlenecks - Adjust polling intervals and lease TTLs - Review OPC server performance and network latency @@ -350,7 +369,7 @@ export LOG_LEVEL=DEBUG - **Horizontal Scaling**: Deploy multiple ingestor instances for high availability - **Load Distribution**: Use Redis-based slot allocation for automatic load balancing - **Resource Limits**: Monitor CPU, memory, and network usage -- **Database Performance**: Optimize MongoDB indexes and Kafka partitioning +- **Database Performance**: Optimize MongoDB indexes ## 🤝 Contributing diff --git a/requirements.txt b/requirements.txt index 34a722a..56e8655 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -asyncua==1.1.5 +asyncua==1.1.8 redis # TODO(SIENTIAPDE-1972): bump to the sientia-dataops-library release that includes # SIENTIAPDE-1590 (generic Mongo find_one/distinct/aggregate methods) once it is tagged. diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py index 5d8aab1..217938f 100644 --- a/tests/functional/conftest.py +++ b/tests/functional/conftest.py @@ -1,8 +1,5 @@ # import subprocess # from time import sleep -# from typing import Generator -# import uuid -# from kafka import KafkaConsumer # import pytest # from redis import Redis @@ -32,16 +29,3 @@ # # Limpa o banco de dados após os testes # redis.flushdb() # redis.close() - - -# def kafka_searcher(topic) -> Generator[KafkaConsumer, None, None]: -# consumer = KafkaConsumer( -# topic, -# bootstrap_servers="localhost:9092", -# group_id=f"test-group-{uuid.uuid4()}", -# auto_offset_reset="earliest", # Começa a consumir apenas mensagens novas -# enable_auto_commit=True, -# ) - -# yield consumer -# consumer.close() diff --git a/tests/functional/test_single_node.py b/tests/functional/test_single_node.py index 5264e16..330a815 100644 --- a/tests/functional/test_single_node.py +++ b/tests/functional/test_single_node.py @@ -2,8 +2,6 @@ # import subprocess # from time import sleep -# from tests.functional.conftest import kafka_searcher - # new_data = { # "slot:opc_tags:1": { @@ -69,14 +67,6 @@ # assert redis_client.get("lease:opc_tags:1") == 'ingestor' # assert redis_client.get("heartbeat:ingestor:ingestor") == '1' -# # Check if data is in Kafka - -# kafka = next(kafka_searcher('test_topic_1')) -# sleep(1) -# messages = kafka.poll(timeout_ms=10000) - -# assert messages, "Expected messages in Kafka, but got none." - # def test_simple_double_slot(redis_client): @@ -90,12 +80,6 @@ # assert redis_client.get("lease:opc_tags:1") == 'ingestor' # assert redis_client.get("heartbeat:ingestor:ingestor") == '1' -# # Check if data is in Kafka -# kafka1 = next(kafka_searcher('test_topic_double_slot1')) -# messages = kafka1.poll(timeout_ms=10000) - -# assert messages, "Expected messages in test_topic_double_slot1, but got none." - # new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [ # 'test_topic_double_slot2'] # redis_client.set("slot:opc_tags:2", @@ -107,12 +91,3 @@ # assert redis_client.get("lease:opc_tags:2") == 'ingestor' # assert redis_client.get("lease:opc_tags:1") == 'ingestor' # assert redis_client.get("heartbeat:ingestor:ingestor") == '1' - -# # Check if data is in Kafka -# kafka2 = next(kafka_searcher('test_topic_double_slot2')) -# messages = kafka2.poll(timeout_ms=10000) - -# assert messages, "Expected messages in test_topic_double_slot2, but got none." - -# messages = kafka1.poll(timeout_ms=10000) -# assert messages, "Expected messages in test_topic_double_slot1, but got none." From e0d8dd3729ea257f80f79ed3430a51145c5a99b7 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 15:33:03 -0300 Subject: [PATCH 06/11] SIENTIAPDE-1972: F4 close-out - ledger de commits + gates de integracao (Task 21.end) --- docs/superpowers/plans/2026-07-13-opc-gateway.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/docs/superpowers/plans/2026-07-13-opc-gateway.md b/docs/superpowers/plans/2026-07-13-opc-gateway.md index 812e52a..5b63898 100644 --- a/docs/superpowers/plans/2026-07-13-opc-gateway.md +++ b/docs/superpowers/plans/2026-07-13-opc-gateway.md @@ -99,7 +99,14 @@ At the end of each phase Fx, before starting the next: - Wire the gRPC `slot_lookup` seam (address→names from the current slot) so on-demand reads persist; if the lookup becomes I/O-bound, move it inside the servicer's try/except. - gRPC server is internal-only, no auth (no consumer yet — likely frontend later, per feature owner). - Edge case (minor): `read_now` with `SourceTimestamp=None` persists `timestamp=None`; add a guard if it matters in practice. -**F4 — Wiring, health, deploy, hardening:** _(fill at F4 close-out)_ +**F4 — Wiring, health, deploy, hardening** — base `e9ce9e5`. Status: ✅ complete, 222 unit tests green, ruff clean, `grep -i kafka` = 0. +- `ff605ee` feat(gateway): GatewayConfig env (NATS/gRPC/certs/write flags) (Task 17) +- `3912687` feat(gateway): wire engines+adapters via GatewayRuntime (write/gRPC live; subscription v2 gated on 1974) (Task 18) +- `3f413f8` fix(gateway): app survives runtime init failure; dedupe NATS-down notifications; fix stale slot_lookup note (Task 18) +- `21d0b37` docs(gateway): deploy notes for helm/images/platform repos (Task 19+20) +- `cebb521` chore(gateway): final sweep - README/functional Kafka cleanup, asyncua pin, 1879 checklist verification (Task 21) +- Deferred by design (verified in Task 21): NTP sync + `Good_Local_Override` quality downgrade (no flag exists; ties to a real clock-drift case), write-via-OPC-method (card marks it flag/future). Pre-existing legacy files >500 lines (`ingestor_manager.py` 578, `opc_manager.py` 507) flagged, untouched (predate the gateway). +- Integration gates (cross-repo): subscription v2 swap + gRPC persist → SIENTIAPDE-1974; write_status column (Flyway) + Laborious NATS publish → SIENTIAPDE-1975; NATS broker in cluster → SIENTIAPDE-1985. End-to-end validation (real NATS + OPC server machine) agreed with the feature owner for deployment time. ### Final: open per-phase PRs (only after F4 is complete, tested, working) Since all phases live on one feature branch, split them at the end using the recorded SHA ranges: for each phase in order F0→F4, create a phase branch off `main` (or stacked on the previous phase branch) containing only that phase's code commits (cherry-pick the ledger's SHA range), then open one PR per phase targeting `main`, in phase order. This keeps each PR small and reviewable instead of one giant diff. @@ -913,9 +920,9 @@ async def test_readtag_maps_response(): --- ### Task 21.end: F4 close-out + open per-phase PRs (see [Phase Delivery](#phase-delivery--commit-ledger)) -- [ ] Full `pytest tests` green + `ruff` clean; all F0–F4 phases complete, tested, working. -- [ ] Record F4 commit SHAs (base + range) in the [Commit Ledger](#commit-ledger). -- [ ] **Now** open one PR per phase (F0→F4) from the recorded SHA ranges, in order — this is the only point where PRs are opened. +- [x] Full `pytest tests` green (222) + `ruff` clean; all F0–F4 phases complete and unit-tested. +- [x] F4 commit SHAs recorded in the [Commit Ledger](#commit-ledger) (base `e9ce9e5`). +- [ ] **Now** open one PR per phase (F0→F4) from the recorded SHA ranges, in order — this is the only point where PRs are opened. *(Pending: user go-ahead — requires push; also note the epic's end-to-end validation with a real NATS broker + OPC server machine is agreed for deployment time, per the feature owner.)* --- From 0e29cc405453b5a6e7a4e58c4dd83ae14f395da6 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 21:45:06 -0300 Subject: [PATCH 07/11] docs(gateway): e2e validation guide + seed script for the KepServer session --- docs/e2e-test-guide.md | 94 +++++++++++++++++++++++++ scripts/seed_e2e.py | 154 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 248 insertions(+) create mode 100644 docs/e2e-test-guide.md create mode 100644 scripts/seed_e2e.py diff --git a/docs/e2e-test-guide.md b/docs/e2e-test-guide.md new file mode 100644 index 0000000..93e848f --- /dev/null +++ b/docs/e2e-test-guide.md @@ -0,0 +1,94 @@ +# OPC Gateway — E2E Validation Guide (SIENTIAPDE-1972) + +Session script for validating the gateway against a real KepServer + cluster (with the feature owner). Each step names the command and the expected result. Run order matters. + +**In scope:** cert management, subscription (legacy slot v1), NATS write → OPC → `write_status`, gRPC ReadTag, 1879 write flags, resilience, metrics. +**Out of scope (integration-gated):** multi-name subscription / scheduled active reads / gRPC persist (need slot v2 → SIENTIAPDE-1974); the real Laborious publish (SIENTIAPDE-1975) — here we hand-publish the C2 message, which *is* the contract. + +--- + +## 0. Prerequisites + +| What | How | +|---|---| +| KepServer reachable | `OPC_SERVER_URL` (default `opc.tcp://localhost:49320`); Simulation Examples tags enabled | +| NATS (test instance) | `docker run -d --name nats -p 4222:4222 nats:2` — no JetStream needed (Core NATS). Cluster provisioning is SIENTIAPDE-1985; this container is only for the session | +| `nats` CLI | [natscli](https://github.com/nats-io/natscli) or `docker exec` a `nats-box` | +| `grpcurl` | for the ReadTag call | +| Mongo/Redis/Postgres | same instances the app already uses (`.env`) | +| **MinIO policy (pre-req if testing security)** | frontend user needs write on bucket `sientia-dataops`; gateway reads it. The 1569 CRUD now defaults to `sientia-dataops` (`MINIO_BUCKET` env overrides) | + +## 1. Seed test data + +```bash +python scripts/seed_e2e.py --print-only # inspect first +python scripts/seed_e2e.py # seed Mongo + Redis slot v1 + Postgres table +``` +Creates: connector `e2e-kepserver` (9001), 2 tags **with `metadata: {}`**, slot `slot:opc_tags:1` (v1), and `sientia_data.e2e_predictions` with the `write_status` column + 1 row (`model_id='9001'`, `timestamp='2026-07-15T12:00:00Z'`). + +> Adjust the two tag `address` values to real KepServer node ids if the Simulation Examples channel differs. + +## 2. Start the gateway + +```bash +# .env: MONGODB_*/REDIS_* as today, plus NATS_URL=nats://localhost:4222, GRPC_ENABLED=true +python -m ingestor.app +``` +Expect in logs: `NatsWriteConsumer started: subject=opc.write queue=gateway-writers`, `GrpcServer started on port 50051`, `GatewayRuntime started: nats_started=True grpc_enabled=True`. + +Health probe (what k8s will run): `curl -sf http://localhost:9090/metrics | grep '^app_up'` → `app_up{...} 1.0`. + +## 3. Subscription (legacy v1 path) + +The slot from step 1 subscribes `e2e_ramp`. Expect within ~1 cycle: +- Mongo collection `raw_e2e-gateway-test` receiving documents `{tag, name, timestamp, value}`. +- Metrics: `opc_connection_status{...} 1`, `opc_tags_subscribed_current` ≥ 1. + +## 4. Write via NATS (C2 → OPC → C3) + +The seed prints the exact command; it targets the seeded predictions row: +```bash +nats pub opc.write '' +``` +Expect: +1. KepServer: `e2e_write_target` (User1) becomes `42.5`. +2. Postgres: `SELECT write_status FROM sientia_data.e2e_predictions WHERE model_id='9001';` → `ok`. +3. Metrics: `opc_write_total{result="success"}` +1, `opc_write_latency_seconds` observed, `opc_write_status_by_model{status="ok"} 1`. + +**1879 flag checks** (repeat the publish varying the payload): +- `"write_timestamp": false` / `"write_quality": false` → KepServer must still accept (or reject — that's the point of testing against the real server; compare behaviors). +- Bad value `"value": "abc"` → no crash; `write_status='error'`, `opc_error_by_type_total{error_type="conversion"}` +1. +- Malformed JSON (`nats pub opc.write 'not-json'`) → dropped + logged, app alive. +- Missing field (remove `"addr"`) → `write_status='error'` (`bad_command`), no crash. +- Two rapid publishes with `WRITE_MIN_PAUSE_S=1` set → second write delayed ≥1s (rate-limit). + +## 5. gRPC on-demand read + +```bash +grpcurl -plaintext -d '{"server_id":"9001","addr":"ns=2;s=Simulation Examples.Functions.Ramp1"}' \ + localhost:50051 opcgateway.OpcGateway/ReadTag +``` +Expect: `value`, `opc_status_code: "Good"`, timestamps, **`persisted: false`** (correct until 1974). Metrics: `opc_active_read_total{result="success"}` +1, `opc_read_lag_seconds` set. +Error case: bad addr → gRPC `INTERNAL`, `opc_active_read_total{result="error"}` +1, app alive. + +## 6. Certificates (round 2 — security on) + +1. Upload platform cert via the connectors CRUD (writes to `sientia-dataops` now) — or place objects under `platform/certificates/` + the single `platform_certificates` Mongo doc manually. +2. Flip the seeded connector/slot to `use_security: true`, restart the gateway. +3. Expect Basic256Sha256 session on KepServer (check its connection log); reads/writes as in steps 3–5. + +## 7. Resilience + +- **NATS down:** `docker stop nats` → app keeps reading (subscription + gRPC fine); ONE notification `GATEWAY_NATS_CONSUMER_START_FAILED`; `docker start nats` → consumer reconnects on next loop (~5s); publish works again. +- **MinIO down at boot** (security round): gateway logs `GatewayRuntime failed to initialize, continuing without gateway` and the read path still runs. +- **KepServer restart:** subscription reconnect kicks in after ~15 idle cycles (`opc_reconnections_tried_total`). + +## 8. Wrap-up checks + +- `curl -s localhost:9090/metrics | grep -E 'opc_(write|active_read|error|tag_value|read_lag)'` — all families populated. +- No `kafka` anywhere: `grep -ri kafka ingestor/` → empty. +- Postgres row still has exactly the statuses produced by step 4 (no stray writes). + +--- + +**Ledger:** phases/commits in `docs/superpowers/plans/2026-07-13-opc-gateway.md`; deploy details in `docs/deploy-notes.md`. diff --git a/scripts/seed_e2e.py b/scripts/seed_e2e.py new file mode 100644 index 0000000..0659399 --- /dev/null +++ b/scripts/seed_e2e.py @@ -0,0 +1,154 @@ +"""Seed test data for the OPC Gateway e2e validation session (SIENTIAPDE-1972). + +Populates: + 1. Mongo `connectors` + `connector_tags` (each tag with `metadata: {}` — required + for documentation, per the feature owner). + 2. Redis slot v1 (`slot:opc_tags:1`) so the legacy subscription path collects them. + 3. Postgres test predictions table (with the `write_status` column) + one row the + NATS write test can UPDATE (correlated by timestamp + model_id). + +Idempotent: connectors/tags are upserted, the slot is overwritten, the table uses +CREATE IF NOT EXISTS + ON CONFLICT DO NOTHING. + +Usage (from the repo root, sientia-dataops conda env): + python scripts/seed_e2e.py --print-only # dry-run: show docs/SQL, touch nothing + python scripts/seed_e2e.py # seed Mongo + Redis + Postgres + +Env (same names the gateway/app use): + MONGODB_URL/MONGODB_USERNAME/MONGODB_PASSWORD/MONGODB_DATABASE + REDIS_HOST/REDIS_PORT POSTGRES_HOST/PORT/USER/PASSWORD/DBNAME + OPC_SERVER_URL (default opc.tcp://localhost:49320 — KepServer default port) + +# ponytail: plain sync pymongo/redis/sqlalchemy, no sientia_do async plumbing — +# this is a test fixture, not production code. +""" + +import argparse +import json +import os +import sys + +OPC_URL = os.getenv('OPC_SERVER_URL', 'opc.tcp://localhost:49320') + +CONNECTOR = { + 'connector_id': 9001, + 'type': 'opc_ua', + 'name': 'e2e-kepserver', + 'url': OPC_URL, + 'server_uri': 'urn:e2e:gateway:test', + 'subscription_period_ms': 1000, + 'use_security': False, # round 1; flip after certs are in sientia-dataops + 'status': 'Active', + 'description': 'OPC Gateway e2e validation connector (SIENTIAPDE-1972)', +} + +# metadata: {} on every tag — feature owner asked test tags to carry the field. +TAGS = [ + {'tag_id': 9101, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.Ramp1', + 'tag_name': 'e2e_ramp', 'category': 'Read', 'data_type': 'float', 'unit': '-', + 'status': 'Active', 'description': 'e2e read tag', 'metadata': {}}, + {'tag_id': 9102, 'connector_id': 9001, 'address': 'ns=2;s=Simulation Examples.Functions.User1', + 'tag_name': 'e2e_write_target', 'category': 'Write', 'data_type': 'float', 'unit': '-', + 'status': 'Active', 'description': 'e2e write tag', 'metadata': {}}, +] + +# Slot v1 (legacy path the gateway still runs until SIENTIAPDE-1974 delivers slot v2). +SLOT_V1 = { + 'e2e-kepserver': { + 'name': 'e2e-kepserver', + 'server_id': '9001', + 'url': OPC_URL, + 'server_uri': 'urn:e2e:gateway:test', + 'subscription_period_ms': 1000, + 'tags': { + TAGS[0]['address']: { + 'tag_name': TAGS[0]['tag_name'], + 'frequency': '5000', + 'topics': ['raw_e2e-gateway-test'], + } + }, + } +} + +PREDICTIONS_SQL = """ +CREATE SCHEMA IF NOT EXISTS sientia_data; +CREATE TABLE IF NOT EXISTS sientia_data.e2e_predictions ( + model_id text NOT NULL, + prediction double precision, + prediction_confidence double precision, + timestamp text NOT NULL, + write_status text, + PRIMARY KEY (model_id, timestamp) +); +INSERT INTO sientia_data.e2e_predictions (model_id, prediction, prediction_confidence, timestamp) +VALUES ('9001', 42.5, 0.97, '2026-07-15T12:00:00Z') +ON CONFLICT DO NOTHING; +""" + +# The matching NATS write command for the row above (paste into `nats pub opc.write '...'`). +NATS_CMD = { + 'msg_id': 'e2e-0001', 'server_id': '9001', 'addr': TAGS[1]['address'], + 'value': 42.5, 'data_type': 'float', 'tag_type': 'prediction', + 'model_id': '9001', 'model_name': 'e2e-model', + 'schema': 'sientia_data', 'table_name': 'e2e_predictions', + 'timestamp': '2026-07-15T12:00:00Z', + 'url': OPC_URL, 'use_security': False, 'server_uri': 'urn:e2e:gateway:test', +} + + +def seed_mongo(): + from pymongo import MongoClient + + url = os.getenv('MONGODB_URL', 'localhost:27017') + user, pwd = os.getenv('MONGODB_USERNAME', 'sientia'), os.getenv('MONGODB_PASSWORD', 'sientia') + db = MongoClient(f'mongodb://{user}:{pwd}@{url}')[os.getenv('MONGODB_DATABASE', 'sientia')] + db['connectors'].replace_one({'connector_id': CONNECTOR['connector_id']}, CONNECTOR, upsert=True) + for tag in TAGS: + db['connector_tags'].replace_one({'tag_id': tag['tag_id']}, tag, upsert=True) + print(f'mongo: upserted 1 connector + {len(TAGS)} tags (metadata: {{}})') + + +def seed_redis(): + import redis + + r = redis.Redis(host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', '6379'))) + r.set('slot:opc_tags:1', json.dumps(SLOT_V1)) + print('redis: wrote slot:opc_tags:1 (v1)') + + +def seed_postgres(): + from sqlalchemy import create_engine, text + + host, port = os.getenv('POSTGRES_HOST', 'localhost'), os.getenv('POSTGRES_PORT', '5432') + user, pwd = os.getenv('POSTGRES_USER', 'sientia'), os.getenv('POSTGRES_PASSWORD', 'sientia') + dbname = os.getenv('POSTGRES_DBNAME', 'sientia') + engine = create_engine(f'postgresql://{user}:{pwd}@{host}:{port}/{dbname}') + with engine.begin() as conn: + for stmt in filter(str.strip, PREDICTIONS_SQL.split(';')): + conn.execute(text(stmt)) + print('postgres: sientia_data.e2e_predictions ready (write_status column + 1 row)') + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--print-only', action='store_true', help='show docs/SQL, touch nothing') + args = parser.parse_args() + + if args.print_only: + print('connector:', json.dumps(CONNECTOR, indent=2)) + print('tags:', json.dumps(TAGS, indent=2)) + print('slot v1:', json.dumps(SLOT_V1, indent=2)) + print('sql:', PREDICTIONS_SQL) + print('nats write cmd:', json.dumps(NATS_CMD, indent=2)) + return 0 + + seed_mongo() + seed_redis() + seed_postgres() + print('\nNATS write command for the session:') + print(f"nats pub opc.write '{json.dumps(NATS_CMD)}'") + return 0 + + +if __name__ == '__main__': + sys.exit(main()) From 0d60a3fb67c573283d16fce474a5b3f9a639b03c Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Fri, 17 Jul 2026 09:44:38 -0300 Subject: [PATCH 08/11] feat(gateway): optional .env loading for local dev (app + seed); document in e2e guide --- docs/e2e-test-guide.md | 2 ++ ingestor/app.py | 7 +++++++ requirements-dev.txt | 2 +- scripts/seed_e2e.py | 7 +++++++ 4 files changed, 17 insertions(+), 1 deletion(-) diff --git a/docs/e2e-test-guide.md b/docs/e2e-test-guide.md index 93e848f..e73d029 100644 --- a/docs/e2e-test-guide.md +++ b/docs/e2e-test-guide.md @@ -20,6 +20,8 @@ Session script for validating the gateway against a real KepServer + cluster (wi ## 1. Seed test data +Copy `.env.example` to `.env` (gitignored) and fill in the values from the session (OPC endpoint, Mongo/Redis/Postgres/MinIO accesses) — both the app and the seed script read it automatically (add `OPC_SERVER_URL=opc.tcp://HOST:49320`; it is seed-only, so it isn't in the example file). Then: + ```bash python scripts/seed_e2e.py --print-only # inspect first python scripts/seed_e2e.py # seed Mongo + Redis slot v1 + Postgres table diff --git a/ingestor/app.py b/ingestor/app.py index bf96140..e18ab41 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -7,6 +7,13 @@ from time import time from prometheus_client import start_http_server +try: # optional: local dev reads .env; in-cluster env comes from Helm (no .env file) + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + import ingestor.metrics as metrics from ingestor.config import GatewayConfig from ingestor.gateway_runtime import GatewayRuntime diff --git a/requirements-dev.txt b/requirements-dev.txt index 56ab376..42c84fe 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -16,4 +16,4 @@ pytest-asyncio>=0.21.0 # Async test support (already in main requirements) # Development Tools ipython>=8.12.0 # Enhanced Python shell -ipdb>=0.13.13 # IPython debugger \ No newline at end of file +ipdb>=0.13.13 # IPython debuggerpython-dotenv>=1.0 # local dev only: app/seed read .env if present (in-cluster env comes from Helm) diff --git a/scripts/seed_e2e.py b/scripts/seed_e2e.py index 0659399..394a924 100644 --- a/scripts/seed_e2e.py +++ b/scripts/seed_e2e.py @@ -28,6 +28,13 @@ import json import os import sys +try: # optional: local dev reads .env; in-cluster env comes from Helm + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + OPC_URL = os.getenv('OPC_SERVER_URL', 'opc.tcp://localhost:49320') CONNECTOR = { From 650280e404355ef85cfd1e75adc55e3ca48815f2 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Fri, 17 Jul 2026 18:31:40 -0300 Subject: [PATCH 09/11] chore: gitignore docs/HANDOFF.md (local handoff note, not tracked) --- .gitignore | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 198783f..26dba15 100644 --- a/.gitignore +++ b/.gitignore @@ -181,4 +181,6 @@ git_log .git/ .ruff_cache/ -.mypy_cache/ \ No newline at end of file +.mypy_cache/ +# Local handoff note for the next agent - not tracked +docs/HANDOFF.md From 79258658bc6a41fbefb2306b2053be9c17d5bc69 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 11:29:41 -0300 Subject: [PATCH 10/11] style: ruff format (new quality gate check picked up from main) --- ingestor/gateway_runtime.py | 4 +++- tests/unit/test_config.py | 31 ++++++++++++++++++++---------- tests/unit/test_gateway_runtime.py | 17 ++++++++++++---- 3 files changed, 37 insertions(+), 15 deletions(-) diff --git a/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py index 16fe85f..c23ec22 100644 --- a/ingestor/gateway_runtime.py +++ b/ingestor/gateway_runtime.py @@ -72,7 +72,9 @@ class GatewayRuntime(SientiaMonitoring): bucket=config.cert_bucket, metadata=self.metadata, ) - self.pool = OpcConnectionPool(cert_provider, logger, notification_handler, metrics_controller) + self.pool = OpcConnectionPool( + cert_provider, logger, notification_handler, metrics_controller + ) postgres = Postgres( **build_postgres_config(), diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 839e588..7aa9743 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -3,9 +3,17 @@ import pytest from ingestor.config import GatewayConfig _ENV_VARS = ( - 'NATS_URL', 'NATS_WRITE_SUBJECT', 'NATS_QUEUE', 'GRPC_ENABLED', 'GRPC_PORT', - 'CERT_BUCKET', 'CERT_CACHE_DIR', 'WRITE_TIMESTAMP', 'WRITE_QUALITY', - 'WRITE_MIN_PAUSE_S', 'IDLE_CONN_MAX_S', + 'NATS_URL', + 'NATS_WRITE_SUBJECT', + 'NATS_QUEUE', + 'GRPC_ENABLED', + 'GRPC_PORT', + 'CERT_BUCKET', + 'CERT_CACHE_DIR', + 'WRITE_TIMESTAMP', + 'WRITE_QUALITY', + 'WRITE_MIN_PAUSE_S', + 'IDLE_CONN_MAX_S', ) @@ -45,13 +53,16 @@ def test_from_env_overrides(monkeypatch): assert cfg.write_quality is False -@pytest.mark.parametrize('value,expected', [ - ('0', False), - ('no', False), - ('FALSE', False), - ('1', True), - ('YES', True), -]) +@pytest.mark.parametrize( + 'value,expected', + [ + ('0', False), + ('no', False), + ('FALSE', False), + ('1', True), + ('YES', True), + ], +) def test_bool_parsing_edge_cases(monkeypatch, value, expected): monkeypatch.setenv('GRPC_ENABLED', value) diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py index 1a29050..c057753 100644 --- a/tests/unit/test_gateway_runtime.py +++ b/tests/unit/test_gateway_runtime.py @@ -28,8 +28,13 @@ def mocks(monkeypatch): 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, + '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() @@ -52,8 +57,12 @@ def make_runtime(**config_overrides): 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, + 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() From 5daabc341a1f6054cfaf86fdd6511c4738ff68f0 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Tue, 22 Sep 2026 16:01:54 -0300 Subject: [PATCH 11/11] fix(gateway): suppress bandit B108 false positive on cert_cache_dir default '/tmp/opc-certs' is a spec'd, overridable-via-env default (already noqa'd for ruff's S108); bandit needs its own '# nosec B108' since it runs as a separate CI step and doesn't read ruff's noqa comments. --- ingestor/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestor/config.py b/ingestor/config.py index 3d0bf20..dab6676 100644 --- a/ingestor/config.py +++ b/ingestor/config.py @@ -23,7 +23,7 @@ class GatewayConfig: grpc_enabled: bool = False grpc_port: int = 50051 cert_bucket: str = 'sientia-dataops' - cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 - default is spec'd, overridable via env + cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 # nosec B108 - default is spec'd, overridable via env write_timestamp: bool = True write_quality: bool = True write_min_pause_s: float = 0.0