chore(gateway): final sweep - README/functional Kafka cleanup, asyncua pin, 1879 checklist verification (Task 21)

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  │
+                       │ - Redis Coord.   │    │ - MongoDB Store │
-                       │ - Slot Leasing   │    │ - MongoDB Store │
+                       │ - Slot Leasing   │    │ - Data Pipeline │
-                       │ - Heartbeats     │    │ - 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**
+3. **Performance Issues**
-   - Verify Kafka cluster is running
-   - Check broker addresses and network connectivity
-   - Review topic configuration and permissions
-
-4. **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
 

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.

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()

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."