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 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: each OPC connector has its own client certificate (opc_certificates), named in the slot and downloaded from MinIO (ingestor/connection/certs.py) rather than shipped in the image.

See Gateway environment variables and the # OPC Gateway blocks in .env.example for the new environment variables.

Write commands (C2) and `write_status` (C3)

The Laborious publishes one command per tag (SIENTIAPDE-1975):

{"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"}

Written DataValue (SIENTIAPDE-2202): the value, plus the prediction timestamp as SourceTimestamp and a Good StatusCode. Timestamp and quality can be turned off separately; the value is always written. Each flag is taken from the command (write_timestamp / write_quality) when present, else from the env (WRITE_TIMESTAMP / WRITE_QUALITY), else true. The envs apply to every server of the gateway (per-server config is a future SIENTIAPDE-2276 card).
Server connection: the command carries only server_id. The connection is resolved by ServerConfigResolver (ingestor/connection/servers.py): the live pooled connection opened by the reads, else the connectors document (type: opc_ua, status: Active) joined with opc_certificates, with the orchestrator's rule for the read slots (ingestor/connection/certificates.py), in the slot format (url, security, certificate). Lookups are cached for 60 s. An unknown or incomplete 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

Clone the repository

git clone <repository-url>
cd sientia-dataops-opc-ingestor

Create virtual environment

python3.11 -m venv venv
source ./venv/bin/activate

Install dependencies
```
pip install -r requirements.txt
```

Create environment configuration file

cp .env.example .env
# Edit .env with your connection details

Configure external dependencies

You'll need to set up port forwarding or connections to external services. For example:

# 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

Register the OPC UA client certificate (only if connecting to a secured OPC UA server)

See OPC Client Certificate below.

Usage

Running the Ingestor

Use the provided script to run the application locally:

# 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:

# 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:

# 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

Activate virtual environment
```
source ./venv/bin/activate
```
Run the Redis feeder
```
python simulator/redis-feeder.py
```

The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage.

🔐 OPC Client Certificate

Each OPC connector has its own X.509 client certificate (SIENTIAPDE-1974): the connector's security.client_certificate_id points to an opc_certificates document (kind: client, status: active). The Orchestrator joins them and writes, for each server of the slot, its security (mode, policy) and its certificate:

"certificate": {
  "certificate_id": 1,
  "application_uri": "urn:sientia:opc-ingestor",
  "cert": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-cert.pem"},
  "private_key": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-key.pem"}
}

The certificate is never read from the repository or the image. On connect, the CertProvider (ingestor/connection/certs.py) downloads both objects from MinIO into CERT_CACHE_DIR/<certificate_id>/ (default /tmp/opc-certs), and caches them per certificate. When the slot names other objects for the same certificate (it was replaced), the server config changes, the connection is re-opened and the files are downloaded again. A failed secured connection also forces a new download.

Security only with a certificate. With a certificate, the connection uses the slot's security.mode (Sign / SignAndEncrypt) and security.policy (Basic256Sha256, Aes128Sha256RsaOaep, Aes256Sha256RsaPss, Basic256, Basic128Rsa15); an unknown one fails the connection. Without a certificate (null), or with mode: None, it connects without security, and a server that requires a certificate refuses the connection (the error shows in the log and the connection metrics). There is no default certificate.
The certificate's application_uri (its SAN URI) is sent as the client application URI (KEPServer rejects a mismatch with BadCertificateUriInvalid).
The server must trust 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 (one client per bucket), so a MinIO outage only affects secured servers, which are retried on the next main-loop iteration.

🧪 Testing

Unit Tests

# 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

End-to-end Tests

Every dependency runs for real (MongoDB, Redis, PostgreSQL, MinIO and NATS via testcontainers; OPC UA servers in process). Requires Docker; not part of the default pytest run.

pytest e2e/ -o testpaths=e2e -m integration

Numbered scenarios (slot leasing, reads, writes, gRPC, active reads, certificates, lifecycle): e2e/scenarios.md. Setup and layout: e2e/README.md.

📊 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

Gateway environment variables

Read by GatewayConfig (ingestor/config.py). Booleans accept true/1/yes (case-insensitive); anything else is false.

Variable	Description	Default
`NATS_URL`	NATS server for write commands (C2)	`nats://localhost:4222`
`NATS_WRITE_SUBJECT`	Subject of the write commands	`opc.write`
`NATS_QUEUE`	Queue group shared by the gateway pods	`gateway-writers`
`GRPC_ENABLED`	Starts the on-demand `ReadTag` gRPC server	`false`
`GRPC_PORT`	gRPC server port	`50051`
`CERT_CACHE_DIR`	Local cache of the client certificates downloaded from MinIO	`/tmp/opc-certs`
`WRITE_TIMESTAMP`	Writes the prediction `timestamp` as `SourceTimestamp` when the command has no `write_timestamp`	`true`
`WRITE_QUALITY`	Writes a Good `StatusCode` when the command has no `write_quality`	`true`
`WRITE_MIN_PAUSE_S`	Minimum pause between two OPC writes (seconds)	`0.0`
`IDLE_CONN_MAX_S`	Idle time before an unpinned pool connection is closed (seconds)	`300.0`

WRITE_TIMESTAMP / WRITE_QUALITY are independent and apply to every OPC server of the gateway; the value is always written. See Write commands for the precedence (command → env → true). MinIO (MINIO_*) and Postgres (POSTGRES_*) settings are listed in .env.example.

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. Each server carries its security and its certificate (null = no security, see OPC Client Certificate):

{
  "slot:opc_tags:1": {
    "server1": {
      "name": "server1",
      "server_id": "kepserver",
      "url": "opc.tcp://localhost:4841",
      "subscription_period_ms": 7500,
      "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256"},
      "certificate": {
        "certificate_id": 1,
        "application_uri": "urn:sientia:opc-ingestor",
        "cert": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-cert.pem"},
        "private_key": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-key.pem"}
      },
      "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"}
          ]
        }
      }
    }
  }
}

Only the slot v2 is accepted (ingestor/io/slots.py): a slot without security, a tag without names, or a name without collection, tag_name or model_id is rejected and notified (OPC_SLOT_CONFIG_ERROR_<slot>), and the last good config keeps being read. The gateway and the Orchestrator of SIENTIAPDE-1974 are deployed together.

🔧 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
│   │   ├── 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)
├── tests/                   # Test suite
└── requirements.txt         # Python dependencies

Adding New Features

Follow the manager pattern for new components
Add comprehensive docstrings for all public methods
Include Prometheus metrics for monitoring
Add unit tests for new functionality
Update this README with new features and configuration

🐛 Troubleshooting

Common Issues

OPC Connection Failures
- Verify server URLs and network connectivity
- Check certificate paths and security settings
- Review server logs for authentication issues
Redis Connection Issues
- Verify Redis server is running and accessible
- Check authentication credentials
- Ensure proper network configuration
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:

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

Fork the repository
Create a feature branch
Make your changes with comprehensive testing
Update documentation and docstrings
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.

README.md Unescape Escape

Sientia DataOps OPC Ingestor