gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #29 from Aignosi/fix/SAM0123-231

Browse Source 
Python Dependency Refactor, Local Dev Enhancements, and OPC UA Cert Automation
This commit is contained in:
vitor-aignosi
2026-07-23 14:15:24 -03:00
committed by GitHub
parent 498030dd5f f75aa322cb
commit c29b06288c
14 changed files with 172 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@@ -1,7 +1,3 @@
# Kafka
KAFKA_SERVERS="localhost:9092"
EXPORT_TO_KAFKA="false"
# Redis
REDIS_HOST="localhost"
REDIS_PORT="6379"

5
.github/workflows/quality-gate.yml vendored
View File

@@ -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

85
README.md
View File

@@ -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='<senha-do-gitea>' ./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.

16
ingestor/ingestor.py
View File

@@ -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

160
ingestor/managers/data_manager.py
View File

@@ -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,

10
ingestor/managers/ingestor_manager.py
View File

@@ -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,

17
ingestor/metrics.py
View File

@@ -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',

5
requirements-local.txt Normal file
View File

@@ -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

2
requirements.txt
View File

@@ -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

95
scripts/generate-and-push-opc-cert.sh Executable file
View File

@@ -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='<senha>' ./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 - <<EOF
import urllib.parse; print(urllib.parse.quote('''${GITEA_PASSWORD}''', safe=''))
EOF
)
clone_url="${GITEA_URL%%/}"
clone_url="${clone_url/:\/\//:\/\/${GITEA_USER}:${encoded_pass}@}/${GITEA_REPO}.git"
git clone --depth 1 --branch "$BRANCH" "$clone_url" "$workdir/repo" 2>&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 <<EOF
== PRONTO ==
Cert no Gitea: ${GITEA_REPO}@${BRANCH} -> ${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

195
tests/unit/managers/test_data_manager.py
View File

@@ -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'})

6
tests/unit/managers/test_ingestor_manager.py
View File

@@ -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,

6
tests/unit/test_ingestor.py
View File

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

24
tests/unit/test_metrics.py
View File

@@ -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