f79d51336f
Review feedback (Vitor): gating the batch-max normalization behind max_count == 1 just hid the same collapsing behavior for the single-point case, where it's redundant anyway (aggregation/export don't need it) while still being the wrong fix conceptually. Removed the normalization block entirely so every record always keeps its own real timestamp from the PI Web API, regardless of max_count. Updated unit tests accordingly and confirmed no e2e scenario needs adjustment: aggregate_data groups by (tag, name) and always collapses to one output row per tag per execution, independent of per-record timestamp values, so pi_web_api_scouter_multiple_tags' aggregated-value assertions are unaffected. SIENTIAPDE-1993 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
81 lines
6.9 KiB
Markdown
81 lines
6.9 KiB
Markdown
# CLAUDE.md
|
|
|
|
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
|
|
|
## What this is
|
|
|
|
Temporal.io worker for industrial time-series data ingestion (`scouter`). Pulls data from MongoDB / PI Web API, runs quality-gate filtering + aggregation, writes to PostgreSQL, caches state in Redis. Heavy dependency on the private `sientia_do` library (aka `sientia-dataops-library`) for base activity classes (Postgres/Redis/Mongo repos), logging, notifications, metrics, retry policy, and connector config builders — read that lib's source when a base method/behavior isn't in this repo.
|
|
|
|
## Setup
|
|
|
|
```bash
|
|
python3.11 -m venv venv
|
|
source ./venv/bin/activate
|
|
pip install -r requirements-local.txt # uses git+ssh to sientia-dataops-library (needs SSH access)
|
|
pip install -r requirements-dev.txt # lint/test/e2e tooling
|
|
cp .env.example .env # fill in connection details
|
|
```
|
|
|
|
`requirements.txt` uses the plain package name `sientia_do` (resolved via index/private registry in CI); `requirements-local.txt` pins it via `git+ssh://...sientia-dataops-library.git@<tag>` for local dev — keep both in sync when bumping the library version (see `git-requirements-mapping.txt`).
|
|
|
|
## Commands
|
|
|
|
```bash
|
|
./run_local.sh # activate venv, load .env, run worker
|
|
python -m scouter.worker.worker # run worker manually
|
|
|
|
pytest # unit tests (tests/), Docker-free
|
|
pytest tests/activities/test_redis.py # single file
|
|
pytest tests/workflow/test_scouter.py::TestClassName::test_name # single test
|
|
pytest --cov=scouter --cov-report=html # coverage (or ./run_coverage.sh, opens browser)
|
|
|
|
# e2e tests are excluded from default `pytest` runs (testpaths=tests) — invoke explicitly:
|
|
pytest e2e/ --override-ini testpaths=e2e -m e2e -v
|
|
pytest e2e/ --override-ini testpaths=e2e -m e2e -x # stop on first failure
|
|
|
|
ruff check . # lint
|
|
ruff format . # format
|
|
mypy scouter # type check
|
|
bandit -r scouter # security scan
|
|
```
|
|
|
|
CI quality gate (`.github/workflows/quality-gate.yml`) calls a shared reusable workflow (`Aignosi/github_workflow_templates`) with `requirements_file: requirements-local.txt`; releases (`release.yml`) fire on merged PRs to `main` via another shared workflow. Both are effectively opaque here — the actual lint/test/coverage steps live in the `github_workflow_templates` repo.
|
|
|
|
## Architecture
|
|
|
|
Parent/child Temporal workflow pattern, two independent worker task queues sharing the same `CoreScouter` sub-workflow:
|
|
|
|
```
|
|
Scouter (scouter/workflow/scouter.py) PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
|
|
reads raw_{schedule_name} from MongoDB queries PI Web API directly (get_tag_values)
|
|
tracks last-processed timestamp in Redis preserves each record's real timestamp
|
|
│ │
|
|
└───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
|
|
1. data_quality_gate (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)
|
|
2. aggregate_data (per-tag: lts/avg/mdn/max/min)
|
|
3. group_and_hold_data (Redis, TTL = retention_time)
|
|
4. export_data_to_postgres
|
|
5. write_metrics (Prometheus)
|
|
6. store_data_package (optional, MongoDB, if debug_data_package=true)
|
|
```
|
|
|
|
- **`Activities`** (`scouter/activities/activities.py`) is a multiple-inheritance god-class combining `Postgres` (from `sientia_do`), `Redis`, `Gates`, `MongoDB`, `API` (this repo's `scouter/activities/*.py`). Each mixin owns its own `__init__`/`close` — the worker calls all of them explicitly. When adding a new activity, decide which mixin it belongs to; don't bypass the split.
|
|
- **Temporal sandboxing**: every module doing non-workflow-safe imports (pandas, sientia_do, os, etc.) wraps them in `with workflow.unsafe.imports_passed_through():`. Follow this pattern in any new workflow/activity file — Temporal's sandbox will fail otherwise.
|
|
- **Two task queues, one worker process**: `worker.py` builds two `prepare_worker(...)` instances (main workflow `Scouter` vs `PIWebAPIScouter`, both with `CoreScouter` as a registered child) and runs them concurrently via `asyncio.gather`. If either crashes, the whole process exits non-zero (see the `finally` block) — that's deliberate, for k8s restart semantics.
|
|
- **Aggregation functions**: `lts` (latest), `avg`, `mdn` (median), `max`, `min` — configured per-tag in `model_tags[tag].aggr_function`.
|
|
- **Quality filters**: `NULL_VALUES_FILTER`, `OUT_OF_BOUNDS_FILTER` (`scouter/utils/quality/filters.py`), each with a configurable `policy` (currently `DISCARD`). Register new filters in the `quality_gate_filters` dict in `scouter/activities/gates.py`.
|
|
- **Config builders** (`build_postgres_config`, `build_redis_config`, `build_mongodb_config`, `build_api_config`) come from `sientia_do.utils.connectors_config`, not this repo — `scouter/utils/connectors_config.py` only has `build_kafka_config` locally.
|
|
- **Metrics** (`scouter/metrics.py`): `APP_UP` gauge, `LABORIOUS_DATA_WRITTEN_COUNT` counter, `TAG_CHANGES_MONITOR` gauge, all labeled by `pod_id`/`model_name`/`workflow_name` (+`tag_name` for the last one). `sonar-project.properties` excludes `scouter/worker/worker.py` from coverage — it's the process entrypoint, exercised by e2e, not unit tests.
|
|
|
|
## Testing structure
|
|
|
|
- `tests/` — Docker-free unit tests, mirrors `scouter/` package layout (`tests/activities/`, `tests/workflow/`, `tests/utils/`). Default `pytest` invocation only runs this dir (`testpaths=tests` in `pyproject.toml`).
|
|
- `e2e/` — production-faithful tests against real testcontainers (Postgres/Mongo/Redis) + `WorkflowEnvironment.start_local()` + an in-process PI Web API test server (`e2e/pi_web_api_test_server.py`). Only `Logger` and an optional notification-insert spy are mocked; everything else is real `Activities` wiring. Must be invoked explicitly (see Commands) — not picked up by default `pytest`.
|
|
- Scenario catalog lives in `e2e/scenarios.md`, numbered and mapped to `test_scenario_*` functions. Section `## 0` = harness smoke tests (`e2e/test_harness_smoke.py`) that check Docker/container/worker wiring liveness before the real scenarios run.
|
|
- A scenario that fails due to a genuine production defect gets marked `xfail(strict=True)` rather than deleted/skipped silently.
|
|
- Markers: `asyncio`, `e2e`, `integration`, `unit` (declared in `pyproject.toml`).
|
|
|
|
## Root-level scripts
|
|
|
|
`get_data_pims.py`, `pi_web_api_fetch_data.py`, `init_port_forward.sh` are standalone helper/debug scripts (PI/PIMS data fetch, k8s port-forwarding for local dev against a real cluster) — not part of the worker's runtime path.
|