# 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

## 🚪 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](#write-commands-c2-and-write_status-c3) 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](#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):

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

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. **Register the OPC UA client certificate** (only if connecting to a secured OPC UA server)

   See [OPC Client Certificate](#-opc-client-certificate) 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.

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

```json
"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
```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

### 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](#write-commands-c2-and-write_status-c3)
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](#-opc-client-certificate)):

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

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.