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

   See [Certificate Generation](#-certificate-generation) 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.

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

```bash
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](#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
```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

## ⚙️ 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](#-certificate-generation)):

```json
{
  "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:
```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.