diff --git a/README.md b/README.md index 4acd9be..c7667c3 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ An `OutboxEvent` is written in the same transaction as any observation or alert, - **Elasticsearch CQRS Projection** — `EsIndexerService` consumer group upserts `patient_encounters` documents, appends to the `observations` index, and updates `openAlertCount` on alert events; patient/encounter search; per-encounter observation trend (hourly avg/min/max); alert volume summary by department and severity; population query (numeric range aggregation across all patients) - **Sepsis Early Warning Engine** — `SepsisEngineService` Kafka consumer evaluates SIRS criteria (temperature, heart rate, respiratory rate, WBC) per encounter using Redis keys with a 30-minute TTL sliding window; on ≥2 active criteria, inserts a `SEPSIS_WARNING / CRITICAL` alert idempotently (`INSERT WHERE NOT EXISTS`) - **RabbitMQ Notification Workers** — `NotificationPublisherService` reads `alert.generated` from Kafka and publishes paging jobs to `alerts.paging.queue`; `PagingWorkerService` sends the page and waits for acknowledgment; if no ack arrives before timeout it NACKs to `alerts.paging.dlq` with `x-message-ttl = 300000ms`; if the host is stopping, in-flight paging messages are NACKed with `requeue=true` so they are retried after restart and do not false-escalate; `EscalationWorkerService` pages the on-call backup and sets alert status to `escalated`; `DischargeSummaryWorkerService` reads `encounter.status.changed`, generates a discharge summary, and stores it in MinIO under `/discharge-summaries/{encounterId}/summary.pdf` -- **Data Lake Writer (Phase 9 - in progress)** — `DataLakeWriterService` Kafka consumer buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/` by date), and commits offsets after successful uploads +- **Data Lake Writer** — `DataLakeWriterService` (consumer group `data-lake-writer`) buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes date-partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/`), and commits Kafka offsets only after successful uploads; `kafka_partition` and `kafka_offset` columns provide audit lineage - **Reconciliation Jobs** — three scheduled checks: (1) unacknowledged CRITICAL alerts older than 30 minutes, (2) pending orders without results after 4 hours, (3) active inpatients with no observation in 2 hours; each finding creates a `reconciliation_alerts` row and publishes to RabbitMQ - **Standard Envelope** — all responses use a consistent `{ success, statusCode, data, error }` wrapper; validation errors use the same shape; `ApiBehaviorOptions` overridden so model validation also produces the standard envelope - **Observability** — Serilog structured logging enriched with `correlationId`, `encounterId`, `patientId` on alert paths; Seq sink (`http://localhost:5345`); Prometheus (`http://localhost:9101`) scrapes `GET /metrics`; Grafana dashboards (`http://localhost:3101`, admin/admin) for clinical metrics including `alerts_unacknowledged_gauge`; per-request correlation IDs in request logs and response headers @@ -225,6 +225,14 @@ tests/ ├── ReconciliationTests.cs # Three reconciliation checks, deduplication, RabbitMQ publish ├── ObservabilityPhase8Tests.cs # /metrics families and correlation header behavior └── DataLakePhase9Tests.cs # Kafka → MinIO Parquet flow and schema checks + +scripts/ +├── run-phase8-verification.sh # Prometheus + alerts_unacknowledged_gauge checks +└── run-phase9-verification.sh # Data lake integration tests + Kafka + MinIO + DuckDB + +docs/ +├── plans/phase-9-plan.md # Phase 9 implementation and verification guide +└── decisions/data-lake-design.md # Parquet vs JSON, partitioning, replay rationale ``` --- @@ -337,7 +345,8 @@ On startup the application: 3. Pre-loads all thresholds into Redis 4. Provisions Kafka topics and Elasticsearch indices 5. Declares the RabbitMQ exchange and queue topology -6. Starts the reconciliation scheduler (three safety checks on a configurable interval) +6. Starts the data lake writer (`data-lake-writer` → Parquet in MinIO) +7. Starts the reconciliation scheduler (three safety checks on a configurable interval) Swagger UI is available at `http://localhost:/swagger` in Development. @@ -349,6 +358,30 @@ dotnet test Tests use Testcontainers to spin up a real PostgreSQL instance. No manual setup required. +### Verification Scripts + +With the API running and Docker Compose up, run phase verification end-to-end: + +```bash +./scripts/run-phase8-verification.sh # Prometheus metrics, alerts_unacknowledged_gauge +./scripts/run-phase9-verification.sh # Data lake tests, Kafka offsets, MinIO Parquet, DuckDB schema +``` + +Phase 9 optional tools (install without sudo): + +```bash +# MinIO client — object listing and download +mkdir -p ~/.local/bin +curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o ~/.local/bin/mc +chmod +x ~/.local/bin/mc + +# DuckDB — query Parquet files locally +curl https://install.duckdb.org | sh +export PATH="$HOME/.duckdb/cli/latest:$HOME/.local/bin:$PATH" +``` + +See `docs/plans/phase-9-plan.md` for manual Kafka replay and DuckDB query examples. + --- ## API Reference @@ -828,4 +861,4 @@ Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Off | 6 | RabbitMQ exchange and queue topology; `NotificationPublisherService`; `PagingWorkerService`; DLQ escalation (`EscalationWorkerService`); discharge summary (`DischargeSummaryWorkerService` → MinIO); integration tests | Done | | 7 | Reconciliation scheduler — unacknowledged critical alerts, stale pending orders, disconnected monitors; `reconciliation_alerts` table; RabbitMQ publish; integration tests | Done | | 8 | Prometheus metrics (`GET /metrics`); Grafana dashboards; eight application metric families | In progress | -| 9 | Data lake writer — Kafka consumer group `data-lake-writer`; Parquet flush to MinIO; integration tests (`DataLakePhase9Tests`) | In progress | +| 9 | Data lake writer — `data-lake-writer` consumer group; Parquet flush to MinIO; `DataLakePhase9Tests`; `run-phase9-verification.sh` | Done | diff --git a/docs/docker-compose-usage-and-troubleshooting.md b/docs/docker-compose-usage-and-troubleshooting.md index 52b4d2e..3dd65f8 100644 --- a/docs/docker-compose-usage-and-troubleshooting.md +++ b/docs/docker-compose-usage-and-troubleshooting.md @@ -306,10 +306,13 @@ Note: if you accidentally create a typo folder like `dashbpards`, Grafana provis docker compose up -d --force-recreate ``` 3. Verify logs and health endpoints. -4. Run project verification scripts (example): +4. Run project verification scripts (examples): ```bash ./scripts/run-phase8-verification.sh + ./scripts/run-phase9-verification.sh ``` + Phase 9 also benefits from the MinIO client (`mc`) and DuckDB CLI for object and Parquet checks. Install without sudo — see `docs/plans/phase-9-plan.md`. + This avoids unnecessary full resets and speeds up local development. diff --git a/scripts/run-phase9-verification.sh b/scripts/run-phase9-verification.sh new file mode 100755 index 0000000..08976c6 --- /dev/null +++ b/scripts/run-phase9-verification.sh @@ -0,0 +1,173 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ROOT_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)" +COMPOSE_FILE="${COMPOSE_FILE:-${ROOT_DIR}/docker-compose.yml}" + +TEST_PROJECT="${TEST_PROJECT:-${ROOT_DIR}/VigilCareClinicalAPI.Tests/VigilCareClinicalAPI.Tests.csproj}" +TEST_FILTER="${TEST_FILTER:-FullyQualifiedName~DataLakePhase9Tests}" +BASE_URL="${BASE_URL:-http://localhost:5270}" +MINIO_ALIAS="${MINIO_ALIAS:-localvc}" +MINIO_URL="${MINIO_URL:-http://localhost:9005}" +MINIO_BUCKET="${MINIO_BUCKET:-vigilcare}" +MINIO_ACCESS_KEY="${MINIO_ACCESS_KEY:-minioadmin}" +MINIO_SECRET_KEY="${MINIO_SECRET_KEY:-minioadmin}" +KAFKA_GROUP="${KAFKA_GROUP:-data-lake-writer}" +KAFKA_SERVICE="${KAFKA_SERVICE:-kafka}" +KAFKA_BOOTSTRAP="${KAFKA_BOOTSTRAP:-localhost:9092}" + +need() { + command -v "$1" >/dev/null 2>&1 || { + echo "Missing dependency: $1" + exit 1 + } +} + +optional() { + command -v "$1" >/dev/null 2>&1 +} + +# Prefer MC_BIN, then user-local install, then first executable on PATH. +resolve_mc() { + if [[ -n "${MC_BIN:-}" && -x "${MC_BIN}" ]]; then + echo "${MC_BIN}" + return 0 + fi + local candidate="${HOME}/.local/bin/mc" + if [[ -x "${candidate}" ]]; then + echo "${candidate}" + return 0 + fi + if command -v mc >/dev/null 2>&1; then + local path_mc + path_mc="$(command -v mc)" + if [[ -x "${path_mc}" ]]; then + echo "${path_mc}" + return 0 + fi + fi + return 1 +} + +# Prefer DUCKDB_BIN, then official install location, then PATH. +resolve_duckdb() { + if [[ -n "${DUCKDB_BIN:-}" && -x "${DUCKDB_BIN}" ]]; then + echo "${DUCKDB_BIN}" + return 0 + fi + local candidate="${HOME}/.duckdb/cli/latest/duckdb" + if [[ -x "${candidate}" ]]; then + echo "${candidate}" + return 0 + fi + if command -v duckdb >/dev/null 2>&1; then + local path_duckdb + path_duckdb="$(command -v duckdb)" + if [[ -x "${path_duckdb}" ]]; then + echo "${path_duckdb}" + return 0 + fi + fi + return 1 +} + +need curl +need dotnet +need docker + +echo "Phase 9 verification starting..." +echo "Repo root: ${ROOT_DIR}" + +echo "[1/6] Preflight API" +api_status="$(curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/metrics" || true)" +[[ "${api_status}" == "200" ]] || { + echo "API /metrics is not reachable at ${BASE_URL} (status=${api_status})" + exit 1 +} + +echo "[2/6] Run Phase 9 integration tests" +dotnet test "${TEST_PROJECT}" --filter "${TEST_FILTER}" + +echo "[3/6] Verify Kafka consumer group appears" +group_list="$(docker compose -f "${COMPOSE_FILE}" exec -T "${KAFKA_SERVICE}" \ + /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server "${KAFKA_BOOTSTRAP}" --list)" +if ! grep -q "^${KAFKA_GROUP}$" <<< "${group_list}"; then + echo "Consumer group '${KAFKA_GROUP}' not found." + echo "Known groups:" + echo "${group_list}" + exit 1 +fi + +echo "[4/6] Describe Kafka consumer group offsets" +docker compose -f "${COMPOSE_FILE}" exec -T "${KAFKA_SERVICE}" \ + /opt/kafka/bin/kafka-consumer-groups.sh \ + --bootstrap-server "${KAFKA_BOOTSTRAP}" \ + --group "${KAFKA_GROUP}" \ + --describe + +MC="$(resolve_mc || true)" +if [[ -z "${MC}" ]]; then + echo "[5/6] Skipping MinIO checks: 'mc' not installed." + echo "Install without sudo:" + echo " mkdir -p ~/.local/bin" + echo " curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o ~/.local/bin/mc" + echo " chmod +x ~/.local/bin/mc" + echo " export PATH=\"\$HOME/.local/bin:\$PATH\"" + echo + echo "Phase 9 verification completed (tests + Kafka checks)." + exit 0 +fi + +echo "[5/6] Verify Parquet objects exist in MinIO (using ${MC})" +"${MC}" alias set "${MINIO_ALIAS}" "${MINIO_URL}" "${MINIO_ACCESS_KEY}" "${MINIO_SECRET_KEY}" >/dev/null + +check_prefix() { + local prefix="$1" + local count + count="$("${MC}" ls --recursive "${MINIO_ALIAS}/${MINIO_BUCKET}/${prefix}" | awk 'END{print NR+0}')" + if [[ "${count}" -le 0 ]]; then + echo "No objects found under ${prefix}" + exit 1 + fi + echo "Found ${count} object(s) under ${prefix}" +} + +check_prefix "observations/" +check_prefix "alerts/" +check_prefix "encounters/" + +DUCKDB="$(resolve_duckdb || true)" +if [[ -z "${DUCKDB}" ]]; then + echo "[6/6] Skipping DuckDB schema check: 'duckdb' not installed." + echo "Install without sudo: curl https://install.duckdb.org | sh" + echo "Then add to PATH: export PATH=\"\$HOME/.duckdb/cli/latest:\$PATH\"" + echo + echo "Phase 9 verification completed (tests + Kafka + MinIO checks)." + exit 0 +fi + +echo "[6/6] Verify Parquet schema with DuckDB (using ${DUCKDB})" +first_obs_rel="$("${MC}" ls --recursive "${MINIO_ALIAS}/${MINIO_BUCKET}/observations/" | awk '/STANDARD/{print $NF; exit}')" +[[ -n "${first_obs_rel}" ]] || { + echo "Could not resolve an observations Parquet key." + exit 1 +} +first_obs_key="observations/${first_obs_rel}" + +tmp_parquet="$(mktemp /tmp/vc-phase9-obs-XXXXXX.parquet)" +trap 'rm -f "${tmp_parquet}"' EXIT +"${MC}" cp "${MINIO_ALIAS}/${MINIO_BUCKET}/${first_obs_key}" "${tmp_parquet}" >/dev/null + +"${DUCKDB}" -csv -c "DESCRIBE SELECT * FROM '${tmp_parquet}';" | tee /tmp/vc-phase9-duckdb-schema.txt >/dev/null + +for col in observation_id encounter_id kafka_partition kafka_offset; do + if ! grep -qE "^${col}," /tmp/vc-phase9-duckdb-schema.txt; then + echo "Expected column '${col}' not found in Parquet schema." + exit 1 + fi +done + +echo +echo "Phase 9 verification checks passed."