SAM0123-231: Remove Kafka integration for data streaming and add comprehensive OPC UA client certificate management.

This commit is contained in:
vitor-aignosi
2026-07-23 13:37:56 -03:00
parent b10c226e4e
commit 648458510d

View File

@@ -7,7 +7,6 @@ A high-performance, scalable OPC UA data ingestion system designed for industria
### Core Functionality ### Core Functionality
- **OPC UA Integration**: Native support for OPC UA servers with secure and unsecured connections - **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 - **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 - **Persistent Storage**: MongoDB integration for historical data persistence
- **Health Monitoring**: Comprehensive Prometheus metrics and health checks - **Health Monitoring**: Comprehensive Prometheus metrics and health checks
- **Fault Tolerance**: Automatic failover, reconnection, and error recovery - **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 │ │ ResourceManager │ │ DataManager │
│ │ │ │ │ │ │ │
│ - Redis Coord. │ │ - Kafka Export │ - Redis Coord. │ │ - MongoDB Store
│ - Slot Leasing │ │ - MongoDB Store │ │ - Slot Leasing │ │ - Data Pipeline │
│ - Heartbeats │ │ - 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 - **Ingestor**: Main application orchestrator managing the overall lifecycle
- **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing - **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing
- **OPC Manager**: Handles individual OPC UA server connections and tag subscriptions - **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 - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
## 📋 Prerequisites ## 📋 Prerequisites
@@ -55,10 +54,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
- Python 3.11+ - Python 3.11+
- Redis server - Redis server
- MongoDB server - MongoDB server
- Kafka cluster (optional, for data streaming)
- OPC UA servers for data collection - 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 - Port forwarding from a Kubernetes cluster
- External Docker Compose setup - External Docker Compose setup
- Cloud-managed services - Cloud-managed services
@@ -99,12 +97,15 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
# Port forwarding from Kubernetes cluster # Port forwarding from Kubernetes cluster
kubectl port-forward svc/redis-master 6379:6379 kubectl port-forward svc/redis-master 6379:6379
kubectl port-forward svc/mongodb 27017:27017 kubectl port-forward svc/mongodb 27017:27017
kubectl port-forward svc/kafka 9092:9092
# Or connect to external Docker Compose # Or connect to external Docker Compose
# Ensure services are accessible on localhost with appropriate ports # 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 ## Usage
### Running the Ingestor ### Running the Ingestor
@@ -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. 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 ## 🧪 Testing
### Unit Tests ### 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_slots_managed_current`: Slots managed by this instance
- `ingestor_active_total`: Total active ingestor instances - `ingestor_active_total`: Total active ingestor instances
- `redis_connection_status`: Redis connection health - `redis_connection_status`: Redis connection health
- `kafka_connection_status`: Kafka connection health
### Data Processing Metrics ### Data Processing Metrics
- `ingestor_tag_written_count`: Data write operations - `ingestor_tag_written_count`: Data write operations
- `kafka_messages_sent_total`: Kafka message count
- `redis_operations_total`: Redis operation count - `redis_operations_total`: Redis operation count
## ⚙️ Configuration ## ⚙️ Configuration
@@ -228,8 +259,6 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
| Variable | Description | Default | Required | | 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_HOST` | Redis server hostname | `localhost` | Yes |
| `REDIS_PORT` | Redis server port | `6379` | Yes | | `REDIS_PORT` | Redis server port | `6379` | Yes |
| `REDIS_USERNAME` | Redis username | `None` | No | | `REDIS_USERNAME` | Redis username | `None` | No |
@@ -246,7 +275,7 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
### OPC Server Configuration ### 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 ```json
{ {
@@ -256,6 +285,8 @@ OPC servers are configured through Redis with the following structure:
"server_id": "1", "server_id": "1",
"url": "opc.tcp://localhost:4841", "url": "opc.tcp://localhost:4841",
"server_uri": "http://opcua-server.simulator", "server_uri": "http://opcua-server.simulator",
"cert_path": "/app/certs/opc-ingestor-cert.pem",
"private_key_path": "/app/certs/opc-ingestor-key.pem",
"tags": { "tags": {
"ns=2;i=2": { "ns=2;i=2": {
"aggr_func": "avg", "aggr_func": "avg",
@@ -292,6 +323,7 @@ sientia-dataops-opc-ingestor/
│ ├── ingestor.py # Core ingestor logic │ ├── ingestor.py # Core ingestor logic
│ └── metrics.py # Prometheus metrics definitions │ └── metrics.py # Prometheus metrics definitions
├── simulator/ # OPC simulation and testing tools ├── simulator/ # OPC simulation and testing tools
├── scripts/ # Ops scripts (e.g. OPC UA cert generation)
├── tests/ # Test suite ├── tests/ # Test suite
├── docker-compose.yaml # Infrastructure services ├── docker-compose.yaml # Infrastructure services
└── requirements.txt # Python dependencies └── requirements.txt # Python dependencies
@@ -319,12 +351,7 @@ sientia-dataops-opc-ingestor/
- Check authentication credentials - Check authentication credentials
- Ensure proper network configuration - 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 - Monitor Prometheus metrics for bottlenecks
- Adjust polling intervals and lease TTLs - Adjust polling intervals and lease TTLs
- Review OPC server performance and network latency - Review OPC server performance and network latency
@@ -350,7 +377,7 @@ export LOG_LEVEL=DEBUG
- **Horizontal Scaling**: Deploy multiple ingestor instances for high availability - **Horizontal Scaling**: Deploy multiple ingestor instances for high availability
- **Load Distribution**: Use Redis-based slot allocation for automatic load balancing - **Load Distribution**: Use Redis-based slot allocation for automatic load balancing
- **Resource Limits**: Monitor CPU, memory, and network usage - **Resource Limits**: Monitor CPU, memory, and network usage
- **Database Performance**: Optimize MongoDB indexes and Kafka partitioning - **Database Performance**: Optimize MongoDB indexes
## 🤝 Contributing ## 🤝 Contributing