sientia-dataops-opc-ingestor/CLAUDE.md

# CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

## Project

OPC UA data ingestion service. Pulls tags from OPC UA servers, streams to Kafka (optional) and persists to MongoDB. Horizontally scaled via Redis-coordinated slot leasing — multiple pods share tag load without central orchestration.

Depends on internal shared lib `sientia-dataops-library` (imported as `sientia_do`, pinned in `requirements.txt` via git+ssh) for logging, notifications, metrics controller, and Mongo/Redis repositories. Base classes/utilities from `sientia_do` won't be visible in this repo — check that lib's source if behavior there is unclear.

## Commands

```bash
python3.11 -m venv venv && source ./venv/bin/activate
pip install -r requirements.txt -r requirements-dev.txt

./run_local.sh                          # loads .env, runs python -m ingestor.app
./run_coverage.sh                       # pytest --cov=ingestor --cov-report=html, opens report

pytest                                  # unit tests (testpaths = tests)
pytest tests/unit                       # unit only
pytest e2e/ -o testpaths=e2e -m integration   # E2E (Docker: testcontainers + in-process OPC servers)
pytest tests/unit/test_ingestor.py::TestClass::test_name   # single test
pytest --cov=ingestor --cov-report=html # coverage report -> htmlcov/

ruff check .                            # lint
ruff format .                           # format
mypy ingestor                           # type check
bandit -r ingestor                      # security scan
```

External deps (Redis, MongoDB, Kafka) must be reachable — via `kubectl port-forward`, local docker-compose, or cloud services. Config lives in `.env` (copy from `.env.example`).

## Architecture

Manager-based pipeline, one manager per concern, composed by `IngestorManager`:

```
app.py (asyncio loop, signals, Prometheus server)
  -> Ingestor (ingestor.py) — lifecycle: prepare_ingestor() -> loop() -> shutdown()
       -> IngestorManager (managers/ingestor_manager.py) — central coordinator
            -> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry
            -> ServerReader (managers/server_reader.py), one per OPC server — SubscriptionEngine over the shared OpcConnectionPool
            -> OpcConnectionPool (connection/pool.py) + CertProvider (connection/certs.py) — pooled connections, cert from MinIO
            -> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish
```

**Slot model**: OPC tag configs live in Redis under keys like `slot:opc_tags:<n>`, each holding one or more OPC servers and their tags (see README "OPC Server Configuration" for the JSON shape). A "slot" is the unit of lease/assignment; each ingestor pod leases some number of slots via `ResourceManager` and only subscribes to the tags in slots it holds.

**Main loop** (`Ingestor.loop()` in `ingestor.py`) each cycle:
1. `declare_active()` + heartbeat (keeps this pod visible to peers via Redis TTL keys — `LEASE_TTL`/`HEARTBEAT_TTL`)
2. Reads current active-ingestor count, lease count, slot count
3. `manage_no_slots()` — grabs one slot if this pod is idle and slots exist
4. `manage_leases()` — rebalances: acquires slots if other pods are lacking, drops extras (keeps at most 1 slot per pod when supply is sufficient) — this is the load-balancing algorithm
5. `update_slot_config()` + `check_opc_servers_integrity()` — refresh OPC configs, detect stalled connections
6. `update_ingestor_manager()` — diffs current vs previous managed tags, subscribes new, resubscribes changed, unsubscribes removed

Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives).

**ServerReader** (SIENTIAPDE-1974) reads one OPC server through a `SubscriptionEngine`: the connection comes from the shared `OpcConnectionPool` (also used by NATS writes and gRPC). Slots are slot v2 only, parsed by `io/slots.py` (`parse_slot`): each server carries `security` (`mode`, `policy`) and `certificate` (certificate_id, application_uri, MinIO `bucket`/`object_name` of cert and private_key, from `opc_certificates` via the Orchestrator; `None` = connect without security, D5). The pool applies security only with a certificate (`connection/security.py`: `is_secured`, `apply_security` maps mode/policy to asyncua, unknown → error), and `CertProvider` downloads each certificate into `<CERT_CACHE_DIR>/<certificate_id>/` (cache, lock and `invalidate` per certificate) — no certificate files in the repo/image, no default certificate. A change of `security`/`certificate` in the slot re-initializes the reader (pool connection evicted). `datachange_notification` fans each value out to every `names` entry via `DataSink`. Subscribed connections are pinned in the pool so the idle sweep never drops them. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles` / `check_opc_listenning`); a lost server is shut down (pool connection evicted) and re-initialized on the next loop.

**Writes** (SIENTIAPDE-1975): `NatsWriteConsumer` → `WriteEngine` → pool, then `DataSink.update_write_status`. Commands (C2) carry only `server_id`; `ServerConfigResolver` (`connection/servers.py`) resolves the connection: live pooled connection → `connectors` aggregation with a `$lookup` on `opc_certificates`, in the slot format. The certificate rule (active + client, `connection/certificates.py`) mirrors the Orchestrator's `tag_database.resolve_certificate` — keep both in sync (SIENTIAPDE-1974 D1; the Mongo fallback goes away when writes are routed to the pod that owns the connection). `write_status` keeps the worst outcome (`ok` < `error` < `session_bad`); the gateway is the only one that updates it (the Laborious inserts the row as `pending` and never records OPC outcomes). The NATS first connect is bounded (it runs on the main loop); reconnection after it is unbounded.

**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.

**Error handling convention**: broad `except Exception` + `traceback.print_exc()` + notification-handler alert is intentional throughout (see `ruff.lint.ignore` for `BLE001` in `pyproject.toml`) — this is an always-on service where a single tag/server failure must not kill the pod; failures are surfaced via notifications/metrics instead of raised.

## Config

Env vars documented in README ("Configuration" section) and `.env.example`. Notable ones affecting behavior: `LEASE_TTL`/`HEARTBEAT_TTL` (failover speed vs stability), `POLL_INTERVAL` (loop cadence), `EXPORT_TO_KAFKA`.

## Testing notes

- `tests/unit/managers/` mirrors `ingestor/managers/` 1:1.
- `e2e/` is the end-to-end suite (Laborious/Orchestrator pattern): real MongoDB/Redis/Postgres/MinIO/NATS via testcontainers, in-process asyncua servers, pods built like `app.py` (`e2e/helpers.Pod`). Scenarios are numbered by topic in `e2e/scenarios.md`, 1:1 with `test_scenario_<n>_<n>_<n>_*` — keep both in sync. Marked `integration`, not collected by the default `pytest`.
- Coverage config (`pyproject.toml` / `.coveragerc`) excludes `ingestor/app.py` (thin entrypoint, signal wiring) — don't chase coverage there.
- `ruff` complexity cap is `max-complexity = 15` (mccabe) — factor out branches in manager methods before that.