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

📋 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

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

2. Create virtual environment

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

3. Install dependencies

   pip install -r requirements.txt
   

4. Create environment configuration file

   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:

   # 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. Generate an OPC UA client certificate (only if connecting to a secured OPC UA server)

   See Certificate Generation 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

1. Activate virtual environment

   source ./venv/bin/activate
   

2. 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.

🔐 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.

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).
• 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

# 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

# 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

⚙️ 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

OPC Server Configuration

OPC servers are configured through Redis with the following structure. cert_path / private_key_path are only required for secured connections (see Certificate Generation):

{
  "slot:opc_tags:1": {
    "server1": {
      "name": "server1",
      "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",
            "data_range": [
               -100,
               100
            ],
            "frequency": "15000",
            "server_id": "1",
            "tag_address": "ns=2;i=2",
            "tag_name": "Counter",
            "topics": [
               "raw_scouter-opcua-orchestrated-pipeline",
               "raw_scouter-basic-sum-model"
            ]
      }
    }
  }
}

🔧 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  # OPC UA server management
│   │   └── resource_manager.py # Resource coordination
│   ├── 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 (e.g. OPC UA 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:

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.