diff --git a/.env.example b/.env.example index d6fcde0..54ea343 100644 --- a/.env.example +++ b/.env.example @@ -1,7 +1,3 @@ -# Kafka -KAFKA_SERVERS="localhost:9092" -EXPORT_TO_KAFKA="false" - # Redis REDIS_HOST="localhost" REDIS_PORT="6379" diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index a8f9b2a..61b07c5 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -4,13 +4,16 @@ on: pull_request: branches: - main + - 'release/**' + - 'feature/**' types: [ opened, synchronize, reopened ] jobs: quality-gate: - uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main + uses: Aignosi/github_workflow_templates/.github/workflows/python-quality-gate.yml@main permissions: write-all with: project_name: 'ingestor' repositories: 'sientia-dataops-library' + requirements_file: 'requirements.txt' secrets: inherit \ No newline at end of file diff --git a/README.md b/README.md index d7ef4ad..0f2db31 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,7 +46,7 @@ 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 persistence (MongoDB) - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis ## 📋 Prerequisites @@ -55,10 +54,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi - 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,12 +97,15 @@ 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 ``` +6. **Generate an OPC UA client certificate** (only if connecting to a secured OPC UA server) + + See [Certificate Generation](#-certificate-generation) below. + ## Usage ### Running the Ingestor @@ -155,7 +156,7 @@ pytest # Run with coverage pytest --cov=ingestor --cov-report=html - +``` ### Populating Redis with OPC Configuration @@ -171,6 +172,38 @@ pytest --cov=ingestor --cov-report=html The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage. +## 🔐 Certificate Generation + +OPC UA servers that require secure connections (e.g. KEPServer with `Basic256` policy) validate the ingestor via an X.509 client certificate. Use `scripts/generate-and-push-opc-cert.sh` to generate that certificate and push it to the Gitea repo the pod clones on boot. + +```bash +GITEA_PASSWORD='' ./scripts/generate-and-push-opc-cert.sh +``` + +Environment variables (all optional except `GITEA_PASSWORD`): + +| Variable | Description | Default | +|----------|-------------|---------| +| `GITEA_URL` | Gitea in-cluster URL | `https://git.sientia.ai` | +| `GITEA_USER` | Gitea user | `gitea_admin` | +| `GITEA_PASSWORD` | Gitea password (**required**) | — | +| `GITEA_REPO` | Repo the ingestor pod clones | `gitea_admin/sientia-dataops-opc-ingestor` | +| `BRANCH` | Branch to push the cert to | `main` | +| `APP_URI` | SAN URI of the cert — **must match** the `uri` field of the corresponding document in the `OPC_servers` Mongo collection, since the ingestor uses it as `application_uri` (KEPServer rejects with `BadCertificateUriInvalid` on mismatch) | `urn:sientia:opc-ingestor` | +| `CN` | Certificate common name | `sientia-opc-ingestor` | +| `DAYS` | Certificate validity (days) | `3650` | +| `FORCE` | Set to `1` to overwrite an existing cert already pushed to the repo | unset | + +What the script does: +1. Generates a self-signed RSA-2048 X.509 cert + key with `openssl`, with the SAN URI, `keyUsage`, and `extendedKeyUsage` extensions OPC UA / KEPServer require. +2. Clones the ingestor's Gitea repo, copies the cert/key (`.pem`) and a `.der` copy (for manual import) into `certs/`, commits, and pushes. +3. Prints the paths the pod will see after cloning (`/app/certs/opc-ingestor-cert.pem`, `/app/certs/opc-ingestor-key.pem`) and the next manual steps. + +After pushing, you still need to: +- Set `cert_path` / `private_key_path` (and `uri`) on the server's document in the `OPC_servers` collection so `OpcManager` picks them up (see [OPC Server Configuration](#opc-server-configuration)). +- Restart the ingestor deployment so it re-clones the repo: `kubectl -n sientia rollout restart deployment/opc-ingestor`. +- On first connection, the server rejects the new cert — trust it in the OPC UA server's console (e.g. KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*) and restart its runtime. + ## 🧪 Testing ### Unit Tests @@ -215,11 +248,9 @@ 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 ## ⚙️ Configuration @@ -228,8 +259,6 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: | 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 | @@ -246,7 +275,7 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: ### OPC Server Configuration -OPC servers are configured through Redis with the following structure: +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)): ```json { @@ -256,6 +285,8 @@ OPC servers are configured through Redis with the following structure: "server_id": "1", "url": "opc.tcp://localhost:4841", "server_uri": "http://opcua-server.simulator", + "cert_path": "/app/certs/opc-ingestor-cert.pem", + "private_key_path": "/app/certs/opc-ingestor-key.pem", "tags": { "ns=2;i=2": { "aggr_func": "avg", @@ -292,8 +323,8 @@ sientia-dataops-opc-ingestor/ │ ├── ingestor.py # Core ingestor logic │ └── metrics.py # Prometheus metrics definitions ├── simulator/ # OPC simulation and testing tools +├── scripts/ # Ops scripts (e.g. OPC UA cert generation) ├── tests/ # Test suite -├── docker-compose.yaml # Infrastructure services └── requirements.txt # Python dependencies ``` @@ -319,12 +350,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 +376,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 @@ -368,10 +394,6 @@ export LOG_LEVEL=DEBUG - Use type hints where appropriate - Follow the established architectural patterns -## 📄 License - -This project is licensed under the terms specified in the LICENSE file. - ## 🆘 Support For support and questions: @@ -383,14 +405,3 @@ For support and questions: --- **Note**: This OPC Ingestor is designed for production use in industrial environments. Ensure proper security configuration and network isolation for production deployments. - -This comprehensive documentation provides: - -1. **Complete feature overview** with architectural details -2. **Detailed installation and setup instructions** -3. **Comprehensive configuration documentation** -4. **Performance tuning and troubleshooting guides** -5. **Development guidelines and contribution standards** -6. **Updated docstrings** for all major classes and methods - -The documentation now serves as a complete reference for users, developers, and operators of the OPC Ingestor system. \ No newline at end of file diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 4b76888..2244d61 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -19,7 +19,7 @@ class Ingestor(SientiaMonitoring): - Managing slot leases for load balancing across multiple instances - Connecting to and monitoring OPC UA servers - Subscribing to OPC tags and collecting real-time data - - Distributing data to Kafka and MongoDB + - Distributing data to MongoDB - Providing health monitoring and metrics collection The ingestor uses a slot-based architecture where each slot represents @@ -27,8 +27,6 @@ class Ingestor(SientiaMonitoring): This allows for horizontal scaling and load distribution. Environment Variables: - KAFKA_SERVERS: Comma-separated list of Kafka server addresses (default: "localhost:9092") - EXPORT_TO_KAFKA: Enable/disable Kafka export (default: "false") REDIS_HOST: Redis server hostname (default: "localhost") REDIS_PORT: Redis server port (default: 6379) REDIS_USERNAME: Redis username (optional) @@ -43,7 +41,6 @@ class Ingestor(SientiaMonitoring): MONGODB_DATABASE: MongoDB database name (default: "sientia") Attributes: - export_to_kafka (bool): Whether to export data to Kafka redis_host (str): Redis server hostname redis_port (int): Redis server port redis_username (str): Redis username (optional) @@ -54,7 +51,6 @@ class Ingestor(SientiaMonitoring): poll_interval (int): Interval in seconds for polling operations mongo_database (str): MongoDB database name mongo_connection_string (str): Complete MongoDB connection string - kafka_servers (list): List of Kafka server addresses logger: Logger instance for application logging notification_handler: Handler for sending notifications metadata (dict): Application metadata for notifications and tracking @@ -66,17 +62,12 @@ class Ingestor(SientiaMonitoring): Initializes the ingestor with configuration values retrieved from environment variables. Sets up all necessary connections and configurations for: - - Kafka connectivity (if enabled) - Redis for slot management and coordination - MongoDB for data persistence and notifications - OPC UA server management - Metrics collection and monitoring """ - kafka_servers = getenv('KAFKA_SERVERS', 'localhost:9092') - export_to_kafka: bool = getenv('EXPORT_TO_KAFKA', 'false') == 'true' - - self.export_to_kafka = export_to_kafka self.redis_host = getenv('REDIS_HOST', 'localhost') self.redis_port = int(getenv('REDIS_PORT', '6379')) self.redis_username = getenv('REDIS_USERNAME', None) @@ -91,7 +82,6 @@ class Ingestor(SientiaMonitoring): self.mongo_database = getenv('MONGODB_DATABASE', 'sientia') self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}' - self.kafka_servers = kafka_servers.split(',') self.logger = get_logger(__name__) self.notification_handler = NotificationHandler( connection_string=self.mongo_connection_string, @@ -156,7 +146,7 @@ class Ingestor(SientiaMonitoring): else: # Subscribe to acquired slots if self.ingestor_manager: - self.ingestor_manager.update_opc_servers() + await self.ingestor_manager.update_opc_servers() await self.ingestor_manager.subscribe_to_tags(acquired) async def prepare_ingestor(self): @@ -179,7 +169,6 @@ class Ingestor(SientiaMonitoring): """ self.ingestor_manager = IngestorManager( - kafka_servers=','.join(self.kafka_servers), redis_data={ 'host': self.redis_host, 'port': self.redis_port, @@ -195,7 +184,6 @@ class Ingestor(SientiaMonitoring): logger=self.logger, notification_handler=self.notification_handler, metrics_controller=self.metrics_controller, - export_to_kafka=self.export_to_kafka, ) assert self.ingestor_manager is not None diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 44d248e..ef3b5ac 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -1,10 +1,6 @@ -import json import os import traceback -from time import sleep -from kafka import KafkaProducer -from kafka.errors import NoBrokersAvailable from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -18,31 +14,22 @@ import ingestor.metrics as metrics class DataManager(SientiaMonitoring): """ - Manages data persistence and export operations for the OPC Ingestor. + Manages data persistence operations for the OPC Ingestor. The DataManager is responsible for: - Storing OPC data in MongoDB for historical analysis and persistence - - Exporting data to Kafka for real-time streaming and downstream processing - Managing database connections and ensuring data integrity - Providing data access interfaces for other components - The manager supports both MongoDB and Kafka operations, with Kafka export - being optional and configurable. It implements retry logic for connection - failures and provides comprehensive error handling and notification. - Args: - kafka_servers (str): Comma-separated string of Kafka server addresses mongo_connection_string (str): MongoDB connection string mongo_database (str): MongoDB database name - export_to_kafka (bool): Whether to enable Kafka export functionality metadata (dict): Application metadata for notifications and tracking logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for sending notifications Attributes: pod_id (str): Pod identifier for metrics labeling - kafka_producer (KafkaProducer): Kafka producer instance for data export - export_to_kafka (bool): Whether Kafka export is enabled connection_string (str): MongoDB connection string database (str): MongoDB database name mongo_client (MongoClient): MongoDB client instance @@ -51,47 +38,25 @@ class DataManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, mongo_connection_string: str, mongo_database: str, - export_to_kafka: bool, metadata: dict, logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, ) -> None: """ - Initializes the DataManager instance with Kafka and MongoDB connections. - - This constructor attempts to establish connections to the specified services: - 1. Kafka: Initializes producer with retry logic (up to 3 attempts) - 2. MongoDB: Establishes connection and verifies server availability - - The initialization process includes: - - Kafka producer setup with JSON serialization - - MongoDB client initialization and connection testing - - Metrics recording for connection status - - Error handling with notifications + Initializes the DataManager instance with a MongoDB connection. Args: - kafka_servers (str): Comma-separated string of Kafka server addresses mongo_connection_string (str): MongoDB connection string mongo_database (str): MongoDB database name - export_to_kafka (bool): Whether to enable Kafka export metadata (dict): Application metadata logger (Logger): Logger instance notification_handler (NotificationHandler): Notification handler - - Raises: - NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts. - - Metrics: - - KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure """ self.pod_id = os.getenv('HOSTNAME', 'localhost') - self.kafka_producer = None - self.export_to_kafka = export_to_kafka SientiaMonitoring.__init__( self, @@ -100,37 +65,6 @@ class DataManager(SientiaMonitoring): notification_handler=notification_handler, ) - if self.export_to_kafka: - for i in range(0, 3): - logger.info( - f'Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}' - ) - try: - self.kafka_producer = KafkaProducer( - bootstrap_servers=kafka_servers, - value_serializer=lambda v: json.dumps(v).encode( - 'utf-8' - ), # Serialize JSON messages - key_serializer=lambda k: str(k).encode('utf-8') if k else None, - ) - # Kafka connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1) - break - except NoBrokersAvailable: - logger.error(f'Kafka servers {kafka_servers} are not available. Retrying...') - sleep(5) - else: - # Kafka not connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - logger.error( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - raise NoBrokersAvailable( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - - logger.info(f'DataManager initialized with Kafka servers: {kafka_servers}') - logger.info( f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}' ) @@ -153,26 +87,7 @@ class DataManager(SientiaMonitoring): def shutdown(self): """ Gracefully shuts down the DataManager and closes all connections. - - This method ensures proper cleanup of: - - Kafka producer connection with message flushing - - MongoDB client connection - - Metrics recording for connection status - - The method handles connection closure gracefully, logging any errors - that occur during shutdown while ensuring all resources are properly released. """ - if self.kafka_producer: - try: - self.kafka_producer.flush(timeout=10) - self.kafka_producer.close() - # Mark as disconnected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - except Exception as e: - self.logger.error(f'Error closing Kafka producer: {e}') - else: - self.logger.warning('Kafka producer is already closed or not initialized.') - try: self.mongo_repository.close() except Exception as e: @@ -181,83 +96,18 @@ class DataManager(SientiaMonitoring): def __del__(self): self.shutdown() - def delivery_report(self, msg): - """ - Callback for successful Kafka message delivery reports. - - This method is called by the Kafka producer when a message is successfully - delivered to a topic. It logs the delivery details including topic, partition, - and offset information for debugging and monitoring purposes. - - Args: - msg: Kafka message object containing delivery details - """ - self.logger.debug( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - def delivery_error(self, err): - """ - Callback for Kafka message delivery error reports. - - This method is called by the Kafka producer when a message delivery fails. - It logs the error details for debugging and monitoring purposes. - - Args: - err: Error information from the failed delivery attempt - """ - self.logger.error(f'Delivery failed for record : {err}') - async def publish(self, topic: str, data: dict) -> None: """ - Publishes a message to a specified Kafka topic. + Persists a message to MongoDB. Args: - topic (str): The name of the Kafka topic to which the message will be published. - data (dict): The message data to be sent to the Kafka topic. + topic (str): The name of the MongoDB collection to which the message will be written. + data (dict): The message data to be stored. Returns: None - - Raises: - Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback. """ - if self.export_to_kafka and self.kafka_producer: - try: - self.logger.debug(f'Publishing message to topic {topic}: {data}') - self.kafka_producer.send(topic=topic, value=data).add_callback( - self.delivery_report - ).add_errback(self.delivery_error) - - self.kafka_producer.flush(timeout=10) - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_SENT, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - - except Exception as e: - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_ERRORS, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - trace = traceback.format_exc() - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: {e}', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=trace, - ) - self.logger.error(trace) - try: await self.mongo_repository.insert( collection_name=topic, diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 030e305..34a09ca 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -19,7 +19,7 @@ class IngestorManager(SientiaMonitoring): Central coordinator for managing OPC data ingestion operations. The IngestorManager orchestrates the interaction between different components: - - DataManager: Handles data persistence and Kafka export + - DataManager: Handles data persistence - OPC Managers: Manage individual OPC UA server connections - ResourceManager: Coordinates slot leasing and load balancing @@ -36,7 +36,6 @@ class IngestorManager(SientiaMonitoring): - Load balancing across multiple ingestor instances Args: - kafka_servers (str): Comma-separated list of Kafka server addresses redis_data (dict): Redis connection parameters (host, port, username, password) lease_ttl (int): Time-to-live for slot leases in seconds heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds @@ -46,10 +45,9 @@ class IngestorManager(SientiaMonitoring): metadata (dict): Application metadata for notifications and tracking logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for sending notifications - export_to_kafka (bool): Whether to export data to Kafka Attributes: - data_manager (DataManager): Manages data persistence and Kafka export + data_manager (DataManager): Manages data persistence opc_managers (dict): Dictionary of OPC managers keyed by server name resource_manager (ResourceManager): Manages Redis-based resource coordination number_of_slots (int): Total number of slots configured in the system @@ -61,7 +59,6 @@ class IngestorManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, @@ -72,7 +69,6 @@ class IngestorManager(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, - export_to_kafka: bool = False, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -87,10 +83,8 @@ class IngestorManager(SientiaMonitoring): ) self.data_manager = DataManager( - kafka_servers=kafka_servers, mongo_connection_string=mongo_connection_string, mongo_database=mongo_database, - export_to_kafka=export_to_kafka, metadata=metadata, logger=logger, notification_handler=notification_handler, diff --git a/ingestor/metrics.py b/ingestor/metrics.py index 503d299..cb18652 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -19,7 +19,6 @@ from prometheus_client import Counter, Gauge, Histogram # Metric label definitions for consistent labeling across all metrics POD_ID_LABEL = ['pod_id'] SERVER_LABELS = ['pod_id', 'server_name', 'server_url'] -KAFKA_LABELS = ['pod_id', 'topic'] REDIS_LABELS = ['pod_id', 'operation'] NOTIFICATION_LABELS = ['pod_id', 'level', 'block'] @@ -134,22 +133,6 @@ OPC_RECONNECTIONS_TOTAL = Counter( ['pod_id', 'server_name'], ) -# --- Data Manager (Kafka) Metrics --- -KAFKA_MESSAGES_SENT = Counter( - 'kafka_messages_sent_total', 'Total messages sent to Kafka', KAFKA_LABELS -) -KAFKA_MESSAGES_ERRORS = Counter( - 'kafka_messages_errors_total', - 'Total errors sending messages to Kafka', - KAFKA_LABELS, -) -KAFKA_CONNECTION_STATUS = Gauge( - 'kafka_connection_status', - 'Connection status with Kafka (1=connected, 0=disconnected)', - POD_ID_LABEL, -) - - # --- Notification Metrics --- NOTIFICATIONS_SENT = Counter( 'notifications_sent_total', diff --git a/requirements-local.txt b/requirements-local.txt new file mode 100644 index 0000000..190e40b --- /dev/null +++ b/requirements-local.txt @@ -0,0 +1,5 @@ +asyncua==1.1.5 +redis +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 +prometheus_client +pymongo \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index bb176bf..ed65ae3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +sientia_do==1.12.1 prometheus_client pymongo \ No newline at end of file diff --git a/scripts/generate-and-push-opc-cert.sh b/scripts/generate-and-push-opc-cert.sh new file mode 100755 index 0000000..713965e --- /dev/null +++ b/scripts/generate-and-push-opc-cert.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# Gera o certificado X.509 de cliente OPC UA do ingestor e sobe pro Gitea +# in-cluster (repo que o pod clona no boot — nada muda no values/deployment, +# só rollout restart depois do push). +# +# Uso: +# GITEA_PASSWORD='' ./scripts/generate-and-push-opc-cert.sh +# +# Vars (defaults pro cluster dev/gcp): +# GITEA_URL default https://git.sientia.ai +# GITEA_USER default gitea_admin +# GITEA_PASSWORD obrigatória +# GITEA_REPO default gitea_admin/sientia-dataops-opc-ingestor +# BRANCH default main +# APP_URI default urn:sientia:opc-ingestor (SAN URI do cert — TEM que +# ser igual ao campo `uri` do documento na collection +# OPC_servers: o ingestor usa esse valor como application_uri +# e o KEPServer rejeita com BadCertificateUriInvalid se divergir) +# DAYS default 3650 +# FORCE=1 sobrescreve cert já existente no repo +set -euo pipefail + +GITEA_URL=${GITEA_URL:-https://git.sientia.ai} +GITEA_USER=${GITEA_USER:-gitea_admin} +GITEA_PASSWORD=${GITEA_PASSWORD:?export GITEA_PASSWORD com a senha do Gitea} +GITEA_REPO=${GITEA_REPO:-gitea_admin/sientia-dataops-opc-ingestor} +BRANCH=${BRANCH:-main} +APP_URI=${APP_URI:-urn:sientia:opc-ingestor} +CN=${CN:-sientia-opc-ingestor} +DAYS=${DAYS:-3650} + +CERT_DIR_IN_REPO="certs" +CERT_NAME="opc-ingestor-cert.pem" +KEY_NAME="opc-ingestor-key.pem" +DER_NAME="opc-ingestor-cert.der" + +workdir=$(mktemp -d) +trap 'rm -rf "$workdir"' EXIT + +echo ">> Gerando certificado (CN=${CN}, URI=${APP_URI}, ${DAYS} dias)..." +# Extensões exigidas por OPC UA (KEPServer valida keyUsage/EKU e o SAN URI). +openssl req -x509 -newkey rsa:2048 -sha256 -nodes -days "$DAYS" \ + -keyout "$workdir/$KEY_NAME" -out "$workdir/$CERT_NAME" \ + -subj "/CN=${CN}/O=Aignosi/OU=Sientia" \ + -addext "subjectAltName=URI:${APP_URI},DNS:${CN}" \ + -addext "keyUsage=critical,digitalSignature,nonRepudiation,keyEncipherment,dataEncipherment" \ + -addext "extendedKeyUsage=critical,clientAuth,serverAuth" \ + -addext "basicConstraints=critical,CA:FALSE" + +# Cópia DER — só referência p/ import manual no KEPServer (o trust normal é +# aceitar o cert rejeitado na 1ª conexão). +openssl x509 -in "$workdir/$CERT_NAME" -outform der -out "$workdir/$DER_NAME" + +echo ">> Clonando ${GITEA_REPO}@${BRANCH} do Gitea..." +encoded_pass=$(python3 - <&1 | sed "s#${encoded_pass}#***#g" + +if [[ -f "$workdir/repo/$CERT_DIR_IN_REPO/$CERT_NAME" && "${FORCE:-0}" != "1" ]]; then + echo "ERRO: $CERT_DIR_IN_REPO/$CERT_NAME já existe no repo. Re-rodar com FORCE=1 sobrescreve" >&2 + echo " (o cert antigo deixa de valer — retrust no KEPServer necessário)." >&2 + exit 1 +fi + +mkdir -p "$workdir/repo/$CERT_DIR_IN_REPO" +cp "$workdir/$CERT_NAME" "$workdir/$KEY_NAME" "$workdir/$DER_NAME" "$workdir/repo/$CERT_DIR_IN_REPO/" + +cd "$workdir/repo" +git add "$CERT_DIR_IN_REPO" +git -c user.name="opc-cert-script" -c user.email="platform@aignosi.com.br" \ + commit -m "Add OPC UA client certificate for KEPServer connection (URI ${APP_URI})" +git push origin "$BRANCH" 2>&1 | sed "s#${encoded_pass}#***#g" + +fingerprint=$(openssl x509 -in "$CERT_DIR_IN_REPO/$CERT_NAME" -noout -fingerprint -sha1) + +cat < ${CERT_DIR_IN_REPO}/{${CERT_NAME},${KEY_NAME},${DER_NAME}} +${fingerprint} + +Paths dentro do pod (o entrypoint clona o repo em /app): + cert_path: /app/${CERT_DIR_IN_REPO}/${CERT_NAME} + private_key_path: /app/${CERT_DIR_IN_REPO}/${KEY_NAME} + +Próximos passos: + 1. Documento na collection OPC_servers com "uri": "${APP_URI}" (ver README/PR). + 2. kubectl -n sientia rollout restart deployment/opc-ingestor # re-clona o repo + 3. Pod tenta conectar -> no KEPServer: OPC UA Configuration Manager -> + Trusted Clients -> cert '${CN}' rejeitado -> Trust -> restart runtime. +EOF diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 2cb4ae9..4c6b2b6 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -1,6 +1,5 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch -from kafka.errors import NoBrokersAvailable from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel @@ -18,14 +17,11 @@ metadata = { @fixture -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def data_manager(mongodb_repository, kafka): +def data_manager(mongodb_repository): data_manager = DataManager( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -38,141 +34,34 @@ def data_manager(mongodb_repository, kafka): return data_manager -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___success(mongodb_repository, kafka): +def test___init___success(mongodb_repository): logger_mock = MagicMock() data_manager = DataManager( metadata=metadata['metadata'], - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=logger_mock, notification_handler=MagicMock(), metrics_controller=MagicMock(), ) - kafka.assert_called_once_with( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert data_manager.kafka_producer is not None logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' + 'DataManager initialized with MongoDB servers: mongodb://localhost:27017' ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') logger_mock.error.assert_not_called() - assert logger_mock.info.call_count == 4 + assert data_manager.connection_string == 'mongodb://localhost:27017' -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___second_attempt(mongodb_repository, kafka): - kafka.side_effect = [NoBrokersAvailable, MagicMock()] - logger_mock = MagicMock() - - data_manager = DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - - kafka.assert_any_call( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert kafka.call_count == 2 - assert data_manager.kafka_producer is not None - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') - logger_mock.error.assert_called_once_with( - 'Kafka servers localhost:9092 are not available. Retrying...' - ) - assert logger_mock.info.call_count == 5 - - -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___failure_max_attempts(mongodb_repository, kafka): - kafka.side_effect = NoBrokersAvailable - logger_mock = MagicMock() - - try: - DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - except NoBrokersAvailable as e: - assert ( - str(e) - == 'NoBrokersAvailable: Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - - assert kafka.call_count == 3 - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (2) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.error.assert_called_with( - 'Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - assert logger_mock.info.call_count == 3 - - else: - raise AssertionError('Expected NoBrokersAvailable exception was not raised.') - - -def test_shutdown_has_producer(data_manager): - flush_mock = MagicMock() +def test_shutdown(data_manager): close_mock = MagicMock() - - data_manager.kafka_producer.flush = flush_mock - data_manager.kafka_producer.close = close_mock + data_manager.mongo_repository.close = close_mock data_manager.shutdown() - flush_mock.assert_called_once() close_mock.assert_called_once() -def test_shutdown_no_producer(data_manager): - data_manager.kafka_producer = None - - data_manager.shutdown() - - data_manager.logger.warning.assert_any_call( - 'Kafka producer is already closed or not initialized.' - ) - - -def test_shutdown_exception(data_manager): - data_manager.kafka_producer.flush = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.close = MagicMock() - - data_manager.shutdown() - data_manager.logger.error.assert_called_once_with('Error closing Kafka producer: Test error') - - def test_shutdown_exception_mongo(data_manager): data_manager.mongo_repository.close = MagicMock(side_effect=Exception('Test error')) @@ -187,88 +76,24 @@ def test___del__(data_manager): data_manager.shutdown.assert_called_once() -def test_delivery_report(data_manager): - msg = MagicMock() - msg.topic = 'test_topic' - msg.partition = 0 - msg.offset = 1 - - data_manager.delivery_report(msg) - - data_manager.logger.debug.assert_called_once_with( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - -def test_delivery_error(data_manager): - err = 'Test error' - data_manager.delivery_error(err) - - data_manager.logger.error.assert_called_once_with(f'Delivery failed for record : {err}') - - @mark.asyncio async def test_publish(data_manager): - topic = 'test_topic' - data = {'key': 'value'} - - # Mock the send method of the Kafka producer - send_mock = MagicMock() - data_manager.kafka_producer.send = send_mock - - # Call the publish method - await data_manager.publish(topic, data) - - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - send_mock.return_value.add_callback.assert_called_once() - - data_manager.kafka_producer.flush.assert_called_once() - - -def test_publish_no_kafka(data_manager): - data_manager.export_to_kafka = False - topic = 'test_topic' - data = {'key': 'value'} - - data_manager.publish(topic, data) - - data_manager.kafka_producer.send.assert_not_called() - - -@patch('ingestor.managers.data_manager.traceback') -@mark.asyncio -async def test_publish_error(traceback, data_manager): topic = 'test_topic' data = {'key': 'value', 'name': 'test_tag'} - # Mock the send method of the Kafka producer to raise an exception - send_mock = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.send = send_mock + data_manager.mongo_repository.insert = AsyncMock() - data_manager.mongo_repository = AsyncMock() - - # Call the publish method await data_manager.publish(topic, data) - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - # Check if the error was logged - data_manager.send_notification_async.assert_called_once_with( - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: Test error', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=traceback.format_exc.return_value, + data_manager.mongo_repository.insert.assert_called_once_with( + collection_name=topic, + document={**data, 'inserted_at': ANY}, metadata=metadata['metadata'], ) @mark.asyncio async def test_publish_error_mongo(data_manager): - data_manager.export_to_kafka = False data_manager.mongo_repository.insert = AsyncMock(side_effect=Exception('Test error')) await data_manager.publish('test_topic', {'key': 'value'}) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 32ac827..9dcdd21 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -20,14 +20,12 @@ metadata = { @patch('ingestor.managers.ingestor_manager.ResourceManager') def ingestor_manager(data_manager_mock, resource_manager_mock): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -49,14 +47,12 @@ def test___init__( notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock ): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -65,10 +61,8 @@ def test___init__( opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, metadata=metadata['metadata'], logger=ingestor.logger, notification_handler=ingestor.notification_handler, diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 63ea638..8bbccc4 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -9,8 +9,6 @@ from ingestor.ingestor import Ingestor @patch('ingestor.ingestor.NotificationHandler') def test___init__(notification_handler, getenv): getenv.side_effect = [ - 'localhost:9092,localhost:35', # KAFKA_SERVERS - 'true', # EXPORT_TO_KAFKA 'localhost', # REDIS_HOST '63790', # REDIS_PORT 'user', # REDIS_USERNAME @@ -27,7 +25,6 @@ def test___init__(notification_handler, getenv): ingestor = Ingestor() - getenv.assert_any_call('KAFKA_SERVERS', 'localhost:9092') getenv.assert_any_call('REDIS_HOST', 'localhost') getenv.assert_any_call('REDIS_PORT', '6379') getenv.assert_any_call('REDIS_USERNAME', None) @@ -37,7 +34,6 @@ def test___init__(notification_handler, getenv): getenv.assert_any_call('HOSTNAME', 'localhost') getenv.assert_any_call('POLL_INTERVAL', '5') - assert ingestor.kafka_servers == ['localhost:9092', 'localhost:35'] assert ingestor.redis_host == 'localhost' assert ingestor.redis_port == 63790 assert ingestor.redis_username == 'user' @@ -124,7 +120,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): await ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( - kafka_servers=','.join(ingestor.kafka_servers), redis_data={ 'host': ingestor.redis_host, 'port': ingestor.redis_port, @@ -139,7 +134,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): metadata=ingestor.metadata, logger=ingestor.logger, notification_handler=ingestor.notification_handler, - export_to_kafka=ingestor.export_to_kafka, metrics_controller=ingestor.metrics_controller, ) ingestor_manager.declare_active.assert_called_once() diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index 809abdf..3c50dea 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -183,30 +183,6 @@ def test_opc_reconnections_total(): assert set(metrics.OPC_RECONNECTIONS_TOTAL._labelnames) == {'pod_id', 'server_name'} -def test_kafka_messages_sent(): - """Verify the definition of KAFKA_MESSAGES_SENT.""" - assert metrics.KAFKA_MESSAGES_SENT is not None - assert isinstance(metrics.KAFKA_MESSAGES_SENT, Counter) - assert metrics.KAFKA_MESSAGES_SENT._name == 'kafka_messages_sent' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_SENT._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_messages_errors(): - """Verify the definition of KAFKA_MESSAGES_ERRORS.""" - assert metrics.KAFKA_MESSAGES_ERRORS is not None - assert isinstance(metrics.KAFKA_MESSAGES_ERRORS, Counter) - assert metrics.KAFKA_MESSAGES_ERRORS._name == 'kafka_messages_errors' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_ERRORS._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_connection_status(): - """Verify the definition of KAFKA_CONNECTION_STATUS.""" - assert metrics.KAFKA_CONNECTION_STATUS is not None - assert isinstance(metrics.KAFKA_CONNECTION_STATUS, Gauge) - assert metrics.KAFKA_CONNECTION_STATUS._name == 'kafka_connection_status' - assert set(metrics.KAFKA_CONNECTION_STATUS._labelnames) == {'pod_id'} - - def test_notifications_sent(): """Verify the definition of NOTIFICATIONS_SENT.""" assert metrics.NOTIFICATIONS_SENT is not None