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