diff --git a/.env.example b/.env.example index 54ea343..27003d2 100644 --- a/.env.example +++ b/.env.example @@ -15,3 +15,32 @@ 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" + +# 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/.gitignore b/.gitignore index 3f24065..5f3a8bd 100644 --- a/.gitignore +++ b/.gitignore @@ -187,4 +187,4 @@ git_log docs/superpowers/ docs/e2e-test-guide.md docs/deploy-notes.md -docs/HANDOFF.md \ No newline at end of file +docs/HANDOFF.md diff --git a/README.md b/README.md index 0f2db31..a240e71 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,27 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi - **Data Manager**: Manages data persistence (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+ @@ -253,6 +274,12 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: - `ingestor_tag_written_count`: Data write operations - `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 @@ -273,6 +300,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. `cert_path` / `private_key_path` are only required for secured connections (see [Certificate Generation](#-certificate-generation)): diff --git a/ingestor/adapters/grpc_server.py b/ingestor/adapters/grpc_server.py index ca2d5c2..d096964 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 36902f6..e18ab41 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -7,7 +7,16 @@ 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 from ingestor.ingestor import Ingestor exit_signal = Event() @@ -48,10 +57,36 @@ 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. 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: + 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 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 +105,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/config.py b/ingestor/config.py new file mode 100644 index 0000000..dab6676 --- /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 # nosec B108 - 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/ingestor/gateway_runtime.py b/ingestor/gateway_runtime.py new file mode 100644 index 0000000..c23ec22 --- /dev/null +++ b/ingestor/gateway_runtime.py @@ -0,0 +1,156 @@ +"""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 + self._nats_notified = False # notify once per outage, not per 5s retry + + 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 + 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', + message=f'Failed to start NATS write consumer: {e}', + block='gateway_runtime', + level=NotificationLevel.ERROR, + ) diff --git a/requirements-dev.txt b/requirements-dev.txt index cbd2882..09381f5 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -17,6 +17,7 @@ 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 +python-dotenv>=1.0 # local dev only: app/seed read .env if present (in-cluster env comes from Helm) # gRPC code generation (regenerate ingestor/proto/*_pb2*.py) — not needed # at runtime: the generated stubs are committed. Kept out of requirements.txt diff --git a/scripts/seed_e2e.py b/scripts/seed_e2e.py new file mode 100644 index 0000000..394a924 --- /dev/null +++ b/scripts/seed_e2e.py @@ -0,0 +1,161 @@ +"""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 + +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 = { + '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()) 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." diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index 558cf1d..ab038a3 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() @@ -317,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_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..7aa9743 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,71 @@ +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 diff --git a/tests/unit/test_gateway_runtime.py b/tests/unit/test_gateway_runtime.py new file mode 100644 index 0000000..c057753 --- /dev/null +++ b/tests/unit/test_gateway_runtime.py @@ -0,0 +1,172 @@ +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'), 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() # 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 == 3 + # flag reset on success: a future outage notifies again + assert runtime._nats_notified is False + + +@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)