396 lines
13 KiB
Markdown
396 lines
13 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
|
||
- **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
|
||
|
||
### 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. │ │ - Kafka Export │
|
||
│ - Slot Leasing │ │ - MongoDB Store │
|
||
│ - Heartbeats │ │ - Data Pipeline │
|
||
└──────────────────┘ └─────────────────┘
|
||
```
|
||
|
||
### 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 streaming (MongoDB | Kafka)
|
||
- **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
|
||
|
||
## 📋 Prerequisites
|
||
|
||
- 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:
|
||
- 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
|
||
kubectl port-forward svc/kafka 9092:9092
|
||
|
||
# Or connect to external Docker Compose
|
||
# Ensure services are accessible on localhost with appropriate ports
|
||
```
|
||
|
||
## 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.
|
||
|
||
## 🧪 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
|
||
- `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
|
||
|
||
### Environment Variables
|
||
|
||
| 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 |
|
||
| `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:
|
||
|
||
```json
|
||
{
|
||
"slot:opc_tags:1": {
|
||
"server1": {
|
||
"name": "server1",
|
||
"server_id": "1",
|
||
"url": "opc.tcp://localhost:4841",
|
||
"server_uri": "http://opcua-server.simulator",
|
||
"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
|
||
├── tests/ # Test suite
|
||
├── docker-compose.yaml # Infrastructure services
|
||
└── 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. **Kafka Export Failures**
|
||
- Verify Kafka cluster is running
|
||
- Check broker addresses and network connectivity
|
||
- Review topic configuration and permissions
|
||
|
||
4. **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 and Kafka partitioning
|
||
|
||
## 🤝 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
|
||
|
||
## 📄 License
|
||
|
||
This project is licensed under the terms specified in the LICENSE file.
|
||
|
||
## 🆘 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.
|
||
|
||
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. |