8ab9af170d
The Laborious no longer records OPC outcomes (it inserts the row as 'pending' and only publishes the commands), so the comments and docs that asked to keep a Laborious copy of the UPDATE in sync are updated. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
493 lines
19 KiB
Markdown
493 lines
19 KiB
Markdown
# Sientia DataOps OPC Ingestor
|
||
|
||
A high-performance, scalable OPC UA data ingestion system designed for industrial data collection and real-time streaming. The OPC Ingestor provides enterprise-grade data acquisition from OPC UA servers with automatic load balancing, fault tolerance, and comprehensive monitoring.
|
||
|
||
## Features
|
||
|
||
### 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
|
||
- **Persistent Storage**: MongoDB integration for historical data persistence
|
||
- **Health Monitoring**: Comprehensive Prometheus metrics and health checks
|
||
- **Fault Tolerance**: Automatic failover, reconnection, and error recovery
|
||
|
||
### Advanced Capabilities
|
||
- **Certificate-based Security**: Support for X.509 certificates and private keys
|
||
- **Dynamic Tag Management**: Runtime configuration updates without service interruption
|
||
- **Resource Coordination**: Redis-based slot leasing and instance coordination
|
||
- **Notification System**: Integrated alerting and notification management
|
||
- **Performance Optimization**: Configurable polling intervals and data collection frequencies
|
||
|
||
## Architecture
|
||
|
||
The OPC Ingestor uses a modular, manager-based architecture designed for scalability and fault tolerance:
|
||
|
||
```
|
||
┌────────────────────────────────────────────
|
||
│ Main App │ │ IngestorManager │ │ OPC Manager │
|
||
│ │◄──►│ │◄──►│ │
|
||
│ - Signal Hand. │ │ - Slot Mgmt │ │ - Connections │
|
||
│ - Metrics │ │ - Load Balancing │ │ - Subscriptions │
|
||
│ - Lifecycle │ │ - Coordination │ │ - Data Handler │
|
||
└─────────────────┘ └──────────────────┘ └─────────────────┘
|
||
│
|
||
▼
|
||
┌──────────────────┐ ┌─────────────────┐
|
||
│ ResourceManager │ │ DataManager │
|
||
│ │ │ │
|
||
│ - Redis Coord. │ │ - MongoDB Store │
|
||
│ - Slot Leasing │ │ - Data Pipeline │
|
||
│ - Heartbeats │ │ │
|
||
└──────────────────┘ └─────────────────┘
|
||
```
|
||
|
||
### Key Components
|
||
|
||
- **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 persistence (MongoDB)
|
||
- **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
|
||
|
||
## 🚪 OPC Gateway (SIENTIAPDE-1972)
|
||
|
||
Alongside the passive ingestion pipeline above, this service also runs an OPC
|
||
Gateway for on-demand and command-driven interactions with OPC UA servers:
|
||
|
||
- **NATS write consumer**: consumes write commands from `opc.write` (see
|
||
`NATS_WRITE_SUBJECT`) and executes them against the target OPC server
|
||
(`ingestor/adapters/nats_consumer.py`, `ingestor/engines/write.py`). See
|
||
[Write commands](#write-commands-c2-and-write_status-c3) below.
|
||
- **On-demand gRPC `ReadTag`**: a gRPC server (`ingestor/adapters/grpc_server.py`,
|
||
`ingestor/proto/opc_gateway.proto`) for synchronous, single-tag reads.
|
||
- **Scheduled active reads**: a polling read engine (`ingestor/engines/active_read.py`)
|
||
for tags that need periodic reads outside the subscription pipeline;
|
||
orchestrator-side wiring lands in SIENTIAPDE-1974.
|
||
- **Certificate management**: client certs for secure OPC connections are
|
||
fetched from MinIO + a Mongo `platform_certificates` collection
|
||
(`ingestor/connection/certs.py`) rather than shipped in the image.
|
||
|
||
See `docs/deploy-notes.md` for the deployment-facing details (env vars,
|
||
health probe, NetworkPolicy, cert cache) and the `# OPC Gateway` block in
|
||
`.env.example` for the full list of new environment variables.
|
||
|
||
### Write commands (C2) and `write_status` (C3)
|
||
|
||
The Laborious publishes one command per tag (SIENTIAPDE-1975):
|
||
|
||
```json
|
||
{"server_id": "1", "addr": "ns=2;s=Pred", "value": 1.25, "data_type": "float",
|
||
"model_id": "2", "model_name": "sin-approx", "schema": "sientia_data",
|
||
"table_name": "predictions", "timestamp": "2026-10-13T12:00:00-03:00",
|
||
"write_timestamp": false, "write_quality": false}
|
||
```
|
||
|
||
- **Server connection**: the command carries only `server_id`. The connection is
|
||
resolved by `ServerConfigResolver` (`ingestor/connection/servers.py`): `url`/
|
||
`server_uri`/`use_security` in the command when present (override), else the
|
||
live pooled connection opened by the reads, else the `connectors` document
|
||
(`type: opc_ua`, `status: Active`), else the `opc-servers` document
|
||
(`use_security = bool(cert_path)`) — the orchestrator's rule for the read slots.
|
||
Lookups are cached for 60 s. An unknown server is a failed write (`server_not_found`).
|
||
- **Outcome**: `UPDATE ... SET write_status` on the prediction row (by `timestamp` +
|
||
`model_id`), keeping the worst outcome: `ok` < `error` < `session_bad`, and
|
||
`pending`/NULL are replaced by any answer, so the order of the answers does not
|
||
matter. The gateway is the only one that updates the column: the Laborious inserts the
|
||
row as `pending` (merged with its PI Web API outcome) and never records OPC outcomes.
|
||
The UPDATE never raises; its outcome (`updated`, `kept`, `row_not_found`,
|
||
`invalid`, `error`) is logged and counted in `opc_write_status_update_total`.
|
||
The reason of a failed write is logged by `WriteEngine`.
|
||
- **NATS connection**: the first connection gives up after 5 s (it runs on the main
|
||
loop: nats-py's defaults would block it for minutes while NATS is down); once
|
||
connected, it reconnects forever. A connection closed for good is started again
|
||
by `GatewayRuntime.idle_sweep`.
|
||
|
||
## 📋 Prerequisites
|
||
|
||
- Python 3.11+
|
||
- Redis server
|
||
- MongoDB server
|
||
- OPC UA servers for data collection
|
||
|
||
**Note**: External dependencies (Redis, MongoDB) must be available either through:
|
||
- Port forwarding from a Kubernetes cluster
|
||
- External Docker Compose setup
|
||
- Cloud-managed services
|
||
- Local installations
|
||
|
||
## ️ Installation
|
||
|
||
### Local Development Setup
|
||
|
||
1. **Clone the repository**
|
||
```bash
|
||
git clone <repository-url>
|
||
cd sientia-dataops-opc-ingestor
|
||
```
|
||
|
||
2. **Create virtual environment**
|
||
```bash
|
||
python3.11 -m venv venv
|
||
source ./venv/bin/activate
|
||
```
|
||
|
||
3. **Install dependencies**
|
||
```bash
|
||
pip install -r requirements.txt
|
||
```
|
||
|
||
4. **Create environment configuration file**
|
||
```bash
|
||
cp .env.example .env
|
||
# Edit .env with your connection details
|
||
```
|
||
|
||
5. **Configure external dependencies**
|
||
|
||
You'll need to set up port forwarding or connections to external services. For example:
|
||
|
||
```bash
|
||
# Port forwarding from Kubernetes cluster
|
||
kubectl port-forward svc/redis-master 6379:6379
|
||
kubectl port-forward svc/mongodb 27017:27017
|
||
|
||
# Or connect to external Docker Compose
|
||
# Ensure services are accessible on localhost with appropriate ports
|
||
```
|
||
|
||
6. **Register the OPC UA client certificate** (only if connecting to a secured OPC UA server)
|
||
|
||
See [OPC Client Certificate](#-opc-client-certificate) below.
|
||
|
||
## Usage
|
||
|
||
### Running the Ingestor
|
||
|
||
Use the provided script to run the application locally:
|
||
|
||
```bash
|
||
# Make script executable (first time only)
|
||
chmod +x run_local.sh
|
||
|
||
# Run the application
|
||
./run_local.sh
|
||
```
|
||
|
||
The script will:
|
||
- Activate the virtual environment
|
||
- Load environment variables from `.env`
|
||
- Start the ingestor application
|
||
|
||
### Running Tests and Coverage
|
||
|
||
Use the provided script to run tests with coverage:
|
||
|
||
```bash
|
||
# Make script executable (first time only)
|
||
chmod +x run_coverage.sh
|
||
|
||
# Run tests with coverage
|
||
./run_coverage.sh
|
||
```
|
||
|
||
The script will:
|
||
- Activate the virtual environment
|
||
- Run pytest with coverage reporting
|
||
- Generate HTML coverage report
|
||
- Open the coverage report in your browser
|
||
|
||
### Manual Test Execution
|
||
|
||
You can also run tests manually:
|
||
|
||
```bash
|
||
# Activate virtual environment
|
||
source ./venv/bin/activate
|
||
|
||
# Run all tests
|
||
pytest
|
||
|
||
# Run with coverage
|
||
pytest --cov=ingestor --cov-report=html
|
||
```
|
||
|
||
### Populating Redis with OPC Configuration
|
||
|
||
1. **Activate virtual environment**
|
||
```bash
|
||
source ./venv/bin/activate
|
||
```
|
||
|
||
2. **Run the Redis feeder**
|
||
```bash
|
||
python simulator/redis-feeder.py
|
||
```
|
||
|
||
The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage.
|
||
|
||
## 🔐 OPC Client Certificate
|
||
|
||
OPC UA servers that require secure connections validate the gateway with a single,
|
||
shared platform X.509 client certificate. Every OPC connection — subscription reads,
|
||
NATS writes and gRPC reads — uses it with `Basic256Sha256` (SIENTIAPDE-1974).
|
||
|
||
The certificate is **never** read from the repository or the image. On connect, the
|
||
`CertProvider` (`ingestor/connection/certs.py`):
|
||
1. reads the single document of the Mongo collection `platform_certificates`:
|
||
`{"client_cert": {"minio_object_name": "..."}, "private_key": {"minio_object_name": "..."}}`;
|
||
2. downloads both objects from the MinIO bucket `CERT_BUCKET` (default
|
||
`streamlit-connectors`) into `CERT_CACHE_DIR` (default `/tmp/opc-certs`).
|
||
|
||
Requirements for a server to accept the connection:
|
||
- the server's slot config has `use_security: true` (a slot v1 server with a
|
||
`cert_path` is treated the same way; the path itself is ignored);
|
||
- the certificate's SAN URI equals the server's `server_uri` in the slot — it is sent
|
||
as the client `application_uri` (KEPServer rejects a mismatch with
|
||
`BadCertificateUriInvalid`);
|
||
- the server exposes a `Basic256Sha256` endpoint and trusts the certificate (e.g.
|
||
KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*, then restart
|
||
its runtime).
|
||
|
||
MinIO is only contacted when a certificate is first needed, so a MinIO outage only
|
||
affects secured servers, which are retried on the next main-loop iteration.
|
||
|
||
### Plan B: legacy read engine (`READ_ENGINE=legacy`)
|
||
|
||
Until the end of SIENTIAPDE-1974 the legacy reader is kept for validation and as a
|
||
fallback. `READ_ENGINE` selects the read path (default `v2`; an invalid value falls
|
||
back to `v2` with a warning):
|
||
|
||
| `READ_ENGINE` | Reader | Slot | Certificate |
|
||
|---|---|---|---|
|
||
| `v2` (default) | `ServerReader` + `SubscriptionEngine` over the shared pool | v1 or v2 | MinIO + Mongo (`CertProvider`), `Basic256Sha256` |
|
||
| `legacy` | the rc3 `OpcManager`, unchanged, with its own connection | **v1 only** (a v2 slot is rejected and notified) | files from the repository the pod clones (`cert_path` / `private_key_path` in the slot), `Basic256` |
|
||
|
||
Writes and gRPC always use the pool (MinIO certificate), whatever the read engine.
|
||
The legacy engine needs the certificate files in the Gitea branch the pod clones;
|
||
`scripts/generate-and-push-opc-cert.sh` generates them and pushes them to
|
||
`certs/` (`GITEA_PASSWORD='<senha>' BRANCH=<branch> ./scripts/generate-and-push-opc-cert.sh`).
|
||
Each version deploy creates a new Gitea branch without `certs/`, so they have to be
|
||
copied again for the legacy engine. Remove the legacy engine (OpcManager, its tests
|
||
and the script) once it is unused and the test suite passes the same without it.
|
||
|
||
## 🧪 Testing
|
||
|
||
### Unit Tests
|
||
```bash
|
||
# Install pytest
|
||
pip install pytest
|
||
|
||
# Run tests
|
||
pytest
|
||
|
||
# Run with coverage
|
||
pip install pytest-cov
|
||
pytest --cov=ingestor
|
||
|
||
# Generate HTML coverage report
|
||
pytest --cov=ingestor --cov-report=html
|
||
```
|
||
|
||
### Functional Tests
|
||
```bash
|
||
# Run functional tests
|
||
pytest tests/functional/
|
||
```
|
||
|
||
## 📊 Monitoring and Metrics
|
||
|
||
The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
|
||
|
||
### Application Metrics
|
||
- `app_up`: Application health status
|
||
- `app_main_loop_total`: Main loop execution count
|
||
- `app_main_loop_duration_seconds`: Loop execution time
|
||
- `app_errors_total`: Error count
|
||
|
||
### OPC Server Metrics
|
||
- `opc_connection_status`: Server connection status
|
||
- `opc_tags_subscribed_current`: Number of subscribed tags
|
||
- `opc_cycles_without_data`: Data reception health
|
||
- `opc_subscriptions_created_total`: Subscription count
|
||
|
||
### Resource Management Metrics
|
||
- `ingestor_slots_managed_current`: Slots managed by this instance
|
||
- `ingestor_active_total`: Total active ingestor instances
|
||
- `redis_connection_status`: Redis connection health
|
||
|
||
### Data Processing Metrics
|
||
- `ingestor_tag_written_count`: Data write operations
|
||
- `redis_operations_total`: Redis operation count
|
||
|
||
### OPC Gateway Metrics
|
||
- `opc_write_total`: Active write attempts (NATS-driven), by result
|
||
- `opc_write_latency_seconds`: Active write latency
|
||
- `opc_write_status_by_model`: Last write outcome per model (1 = ok, 0 = failed; label `status`)
|
||
- `opc_write_status_update_total`: `write_status` UPDATEs, by result
|
||
(`updated`, `kept`, `row_not_found`, `invalid`, `error`)
|
||
- `opc_active_read_total`: Scheduled/on-demand read attempts, by result
|
||
- `opc_error_by_type_total`: OPC errors by type (connection, conversion, session, ...)
|
||
|
||
## ⚙️ Configuration
|
||
|
||
### Environment Variables
|
||
|
||
| Variable | Description | Default | Required |
|
||
|----------|-------------|---------|----------|
|
||
| `REDIS_HOST` | Redis server hostname | `localhost` | Yes |
|
||
| `REDIS_PORT` | Redis server port | `6379` | Yes |
|
||
| `REDIS_USERNAME` | Redis username | `None` | No |
|
||
| `REDIS_PASSWORD` | Redis password | `None` | No |
|
||
| `LEASE_TTL` | Slot lease time-to-live (seconds) | `10` | No |
|
||
| `HEARTBEAT_TTL` | Heartbeat time-to-live (seconds) | `20` | No |
|
||
| `HOSTNAME` | Pod identifier | `localhost` | No |
|
||
| `POLL_INTERVAL` | Main loop polling interval (seconds) | `5` | No |
|
||
| `MONGODB_URL` | MongoDB server address | `localhost:27017` | Yes |
|
||
| `MONGODB_DATABASE` | MongoDB database name | `sientia` | No |
|
||
| `MONGODB_USERNAME` | MongoDB username | `sientia` | No |
|
||
| `MONGODB_PASSWORD` | MongoDB password | `sientia` | No |
|
||
| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No |
|
||
|
||
See the `# OPC Gateway` block in `.env.example` (`NATS_URL`, `GRPC_ENABLED`,
|
||
`CERT_BUCKET`, `WRITE_TIMESTAMP`, ...) and `docs/deploy-notes.md` for the
|
||
gateway-specific environment variables.
|
||
|
||
### OPC Server Configuration
|
||
|
||
OPC servers are configured by the Orchestrator through Redis, one key per slot
|
||
(`slot:opc_tags:<n>`). The gateway reads the **slot v2** structure (contract C1):
|
||
one entry per address, with one `names` item per collection / tag name / model, so
|
||
the same address can feed several models under different names. For secured
|
||
connections set `use_security: true` (the certificate comes from MinIO, see
|
||
[OPC Client Certificate](#-opc-client-certificate)):
|
||
|
||
```json
|
||
{
|
||
"slot:opc_tags:1": {
|
||
"server1": {
|
||
"name": "server1",
|
||
"server_id": "1",
|
||
"url": "opc.tcp://localhost:4841",
|
||
"server_uri": "urn:sientia:opc-ingestor",
|
||
"use_security": true,
|
||
"subscription_period_ms": 7500,
|
||
"tags": {
|
||
"ns=2;i=2": {
|
||
"frequency": "15000",
|
||
"active_read": {"enabled": false},
|
||
"names": [
|
||
{"collection": "raw_scouter-opcua-orchestrated-pipeline", "tag_name": "Counter", "model_name": "m1", "model_id": "1"},
|
||
{"collection": "raw_scouter-basic-sum-model", "tag_name": "Contador", "model_name": "m2", "model_id": "2"}
|
||
]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
The legacy **slot v1** (each tag with `tag_name` + `topics`, the server with
|
||
`cert_path`) is still accepted and converted on read (`ingestor/io/slots.py`): one
|
||
`names` item per topic, and `use_security` is true when `cert_path` is set (the path
|
||
itself is ignored). This lets the gateway and the Orchestrator be deployed in any
|
||
order. Documents written from a v1 slot have no `model_id`, exactly as before.
|
||
|
||
## 🔧 Development
|
||
|
||
### Project Structure
|
||
```
|
||
sientia-dataops-opc-ingestor/
|
||
├── ingestor/ # Main application code
|
||
│ ├── managers/ # Component managers
|
||
│ │ ├── data_manager.py # Data persistence and export
|
||
│ │ ├── ingestor_manager.py # Main coordination
|
||
│ │ ├── opc_manager.py # Legacy reader, plan B (READ_ENGINE=legacy)
|
||
│ │ ├── resource_manager.py # Resource coordination
|
||
│ │ └── server_reader.py # OPC UA server reads (SubscriptionEngine + pool)
|
||
│ ├── app.py # Main application entry point
|
||
│ ├── ingestor.py # Core ingestor logic
|
||
│ └── metrics.py # Prometheus metrics definitions
|
||
├── simulator/ # OPC simulation and testing tools
|
||
├── scripts/ # Ops scripts (e2e seed; legacy OPC cert generation)
|
||
├── tests/ # Test suite
|
||
└── requirements.txt # Python dependencies
|
||
```
|
||
|
||
### Adding New Features
|
||
|
||
1. **Follow the manager pattern** for new components
|
||
2. **Add comprehensive docstrings** for all public methods
|
||
3. **Include Prometheus metrics** for monitoring
|
||
4. **Add unit tests** for new functionality
|
||
5. **Update this README** with new features and configuration
|
||
|
||
## 🐛 Troubleshooting
|
||
|
||
### Common Issues
|
||
|
||
1. **OPC Connection Failures**
|
||
- Verify server URLs and network connectivity
|
||
- Check certificate paths and security settings
|
||
- Review server logs for authentication issues
|
||
|
||
2. **Redis Connection Issues**
|
||
- Verify Redis server is running and accessible
|
||
- Check authentication credentials
|
||
- Ensure proper network configuration
|
||
|
||
3. **Performance Issues**
|
||
- Monitor Prometheus metrics for bottlenecks
|
||
- Adjust polling intervals and lease TTLs
|
||
- Review OPC server performance and network latency
|
||
|
||
### Debug Mode
|
||
|
||
Enable debug logging by setting the log level in your environment:
|
||
```bash
|
||
export LOG_LEVEL=DEBUG
|
||
```
|
||
|
||
## Performance Tuning
|
||
|
||
### Key Parameters
|
||
|
||
- **`POLL_INTERVAL`**: Main loop frequency (lower = more responsive, higher = less CPU)
|
||
- **`LEASE_TTL`**: Slot lease duration (lower = faster failover, higher = more stable)
|
||
- **`HEARTBEAT_TTL`**: Instance health check frequency
|
||
- **Tag frequency**: OPC tag collection rate (Hz)
|
||
|
||
### Scaling Considerations
|
||
|
||
- **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
|
||
|
||
## 🤝 Contributing
|
||
|
||
1. Fork the repository
|
||
2. Create a feature branch
|
||
3. Make your changes with comprehensive testing
|
||
4. Update documentation and docstrings
|
||
5. Submit a pull request
|
||
|
||
### Code Quality Standards
|
||
|
||
- Follow PEP 8 style guidelines
|
||
- Include comprehensive docstrings for all public methods
|
||
- Maintain test coverage above 80%
|
||
- Use type hints where appropriate
|
||
- Follow the established architectural patterns
|
||
|
||
## 🆘 Support
|
||
|
||
For support and questions:
|
||
- Check the troubleshooting section above
|
||
- Review the metrics and logs for error patterns
|
||
- Open an issue in the project repository
|
||
- Contact the development team
|
||
|
||
---
|
||
|
||
**Note**: This OPC Ingestor is designed for production use in industrial environments. Ensure proper security configuration and network isolation for production deployments.
|