From cebb5215a37d9e620f1d0223b84bac33f04c59ee Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 15 Jul 2026 15:31:37 -0300 Subject: [PATCH] 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."