diff --git a/CHANGELOG.md b/CHANGELOG.md index 87a17492f..c578f0a0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,14 @@ history of this codebase and was never published. This release is `[0.5.0]`. ### Added +- Bulk-ingest preflight (#278): `POST /curation/projects/{project}/ingest/bulk/preview`, + `scripts/curation/bulk_ingest.py --preview` and a Preview button on the Cropwright Ingest page report, + without ingesting anything, a folder's, manifest's, S3 prefix's or registered source's file count, total + bytes, extension breakdown, a header-sampled megapixel histogram, unreadable directories, symlinks and + missing manifest entries, and an estimate of the ingest time and the OpenSearch footprint. The estimate + uses only figures measured in `docs/PERFORMANCE.md`, served (with the section each came from) in + `GET .../ingest/bulk/config` under `preview`. The walk is capped (200,000 files, 60 seconds) and answers + `partial: true` with the reason, stops when the client disconnects, and refuses an unreadable root. - `docs/design/engineering_standards.md`: the portable standards and lessons from the repo cleanup (principles, the lint, type and test config that worked, every guard with its baseline format, `--tighten` workflow and plant-a-violation proof, an ordered port checklist for a sister repo, @@ -408,6 +416,22 @@ history of this codebase and was never published. This release is `[0.5.0]`. ### Fixed +- Monitoring alerts (#290): the Grafana rules **Triton Model Not Ready** (`nv_model_ready_state`), + **API High Error Rate** (`http_requests_total`) and **OpenSearch Cluster Unhealthy** + (`opensearch_cluster_status`) queried series nothing exports and could never fire; the first and last are + removed (no exporter has the data) and the error rate reads `http_request_duration_seconds_count`. + **High GPU Temperature** read the missing `nv_gpu_temperature` and now reads DCGM + `DCGM_FI_DEV_GPU_TEMP`. `tests/test_monitoring_metrics.py` now checks the Grafana rules against the + recorded exported names, not only the dashboards and Prometheus rules. The alert folder `Visual AI Alerts` + is renamed `OpenProcessor Alerts` (an existing Grafana keeps the old, now empty, folder until you delete + it), and the two contact points are renamed `unconfigured-default` and `unconfigured-critical` because + they post to a localhost address nothing listens on: nobody is notified until you configure a receiver + (`monitoring/README.md`, `docs-site/docs/operations/monitoring.mdx`). +- A bulk-ingest folder walk that was stopped (a cancelled run, a closed walk) kept listing the rest of the + tree and waited for a stuck directory listing before returning; it now stops at once. +- Starting a bulk ingest on a folder the API cannot read is refused (`bulk_ingest_invalid_source`, "not + readable") instead of completing with zero images discovered. + - Visual-search ingest (#318): a face response with fewer landmark entries than faces no longer raises `IndexError` and fails the whole batch. Only the affected image is refused, with the reason `face_landmarks_length_mismatch`; `detect_faces`, `recognize_faces` and the single-image face paths diff --git a/contracts/openapi/curation.json b/contracts/openapi/curation.json index 1d2b23e0e..b9b5f5314 100644 --- a/contracts/openapi/curation.json +++ b/contracts/openapi/curation.json @@ -2460,6 +2460,9 @@ "title": "Max Concurrency", "type": "integer" }, + "preview": { + "$ref": "#/components/schemas/BulkPreviewConfig" + }, "source_roots": { "items": { "type": "string" @@ -2474,7 +2477,8 @@ "default_batch_size", "default_concurrency", "max_batch_size", - "max_concurrency" + "max_concurrency", + "preview" ], "title": "BulkIngestConfig", "type": "object" @@ -2910,6 +2914,252 @@ "title": "BulkOptionsWire", "type": "object" }, + "BulkPreview": { + "properties": { + "by_extension": { + "additionalProperties": { + "$ref": "#/components/schemas/PreviewExtensionWire" + }, + "title": "By Extension", + "type": "object" + }, + "estimate": { + "$ref": "#/components/schemas/PreviewEstimateWire" + }, + "extensions": { + "items": { + "type": "string" + }, + "title": "Extensions", + "type": "array" + }, + "files": { + "title": "Files", + "type": "integer" + }, + "gpu_decode": { + "title": "Gpu Decode", + "type": "boolean" + }, + "kind": { + "enum": [ + "folder", + "manifest", + "s3" + ], + "title": "Kind", + "type": "string" + }, + "megapixels": { + "$ref": "#/components/schemas/PreviewMegapixelsWire" + }, + "missing": { + "title": "Missing", + "type": "integer" + }, + "outside_roots": { + "title": "Outside Roots", + "type": "integer" + }, + "partial": { + "title": "Partial", + "type": "boolean" + }, + "project": { + "title": "Project", + "type": "string" + }, + "stop_reason": { + "anyOf": [ + { + "enum": [ + "file_cap", + "time_cap", + "cancelled" + ], + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Stop Reason" + }, + "symlinks_skipped": { + "title": "Symlinks Skipped", + "type": "integer" + }, + "target": { + "title": "Target", + "type": "string" + }, + "total_bytes": { + "title": "Total Bytes", + "type": "integer" + }, + "unreadable_dirs": { + "title": "Unreadable Dirs", + "type": "integer" + }, + "unreadable_files": { + "title": "Unreadable Files", + "type": "integer" + }, + "unreadable_samples": { + "items": { + "type": "string" + }, + "title": "Unreadable Samples", + "type": "array" + } + }, + "required": [ + "project", + "kind", + "target", + "extensions", + "files", + "total_bytes", + "by_extension", + "megapixels", + "estimate", + "partial", + "unreadable_dirs", + "unreadable_samples", + "symlinks_skipped", + "missing", + "outside_roots", + "unreadable_files", + "gpu_decode" + ], + "title": "BulkPreview", + "type": "object" + }, + "BulkPreviewConfig": { + "description": "The bounds of a preview and the measured figures its estimate uses, all from\n``docs/PERFORMANCE.md`` (the section is named per figure).", + "properties": { + "embeddings_per_image": { + "$ref": "#/components/schemas/PreviewMeasuredWire" + }, + "gpu_decode": { + "title": "Gpu Decode", + "type": "boolean" + }, + "gpu_min_megapixels": { + "title": "Gpu Min Megapixels", + "type": "number" + }, + "header_sample": { + "title": "Header Sample", + "type": "integer" + }, + "max_files": { + "title": "Max Files", + "type": "integer" + }, + "max_seconds": { + "title": "Max Seconds", + "type": "number" + }, + "opensearch_bytes_per_image": { + "$ref": "#/components/schemas/PreviewMeasuredWire" + }, + "rates": { + "items": { + "$ref": "#/components/schemas/PreviewRateWire" + }, + "title": "Rates", + "type": "array" + }, + "source_document": { + "title": "Source Document", + "type": "string" + } + }, + "required": [ + "max_files", + "max_seconds", + "header_sample", + "gpu_decode", + "gpu_min_megapixels", + "rates", + "embeddings_per_image", + "opensearch_bytes_per_image", + "source_document" + ], + "title": "BulkPreviewConfig", + "type": "object" + }, + "BulkPreviewRequest": { + "additionalProperties": false, + "description": "Exactly one of ``folder``, ``manifest``, ``s3`` and ``source_id``, as for a start.\nNothing is ingested and nothing is written.", + "properties": { + "extensions": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "maxItems": 32, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Extensions" + }, + "folder": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Folder" + }, + "manifest": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Manifest" + }, + "s3": { + "anyOf": [ + { + "$ref": "#/components/schemas/S3Location" + }, + { + "type": "null" + } + ] + }, + "source_id": { + "anyOf": [ + { + "maxLength": 64, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Id" + } + }, + "title": "BulkPreviewRequest", + "type": "object" + }, "BulkSourceWire": { "properties": { "extensions": { @@ -15482,6 +15732,164 @@ "title": "PresetsResponse", "type": "object" }, + "PreviewEstimateWire": { + "properties": { + "basis": { + "enum": [ + "megapixel_sample", + "rate_range" + ], + "title": "Basis", + "type": "string" + }, + "duration_s_max": { + "title": "Duration S Max", + "type": "number" + }, + "duration_s_min": { + "title": "Duration S Min", + "type": "number" + }, + "opensearch_bytes": { + "title": "Opensearch Bytes", + "type": "integer" + }, + "vectors": { + "title": "Vectors", + "type": "number" + } + }, + "required": [ + "basis", + "duration_s_min", + "duration_s_max", + "vectors", + "opensearch_bytes" + ], + "title": "PreviewEstimateWire", + "type": "object" + }, + "PreviewExtensionWire": { + "properties": { + "bytes": { + "title": "Bytes", + "type": "integer" + }, + "files": { + "title": "Files", + "type": "integer" + } + }, + "required": [ + "files", + "bytes" + ], + "title": "PreviewExtensionWire", + "type": "object" + }, + "PreviewMeasuredWire": { + "properties": { + "measured_on": { + "title": "Measured On", + "type": "string" + }, + "source_section": { + "title": "Source Section", + "type": "string" + }, + "unit": { + "title": "Unit", + "type": "string" + }, + "value": { + "title": "Value", + "type": "number" + } + }, + "required": [ + "value", + "unit", + "source_section", + "measured_on" + ], + "title": "PreviewMeasuredWire", + "type": "object" + }, + "PreviewMegapixelsWire": { + "properties": { + "gate_megapixels": { + "title": "Gate Megapixels", + "type": "number" + }, + "histogram": { + "additionalProperties": { + "type": "integer" + }, + "title": "Histogram", + "type": "object" + }, + "sampled": { + "title": "Sampled", + "type": "integer" + }, + "share_at_or_above_gate": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Share At Or Above Gate" + }, + "unreadable": { + "title": "Unreadable", + "type": "integer" + } + }, + "required": [ + "sampled", + "unreadable", + "histogram", + "gate_megapixels" + ], + "title": "PreviewMegapixelsWire", + "type": "object" + }, + "PreviewRateWire": { + "properties": { + "images_per_s": { + "title": "Images Per S", + "type": "number" + }, + "measured_on": { + "title": "Measured On", + "type": "string" + }, + "route": { + "enum": [ + "cpu_small", + "cpu_large", + "gpu_large" + ], + "title": "Route", + "type": "string" + }, + "source_section": { + "title": "Source Section", + "type": "string" + } + }, + "required": [ + "route", + "images_per_s", + "source_section", + "measured_on" + ], + "title": "PreviewRateWire", + "type": "object" + }, "ProbeRunRequest": { "properties": { "architecture": { @@ -37686,6 +38094,59 @@ ] } }, + "/curation/projects/{project}/ingest/bulk/preview": { + "post": { + "description": "Count the files, bytes and extensions of a source, sample image sizes, and estimate\nduration and index footprint from measured figures (``GET /ingest/bulk/config``).\nCapped by file count and seconds (``partial: true`` then); a client disconnect stops\nthe walk.", + "operationId": "preview_bulk_ingest_curation_projects__project__ingest_bulk_preview_post", + "parameters": [ + { + "in": "path", + "name": "project", + "required": true, + "schema": { + "title": "Project", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BulkPreviewRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BulkPreview" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Preview Bulk Ingest", + "tags": [ + "Curation" + ] + } + }, "/curation/projects/{project}/ingest/bulk/{run_id}": { "get": { "operationId": "get_bulk_ingest_curation_projects__project__ingest_bulk__run_id__get", diff --git a/docs-site/docs/guides/ingest-large-folder.mdx b/docs-site/docs/guides/ingest-large-folder.mdx index 4b4813c9b..794406582 100644 --- a/docs-site/docs/guides/ingest-large-folder.mdx +++ b/docs-site/docs/guides/ingest-large-folder.mdx @@ -189,9 +189,61 @@ curl -s http://localhost:4603/curation/projects/nas-photos/ingest/policy Run a small slice first. It shows whether the mount, the detector and the policy do what you expect, and it gives you a measured speed for your hardware. -There is no dry-run or file-count option today: `bulk_ingest.py` has no such flag and -the start request accepts only the fields in the flags table below. The closest you -can get is to limit the input yourself. +### Preview before you ingest + +Before any run, ask the API what the folder or manifest holds. A preview walks the source +read-only and ingests nothing: it returns the file count, total bytes, the breakdown by +extension, a sample of image sizes (headers only, no pixel decoded), the directories and +symlinks it could not follow, and an estimate of the ingest time and the OpenSearch index +size. In Cropwright press **Preview** next to **Start bulk ingest** on the Ingest page; on the +command line add `--preview` to the same flags: + +```bash +docker compose -p exec api python scripts/curation/bulk_ingest.py \ + --project nas-photos --folder /data/source/2024-03-sat \ + --api-base http://localhost:8000/curation --preview +``` + +The route is `POST /curation/projects/{project}/ingest/bulk/preview` (the same single source +field as a start: `folder`, `manifest`, `s3` or `source_id`). An unreadable folder, a path +outside the source roots or an S3 prefix outside `OP_S3_ALLOWED` is refused; nothing is +guessed. Output of the command line renderer on the 10-image public COCO sample (this is the +service run on that folder in a script, not against a deployed stack): + +```text +folder /data/source/coco_sample + files: 10, 1.7 MiB + .jpg: 10 files, 1.7 MiB + image sizes (header sample of 10): <1 MP: 10; 0% at or above 3 MP + estimated ingest time: about 0.3 s (megapixel_sample) + estimated index size: 637.7 KiB, about 74 vectors + basis: measured figures in docs/PERFORMANCE.md, not a promise +``` + +What the numbers rest on: + +- **The walk is bounded.** It stops at 200,000 files or 60 seconds, whichever comes first + (`GET .../ingest/bulk/config` serves the limits under `preview`). A source bigger than that + answers `partial: true` with the reason, and every figure is then a lower bound. Closing the + connection (Ctrl-C on the command line, or pressing the button again) stops the walk. +- **The estimate uses only measured figures** from [`docs/PERFORMANCE.md`](https://github.com/davidamacey/OpenProcessor/blob/main/docs/PERFORMANCE.md), served in the same config with the section each came from: ingest + speed by decode route (29.0 images/s for small frames, 27.0 for 12 to 24 MP photos on the CPU + path, 35.0 with the GPU decode) from "Decode once and the GPU decode", 7.4 embeddings per + image, and 65,298 bytes of index per image from "Storage per image and per vector". They were + measured on one RTX A6000 with public COCO and Commons photos, so another card or detector + profile will differ: treat the time as an order of magnitude, not a promise. When the image + sizes are unknown (an S3 prefix is only listed), the time is the range from the slowest to the + fastest measured route. +- **Image sizes** come from the headers of a fixed-size sample (300 files), shown as a + histogram in megapixels and as the share at or above the GPU-decode gate + (`OP_INGEST_GPU_MIN_MEGAPIXELS`, 3.0). +- **Skipped entries are counted, not hidden:** unreadable directories (with the first few + paths), symlinks (not followed), manifest lines that do not exist or lie outside the source + roots. + +### Run a small slice + +Then run a slice to get a measured speed for your own hardware. Limit the input yourself: The simplest trial is a small sub-folder: @@ -287,6 +339,7 @@ the run's outcome. |---|---| | `--folder PATH` | Walk this directory (recursive). One of `--folder`, `--manifest`, `--status`, `--cancel` is required. | | `--manifest PATH` | Text file of image paths. | +| `--preview` | Do not ingest: print the file count, bytes, extensions, image sizes, skipped entries and the time and index-size estimate for the source, then exit. Works with `--folder`, `--manifest`, `--s3` and `--source-id`. | | `--status RUN_ID` | Print one run's progress and exit. | | `--cancel RUN_ID` | Ask a live run to stop and exit. | | `--project SLUG` | Project to bind. Default `$OP_CURATION_PROJECT`, else `default`. | diff --git a/docs-site/docs/operations/monitoring.mdx b/docs-site/docs/operations/monitoring.mdx index 5ba9eb249..528cf2f2f 100644 --- a/docs-site/docs/operations/monitoring.mdx +++ b/docs-site/docs/operations/monitoring.mdx @@ -74,16 +74,19 @@ requests) and `TritonServerDown`. They show on Prometheus's Alerts page (`http://localhost:4604/alerts`). `monitoring/grafana-alerting.yml` is provisioned into Grafana as a second, -separate rule set: API error rate and slow responses, Triton down, latency -and queue, GPU utilization, memory and temperature, host memory, CPU and -load, and OpenSearch cluster status. They show at +separate rule set under the `OpenProcessor Alerts` folder: API error rate and +slow responses, Triton down, latency and queue, GPU utilization, memory and +temperature (DCGM), and host memory, CPU and load. They show at `http://localhost:4605/alerting/list`. -Neither set notifies anyone out of the box. Prometheus has no Alertmanager. -Grafana's two contact points are webhooks to `http://localhost:9999/...`, -which is the Grafana container itself, so nothing receives them. Point the -contact points in `grafana-alerting.yml` at your own webhook, Slack or email -and restart Grafana. +**Neither set notifies anyone out of the box.** Prometheus has no Alertmanager. +Grafana's two contact points are named `unconfigured-default` and +`unconfigured-critical` and post to `http://localhost:9999/...`, which is the +Grafana container itself, where nothing listens. A firing alert shows in +Grafana and its delivery fails with a logged error; nobody is told. To be +notified, replace the receivers in `grafana-alerting.yml` with your own +webhook, Slack or email and restart Grafana (the file is provisioned +read-only, so the Grafana UI cannot change them). ## Grafana dashboards @@ -125,15 +128,14 @@ Panels only move while the API is serving requests; an idle stack shows flat lines. To add a dashboard, drop its JSON into `monitoring/dashboards/` and restart Grafana. -:::caution[Two Grafana alert rules can never fire] +:::note[No rule for model readiness or OpenSearch status] The shipped Triton does not export `nv_model_ready_state` or -`nv_gpu_temperature`. The Prometheus `ModelNotReady` rule and the Triton -dashboard's Model Ready panel were removed for that reason, and the -dashboard's GPU Temperature panel reads DCGM (`DCGM_FI_DEV_GPU_TEMP`). The -provisioned Grafana rules **Triton Model Not Ready** and **High GPU -Temperature** still query the missing metrics, so they stay silent. Use -`GET /models/` (below) for model readiness and the GPUs dashboard (DCGM) -for temperature. +`nv_gpu_temperature`, and no OpenSearch exporter ships with the stack, so there +is no rule on per-model readiness or on OpenSearch cluster status. Use +`GET /models/` (below) for model readiness and the GPUs dashboard (DCGM) for +temperature; the Grafana temperature rule reads `DCGM_FI_DEV_GPU_TEMP`. An +offline test checks that every metric an alert rule or dashboard panel reads +is one the stack exports, so a rule on a missing series cannot come back. ::: ## Logs with Loki diff --git a/docs/design/curation_api_contract.md b/docs/design/curation_api_contract.md index 8b9976c33..9e8d1b5c9 100644 --- a/docs/design/curation_api_contract.md +++ b/docs/design/curation_api_contract.md @@ -1456,7 +1456,8 @@ can fuse. The default pool is fresh and class-bucketed unvalidated items. | GET | `/ingest/region_drain` | | drain state | | POST | `/ingest/bulk` | `BulkIngestRequest`: exactly one of `folder`, `manifest`; `source`, `extensions`, `batch_size`, `concurrency` (`extra='forbid'`) | `BulkIngestRun` (202) | | GET | `/ingest/bulk` | query `page`, `page_size` | `BulkIngestList`, newest first | -| GET | `/ingest/bulk/config` | | `BulkIngestConfig` | +| GET | `/ingest/bulk/config` | | `BulkIngestConfig` (includes `preview`: the caps and the measured figures the estimate uses) | +| POST | `/ingest/bulk/preview` | `BulkPreviewRequest`: exactly one of `folder`, `manifest`, `s3`, `source_id`; `extensions` (`extra='forbid'`) | `BulkPreview` (200): read-only; counts, bytes, extensions, size sample, estimate; `partial` with `stop_reason` when a cap or a client disconnect ended the walk | | GET | `/ingest/bulk/{run_id}` | | `BulkIngestRun` | | GET | `/ingest/bulk/{run_id}/failures` | query `offset`, `limit` | `BulkIngestFailurePage` | | POST | `/ingest/bulk/{run_id}/cancel` | | `BulkIngestRun` | diff --git a/frontend/CHANGELOG.md b/frontend/CHANGELOG.md index f06ca715e..60218dd09 100644 --- a/frontend/CHANGELOG.md +++ b/frontend/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- The Bulk ingest panel has a **Preview** button: it asks `POST /ingest/bulk/preview` what the folder or + manifest holds and shows the served file count, bytes, extensions, image-size sample, skipped entries + and the estimated time and index size with the measured figures behind them. A capped scan is marked + "partial" (every figure a lower bound); pressing the button again while it scans cancels it. - `/ingest` has a "Bulk ingest (folder or manifest)" panel, shown with the server-path ingest it sits beside: start a run over a folder or a manifest the backend has mounted, follow served progress (counts, speed, time left, seconds per stage), cancel, resume, and read the failed images with their reasons. Everything shown and every diff --git a/frontend/e2e/stubbed/test_bulk_ingest.py b/frontend/e2e/stubbed/test_bulk_ingest.py index 55d085a05..73df932b5 100644 --- a/frontend/e2e/stubbed/test_bulk_ingest.py +++ b/frontend/e2e/stubbed/test_bulk_ingest.py @@ -219,3 +219,61 @@ def test_no_horizontal_overflow_at_430px_with_a_live_run(stub, page, app_url): "() => ({sw: document.documentElement.scrollWidth, iw: window.innerWidth})" ) assert width["sw"] <= width["iw"], width + + +PREVIEW = { + "project": "default", "kind": "folder", "target": "/data/archive/coco", + "extensions": [".jpg", ".jpeg", ".png"], "files": 2000, "total_bytes": 836000000, + "by_extension": {".jpg": {"files": 1990, "bytes": 830000000}, ".png": {"files": 10, "bytes": 6000000}}, + "megapixels": {"sampled": 300, "unreadable": 0, "histogram": {"<1": 300}, + "gate_megapixels": 3.0, "share_at_or_above_gate": 0.0}, + "estimate": {"basis": "megapixel_sample", "duration_s_min": 69.0, "duration_s_max": 69.0, + "vectors": 14800.0, "opensearch_bytes": 130596000}, + "partial": False, "stop_reason": None, "unreadable_dirs": 0, "unreadable_samples": [], + "symlinks_skipped": 0, "missing": 0, "outside_roots": 0, "unreadable_files": 0, "gpu_decode": True, +} + + +def test_preview_shows_the_served_counts_and_estimate_and_starts_nothing(stub, page, app_url): + asked = [] + + def preview(request, _m): + asked.append(json.loads(request.post_data)) + return (200, PREVIEW) + + stub.on("POST", r"/ingest/bulk/preview$", preview) + _open(stub, page, app_url) + expect(page.get_by_test_id("bulk-preview-button")).to_be_disabled() + page.get_by_test_id("bulk-path").fill("/data/archive/coco") + page.get_by_test_id("bulk-preview-button").click() + box = page.get_by_test_id("bulk-preview") + box.wait_for(timeout=ACTION_TIMEOUT_MS) + assert asked == [{"folder": "/data/archive/coco"}] + expect(page.get_by_test_id("bulk-preview-total")).to_contain_text("2000 files") + expect(page.get_by_test_id("bulk-preview-estimate")).to_contain_text("1m 09s") + expect(page.get_by_test_id("bulk-preview-partial")).to_have_count(0) + page.get_by_text("Basis: measured figures from docs/PERFORMANCE.md").click() + expect(box).to_contain_text("Decode once and the GPU decode (#217)") + assert not [c for c in stub.calls if c[0] == "POST" and c[1].endswith("/ingest/bulk")] + + page.set_viewport_size(NARROW) + overflow = page.evaluate("document.documentElement.scrollWidth > window.innerWidth") + assert not overflow + + +def test_preview_says_partial_and_surfaces_a_refusal(stub, page, app_url): + answers = [ + (200, {**PREVIEW, "partial": True, "stop_reason": "file_cap", "files": 200000, + "unreadable_dirs": 1, "unreadable_samples": ["/data/archive/coco/locked"]}), + (422, {"detail": {"error": "bulk_ingest_invalid_source", "message": "/data/nope: not a directory"}}), + ] + stub.on("POST", r"/ingest/bulk/preview$", lambda _r, _m: answers.pop(0)) + _open(stub, page, app_url) + page.get_by_test_id("bulk-path").fill("/data/archive/coco") + page.get_by_test_id("bulk-preview-button").click() + expect(page.get_by_test_id("bulk-preview-partial")).to_contain_text("at least", timeout=ACTION_TIMEOUT_MS) + expect(page.get_by_test_id("bulk-preview-skipped")).to_contain_text("/data/archive/coco/locked") + page.get_by_test_id("bulk-path").fill("/data/nope") + page.get_by_test_id("bulk-preview-button").click() + expect(page.get_by_test_id("bulk-preview-error")).to_contain_text("not a directory", timeout=ACTION_TIMEOUT_MS) + expect(page.get_by_test_id("bulk-preview")).to_have_count(0) diff --git a/frontend/e2e/stubbed/test_ingest.py b/frontend/e2e/stubbed/test_ingest.py index 1673c3b3d..e77a022b9 100644 --- a/frontend/e2e/stubbed/test_ingest.py +++ b/frontend/e2e/stubbed/test_ingest.py @@ -39,6 +39,23 @@ "default_concurrency": 8, "max_batch_size": 64, "max_concurrency": 32, + "preview": { + "max_files": 200000, "max_seconds": 60.0, "header_sample": 300, + "gpu_decode": True, "gpu_min_megapixels": 3.0, + "rates": [ + {"route": "cpu_small", "images_per_s": 29.0, "measured_on": "COCO frames (0.3 MP), CPU decode", + "source_section": "Decode once and the GPU decode (#217)"}, + {"route": "cpu_large", "images_per_s": 27.0, "measured_on": "Commons photos (12 to 24 MP), CPU decode", + "source_section": "Decode once and the GPU decode (#217)"}, + {"route": "gpu_large", "images_per_s": 35.0, "measured_on": "Commons photos (12 to 24 MP), GPU decode", + "source_section": "Decode once and the GPU decode (#217)"}, + ], + "embeddings_per_image": {"value": 7.4, "unit": "embeddings per image", "measured_on": "COCO, policy all", + "source_section": "v0.5.0 baseline (set A, public COCO, 2,000 images)"}, + "opensearch_bytes_per_image": {"value": 65298, "unit": "bytes per image", "measured_on": "COCO, policy all", + "source_section": "Storage per image and per vector"}, + "source_document": "docs/PERFORMANCE.md", + }, } diff --git a/frontend/src/lib/api_bulkIngest.ts b/frontend/src/lib/api_bulkIngest.ts index 04445b87a..62633069a 100644 --- a/frontend/src/lib/api_bulkIngest.ts +++ b/frontend/src/lib/api_bulkIngest.ts @@ -11,6 +11,8 @@ import type { BulkIngestList, BulkIngestRequest, BulkIngestRun, + BulkPreview, + BulkPreviewRequest, } from '$lib/types_bulkIngest'; export function getBulkIngestConfig(signal?: AbortSignal): Promise { @@ -30,6 +32,19 @@ export function startBulkIngest( ); } +/** `POST /ingest/bulk/preview`: counts, sizes and an estimate for a source, no ingestion. + * Aborting the request stops the server-side walk. */ +export function previewBulkIngest( + body: BulkPreviewRequest, + signal?: AbortSignal, +): Promise { + return apiFetch( + `${scoped()}/ingest/bulk/preview`, + { method: 'POST', body: JSON.stringify(body) }, + signal, + ); +} + export function listBulkIngests( params: { page?: number; page_size?: number } = {}, signal?: AbortSignal, diff --git a/frontend/src/lib/components/ingest/BulkIngestPanel.svelte b/frontend/src/lib/components/ingest/BulkIngestPanel.svelte index 533082a35..a0d388a39 100644 --- a/frontend/src/lib/components/ingest/BulkIngestPanel.svelte +++ b/frontend/src/lib/components/ingest/BulkIngestPanel.svelte @@ -12,6 +12,7 @@ } from '$lib/ingest/bulkIngestController.svelte'; import { formatDuration } from '$lib/format/duration'; import { headlineRate, progressPercent, stageRows } from '$lib/ingest/bulkIngestFormat'; + import BulkPreviewResult from './BulkPreviewResult.svelte'; import type { BulkIngestStatus } from '$lib/types_bulkIngest'; interface Props { @@ -116,6 +117,16 @@ > {controller.busy && !controller.live ? 'Starting…' : 'Start bulk ingest'} +
Advanced @@ -157,6 +168,15 @@
+ {#if controller.previewError} +

+ {controller.previewError} +

+ {/if} + {#if controller.preview} + + {/if} + {#if controller.actionError}

{controller.actionError} diff --git a/frontend/src/lib/components/ingest/BulkIngestPanel.test.ts b/frontend/src/lib/components/ingest/BulkIngestPanel.test.ts index 5ae59afa2..2c2689920 100644 --- a/frontend/src/lib/components/ingest/BulkIngestPanel.test.ts +++ b/frontend/src/lib/components/ingest/BulkIngestPanel.test.ts @@ -8,6 +8,7 @@ import BulkIngestPanel from './BulkIngestPanel.svelte'; import { BulkIngest } from '$lib/ingest/bulkIngestController.svelte'; import { configFixture, + previewFixture, progressFixture, runFixture, } from '$lib/test/fixtures/bulkIngest'; @@ -38,6 +39,7 @@ function controller(run = runFixture(), over: Record = {}) { .fn() .mockResolvedValue({ items: [], total: 0, offset: 0, limit: 20 }), cancelBulkIngest: vi.fn().mockResolvedValue(run), + previewBulkIngest: vi.fn().mockResolvedValue(previewFixture()), ...over, }; return { ctl: new BulkIngest(deps), deps }; @@ -185,3 +187,84 @@ describe('BulkIngestPanel', () => { await vi.waitFor(() => expect(text()).toContain('Could not load bulk ingest: boom')); }); }); + +describe('BulkIngestPanel preview', () => { + async function previewed(over = {}, ctlOver = {}) { + const { ctl, deps } = controller( + runFixture({ status: 'completed', poll_after_s: null }), + { previewBulkIngest: vi.fn().mockResolvedValue(previewFixture(over)), ...ctlOver }, + ); + await render(ctl); + ctl.form.path = '/data/source/coco'; + flushSync(); + (byId('bulk-preview-button') as HTMLButtonElement).click(); + await vi.waitFor(() => expect(byId('bulk-preview')).toBeTruthy()); + return { ctl, deps }; + } + + it('is disabled until a path is typed and ingests nothing', async () => { + const { ctl, deps } = controller( + runFixture({ status: 'completed', poll_after_s: null }), + ); + await render(ctl); + expect((byId('bulk-preview-button') as HTMLButtonElement).disabled).toBe(true); + ctl.form.path = '/data/x'; + flushSync(); + expect((byId('bulk-preview-button') as HTMLButtonElement).disabled).toBe(false); + expect(deps.startBulkIngest).not.toHaveBeenCalled(); + }); + + it('renders the served counts, extensions, sizes and estimate with their basis', async () => { + await previewed(); + const t = byId('bulk-preview')?.textContent ?? ''; + expect(t).toContain('2000 files'); + expect(t).toContain('797.3 MB'); + expect(t).toContain('.jpg'); + expect(t).toContain('1990'); + expect(t).toContain('1m 09s'); + expect(t).toContain('124.5 MB'); + expect(t).toContain('14800'); + expect(t).toContain('docs/PERFORMANCE.md'); + expect(t).toContain('Decode once and the GPU decode (#217)'); + expect(byId('bulk-preview-partial')).toBeNull(); + }); + + it('says partial and lower bound when the walk was capped, and lists what was skipped', async () => { + await previewed({ + partial: true, + stop_reason: 'file_cap', + unreadable_dirs: 2, + unreadable_samples: ['/data/source/coco/locked'], + symlinks_skipped: 3, + }); + expect(byId('bulk-preview-partial')?.textContent).toContain('at least'); + const t = byId('bulk-preview')?.textContent ?? ''; + expect(t).toContain('/data/source/coco/locked'); + expect(t).toContain('3 symlinks'); + }); + + it('shows a range when the sizes are unknown', async () => { + await previewed({ + estimate: { + basis: 'rate_range', + duration_s_min: 60, + duration_s_max: 120, + vectors: 100, + opensearch_bytes: 1000, + }, + }); + expect(byId('bulk-preview')?.textContent).toContain('1m 00s to 2m 00s'); + }); + + it('shows the served refusal', async () => { + const { ctl } = controller(runFixture({ status: 'completed', poll_after_s: null }), { + previewBulkIngest: vi.fn().mockRejectedValue(new Error('not a directory')), + }); + await render(ctl); + ctl.form.path = '/nope'; + flushSync(); + (byId('bulk-preview-button') as HTMLButtonElement).click(); + await vi.waitFor(() => expect(byId('bulk-preview-error')).toBeTruthy()); + expect(byId('bulk-preview-error')?.textContent).toContain('not a directory'); + }); +}); diff --git a/frontend/src/lib/components/ingest/BulkPreviewResult.svelte b/frontend/src/lib/components/ingest/BulkPreviewResult.svelte new file mode 100644 index 000000000..b631d4c93 --- /dev/null +++ b/frontend/src/lib/components/ingest/BulkPreviewResult.svelte @@ -0,0 +1,143 @@ + + +

+

{preview.target}

+ {#if preview.partial && preview.stop_reason} +

+ Partial: the scan stopped at {STOP_TEXT[preview.stop_reason](config)}, so these are + at least the figures; the source holds more. +

+ {/if} +

+ {preview.files} files · {formatBytes(preview.total_bytes)} +

+ + + {#each extensions as [ext, totals] (ext)} + + + + + + {/each} + +
{ext}{totals.files}{formatBytes(totals.bytes)}
+ {#if preview.megapixels.sampled > 0} +

+ Image sizes (headers of {preview.megapixels.sampled} files, megapixels): + {#each sizes as [label, n], i (label)}{i > 0 ? ' · ' : ' '}{label}: {n}{/each}. + {#if preview.megapixels.share_at_or_above_gate != null} + {Math.round(preview.megapixels.share_at_or_above_gate * 100)}% at or above + {preview.megapixels.gate_megapixels} MP, the GPU-decode gate ({preview.gpu_decode + ? 'on' + : 'off'}). + {/if} + {#if preview.megapixels.unreadable > 0} + {preview.megapixels.unreadable} headers could not be read. + {/if} +

+ {:else if preview.kind === 's3'} +

Image sizes are not read for S3 objects.

+ {/if} + {#if skipped.length > 0} +
    + {#each skipped as [n, label] (label)} +
  • {n} {label}
    • + {/each} + {#each preview.unreadable_samples as path (path)} +
  • {path}
    • + {/each} +
+ {/if} +
+
+
Estimated ingest time
+
{duration}
+
+
+
Estimated index size
+
{formatBytes(est.opensearch_bytes)}
+
+
+
Estimated vectors
+
{Math.round(est.vectors)}
+
+
+
+ + Basis: measured figures from {config.source_document} + +
    + {#each config.rates as rate (rate.route)} +
  • + {rate.images_per_s.toFixed(1)} images/s, {rate.measured_on} + ({rate.source_section}) +
    • + {/each} +
  • + {config.embeddings_per_image.value} + {config.embeddings_per_image.unit} ({config.embeddings_per_image.source_section}) +
    • +
  • + {config.opensearch_bytes_per_image.value} + {config.opensearch_bytes_per_image.unit} ({config.opensearch_bytes_per_image + .source_section}) +
    • +
  • + {est.basis === 'megapixel_sample' + ? 'Time uses the sampled image sizes.' + : 'Image sizes are unknown, so the time is the span from the slowest to the fastest measured route.'} + Another host or detector profile will differ. +
    • +
+
+
diff --git a/frontend/src/lib/contract/bulkIngestContract.test.ts b/frontend/src/lib/contract/bulkIngestContract.test.ts index 554884dc1..5f6d90fa5 100644 --- a/frontend/src/lib/contract/bulkIngestContract.test.ts +++ b/frontend/src/lib/contract/bulkIngestContract.test.ts @@ -15,6 +15,14 @@ import type { BulkIngestRun, BulkIngestStatus, BulkOptions, + BulkPreview, + BulkPreviewConfig, + BulkPreviewRequest, + PreviewEstimate, + PreviewExtension, + PreviewMeasured, + PreviewMegapixels, + PreviewRate, BulkSource, } from '$lib/types_bulkIngest'; @@ -55,6 +63,7 @@ describe('bulk ingest contract', () => { 'default_concurrency', 'max_batch_size', 'max_concurrency', + 'preview', ], 'BulkIngestConfig', ); @@ -62,6 +71,67 @@ describe('bulk ingest contract', () => { pin(['source_tag', 'batch_size', 'concurrency'], 'BulkOptionsWire'); }); + it('preview request, config and answer types', () => { + pin( + ['folder', 'manifest', 's3', 'source_id', 'extensions'], + 'BulkPreviewRequest', + ); + pin( + [ + 'max_files', + 'max_seconds', + 'header_sample', + 'gpu_decode', + 'gpu_min_megapixels', + 'rates', + 'embeddings_per_image', + 'opensearch_bytes_per_image', + 'source_document', + ], + 'BulkPreviewConfig', + ); + pin( + ['route', 'images_per_s', 'source_section', 'measured_on'], + 'PreviewRateWire', + ); + pin( + ['value', 'unit', 'source_section', 'measured_on'], + 'PreviewMeasuredWire', + ); + pin( + [ + 'project', + 'kind', + 'target', + 'extensions', + 'files', + 'total_bytes', + 'by_extension', + 'megapixels', + 'estimate', + 'partial', + 'stop_reason', + 'unreadable_dirs', + 'unreadable_samples', + 'symlinks_skipped', + 'missing', + 'outside_roots', + 'unreadable_files', + 'gpu_decode', + ], + 'BulkPreview', + ); + pin(['files', 'bytes'], 'PreviewExtensionWire'); + pin( + ['sampled', 'unreadable', 'histogram', 'gate_megapixels', 'share_at_or_above_gate'], + 'PreviewMegapixelsWire', + ); + pin( + ['basis', 'duration_s_min', 'duration_s_max', 'vectors', 'opensearch_bytes'], + 'PreviewEstimateWire', + ); + }); + it('run and progress types', () => { pin( [ diff --git a/frontend/src/lib/ingest/bulkIngestController.svelte.ts b/frontend/src/lib/ingest/bulkIngestController.svelte.ts index 25c8a37e8..72a4994e3 100644 --- a/frontend/src/lib/ingest/bulkIngestController.svelte.ts +++ b/frontend/src/lib/ingest/bulkIngestController.svelte.ts @@ -12,6 +12,7 @@ import { getBulkIngestConfig, getBulkIngestFailures, listBulkIngests, + previewBulkIngest, startBulkIngest, } from '$lib/api_bulkIngest'; import type { @@ -19,6 +20,8 @@ import type { BulkIngestFailurePage, BulkIngestRequest, BulkIngestRun, + BulkPreview, + BulkPreviewRequest, BulkSourceKind, } from '$lib/types_bulkIngest'; @@ -32,6 +35,7 @@ export interface BulkDeps { getBulkIngest: typeof getBulkIngest; getBulkIngestFailures: typeof getBulkIngestFailures; cancelBulkIngest: typeof cancelBulkIngest; + previewBulkIngest: typeof previewBulkIngest; } const DEFAULT_DEPS: BulkDeps = { @@ -41,6 +45,7 @@ const DEFAULT_DEPS: BulkDeps = { getBulkIngest, getBulkIngestFailures, cancelBulkIngest, + previewBulkIngest, }; export interface BulkForm { @@ -62,6 +67,9 @@ export class BulkIngest { actionError = $state(null); loadError = $state(null); busy = $state(false); + preview = $state(null); + previewing = $state(false); + previewError = $state(null); form = $state({ kind: 'folder', @@ -75,6 +83,7 @@ export class BulkIngest { #deps: BulkDeps; #timer: ReturnType | null = null; #stopped = false; + #previewAbort: AbortController | null = null; constructor(deps: Partial = {}) { this.#deps = { ...DEFAULT_DEPS, ...deps }; @@ -84,24 +93,38 @@ export class BulkIngest { return this.run?.poll_after_s != null; } - /** The request exactly as the form states it; unset options stay unset so the - * backend's served defaults apply. */ - get request(): BulkIngestRequest { + /** The source and extensions exactly as the form states them (a preview sends + * only these). */ + get previewRequest(): BulkPreviewRequest { const f = this.form; - const body: BulkIngestRequest = { + const body: BulkPreviewRequest = { [f.kind === 'folder' ? 'folder' : 'manifest']: f.path.trim(), - source: f.source.trim() || 'bulk_ingest', }; const extensions = f.extensions .split(',') .map((e) => e.trim()) .filter((e) => e.length > 0); if (extensions.length > 0) body.extensions = extensions; + return body; + } + + /** The request exactly as the form states it; unset options stay unset so the + * backend's served defaults apply. */ + get request(): BulkIngestRequest { + const f = this.form; + const body: BulkIngestRequest = { + ...this.previewRequest, + source: f.source.trim() || 'bulk_ingest', + }; if (f.batchSize) body.batch_size = f.batchSize; if (f.concurrency) body.concurrency = f.concurrency; return body; } + get canPreview(): boolean { + return !this.previewing && this.form.path.trim().length > 0; + } + get canStart(): boolean { return !this.busy && !this.live && this.form.path.trim().length > 0; } @@ -124,6 +147,7 @@ export class BulkIngest { stop(): void { this.#stopped = true; this.#clearTimer(); + this.#previewAbort?.abort(); } async refreshRecent(): Promise { @@ -183,11 +207,39 @@ export class BulkIngest { } } + /** Ask the backend what the source holds and costs; nothing is ingested. A second + * press while one is running cancels it (the aborted request stops the walk). */ + async runPreview(): Promise { + if (this.previewing) { + this.#previewAbort?.abort(); + return false; + } + const abort = new AbortController(); + this.#previewAbort = abort; + this.previewing = true; + this.previewError = null; + this.preview = null; + try { + this.preview = await this.#deps.previewBulkIngest( + this.previewRequest, + abort.signal, + ); + return true; + } catch (e) { + if (!abort.signal.aborted) this.previewError = apiErrorText(e); + return false; + } finally { + this.previewing = false; + this.#previewAbort = null; + } + } + async start(request: BulkIngestRequest = this.request): Promise { if (this.busy) return false; this.busy = true; this.actionError = null; this.failures = null; + this.preview = null; try { this.#adopt(await this.#deps.startBulkIngest(request)); return true; diff --git a/frontend/src/lib/ingest/bulkIngestController.test.ts b/frontend/src/lib/ingest/bulkIngestController.test.ts index 0398b6da7..e41e97720 100644 --- a/frontend/src/lib/ingest/bulkIngestController.test.ts +++ b/frontend/src/lib/ingest/bulkIngestController.test.ts @@ -8,6 +8,7 @@ import { ApiError } from '$lib/api_errors'; import { RUN_ID, configFixture, + previewFixture, progressFixture, runFixture, } from '$lib/test/fixtures/bulkIngest'; @@ -30,6 +31,7 @@ function make(sequence: ReturnType[], recent = sequence.slice cancelBulkIngest: vi .fn() .mockResolvedValue(runFixture({ status: 'cancelled', poll_after_s: null })), + previewBulkIngest: vi.fn().mockResolvedValue(previewFixture()), }; return { ctl: new BulkIngest(deps), deps }; } @@ -193,3 +195,71 @@ describe('actions', () => { ctl.stop(); }); }); + +describe('previewing a source', () => { + it('sends only the source and extensions and keeps the served answer', async () => { + const { ctl, deps } = make([runFixture()]); + ctl.form.path = ' /data/source/coco '; + ctl.form.extensions = 'jpg, png'; + ctl.form.source = 'census'; + ctl.form.batchSize = 8; + expect(await ctl.runPreview()).toBe(true); + expect(deps.previewBulkIngest.mock.calls[0][0]).toEqual({ + folder: '/data/source/coco', + extensions: ['jpg', 'png'], + }); + expect(ctl.preview?.files).toBe(2000); + expect(deps.startBulkIngest).not.toHaveBeenCalled(); + expect(ctl.previewing).toBe(false); + }); + + it('cannot preview without a path', () => { + const { ctl } = make([runFixture()]); + expect(ctl.canPreview).toBe(false); + ctl.form.path = '/data/x'; + expect(ctl.canPreview).toBe(true); + }); + + it('shows the served refusal and no stale answer', async () => { + const { ctl, deps } = make([runFixture()]); + ctl.form.path = '/data/x'; + await ctl.runPreview(); + expect(ctl.preview).not.toBeNull(); + deps.previewBulkIngest.mockRejectedValueOnce( + new ApiError(422, 'bulk_ingest_invalid_source', '/x: not a directory'), + ); + expect(await ctl.runPreview()).toBe(false); + expect(ctl.preview).toBeNull(); + expect(ctl.previewError).toContain('not a directory'); + }); + + it('a second press aborts the walk and shows nothing', async () => { + const { ctl, deps } = make([runFixture()]); + ctl.form.path = '/data/x'; + let signal: AbortSignal | undefined; + deps.previewBulkIngest.mockImplementationOnce( + (_b: unknown, s?: AbortSignal) => + new Promise((_res, rej) => { + signal = s; + s?.addEventListener('abort', () => + rej(new DOMException('aborted', 'AbortError')), + ); + }), + ); + const first = ctl.runPreview(); + expect(ctl.previewing).toBe(true); + await ctl.runPreview(); + expect(signal?.aborted).toBe(true); + expect(await first).toBe(false); + expect(ctl.previewError).toBeNull(); + expect(ctl.previewing).toBe(false); + }); + + it('starting a run drops the preview', async () => { + const { ctl } = make([runFixture()]); + ctl.form.path = '/data/x'; + await ctl.runPreview(); + await ctl.start(); + expect(ctl.preview).toBeNull(); + }); +}); diff --git a/frontend/src/lib/test/fixtures/bulkIngest.ts b/frontend/src/lib/test/fixtures/bulkIngest.ts index 995070c49..886e8ae23 100644 --- a/frontend/src/lib/test/fixtures/bulkIngest.ts +++ b/frontend/src/lib/test/fixtures/bulkIngest.ts @@ -1,5 +1,6 @@ import type { BulkIngestConfig, + BulkPreview, BulkIngestProgress, BulkIngestRun, } from '$lib/types_bulkIngest'; @@ -14,6 +15,85 @@ export function configFixture(over: Partial = {}): BulkIngestC default_concurrency: 8, max_batch_size: 64, max_concurrency: 32, + preview: { + max_files: 200000, + max_seconds: 60, + header_sample: 300, + gpu_decode: true, + gpu_min_megapixels: 3, + rates: [ + { + route: 'cpu_small', + images_per_s: 29, + source_section: 'Decode once and the GPU decode (#217)', + measured_on: 'COCO frames (0.3 MP), CPU decode', + }, + { + route: 'cpu_large', + images_per_s: 27, + source_section: 'Decode once and the GPU decode (#217)', + measured_on: 'Commons photos (12 to 24 MP), CPU decode', + }, + { + route: 'gpu_large', + images_per_s: 35, + source_section: 'Decode once and the GPU decode (#217)', + measured_on: 'Commons photos (12 to 24 MP), GPU decode', + }, + ], + embeddings_per_image: { + value: 7.4, + unit: 'embeddings per image', + source_section: 'v0.5.0 baseline (set A, public COCO, 2,000 images)', + measured_on: 'COCO, policy all', + }, + opensearch_bytes_per_image: { + value: 65298, + unit: 'bytes per image', + source_section: 'Storage per image and per vector', + measured_on: 'COCO, policy all', + }, + source_document: 'docs/PERFORMANCE.md', + }, + ...over, + }; +} + +export function previewFixture(over: Partial = {}): BulkPreview { + return { + project: 'default', + kind: 'folder', + target: '/data/source/coco', + extensions: ['.jpg', '.jpeg', '.png'], + files: 2000, + total_bytes: 836_000_000, + by_extension: { + '.jpg': { files: 1990, bytes: 830_000_000 }, + '.png': { files: 10, bytes: 6_000_000 }, + }, + megapixels: { + sampled: 300, + unreadable: 0, + histogram: { '<1': 300 }, + gate_megapixels: 3, + share_at_or_above_gate: 0, + }, + estimate: { + basis: 'megapixel_sample', + duration_s_min: 69, + duration_s_max: 69, + vectors: 14800, + opensearch_bytes: 130_596_000, + }, + partial: false, + stop_reason: null, + unreadable_dirs: 0, + unreadable_samples: [], + symlinks_skipped: 0, + missing: 0, + outside_roots: 0, + unreadable_files: 0, + gpu_decode: true, ...over, }; } diff --git a/frontend/src/lib/types_bulkIngest.ts b/frontend/src/lib/types_bulkIngest.ts index f6b56ffa6..f7f78e791 100644 --- a/frontend/src/lib/types_bulkIngest.ts +++ b/frontend/src/lib/types_bulkIngest.ts @@ -25,6 +25,90 @@ export interface BulkIngestConfig { default_concurrency: number; max_batch_size: number; max_concurrency: number; + preview: BulkPreviewConfig; +} + +/** Served bounds of a preview and the measured figures its estimate uses. */ +export interface BulkPreviewConfig { + max_files: number; + max_seconds: number; + header_sample: number; + gpu_decode: boolean; + gpu_min_megapixels: number; + rates: PreviewRate[]; + embeddings_per_image: PreviewMeasured; + opensearch_bytes_per_image: PreviewMeasured; + /** The document the figures are measured in (named per figure by `source_section`). */ + source_document: string; +} + +export interface PreviewRate { + route: 'cpu_small' | 'cpu_large' | 'gpu_large'; + images_per_s: number; + source_section: string; + measured_on: string; +} + +export interface PreviewMeasured { + value: number; + unit: string; + source_section: string; + measured_on: string; +} + +/** `POST /ingest/bulk/preview`: exactly one of `folder`, `manifest`, `s3`, `source_id`. */ +export interface BulkPreviewRequest { + folder?: string | null; + manifest?: string | null; + s3?: S3Location | null; + source_id?: string | null; + extensions?: string[] | null; +} + +export interface PreviewExtension { + files: number; + bytes: number; +} + +export interface PreviewMegapixels { + sampled: number; + unreadable: number; + /** Bucket label (`<1`, `1-3`, `3-8`, `8-16`, `16-24`, `24+` megapixels) to count. */ + histogram: Record; + gate_megapixels: number; + share_at_or_above_gate?: number | null; +} + +export interface PreviewEstimate { + basis: 'megapixel_sample' | 'rate_range'; + duration_s_min: number; + duration_s_max: number; + vectors: number; + opensearch_bytes: number; +} + +export type PreviewStopReason = 'file_cap' | 'time_cap' | 'cancelled'; + +export interface BulkPreview { + project: string; + kind: 'folder' | 'manifest' | 's3'; + target: string; + extensions: string[]; + files: number; + total_bytes: number; + by_extension: Record; + megapixels: PreviewMegapixels; + estimate: PreviewEstimate; + /** A cap or a cancel ended the walk: every figure is a lower bound. */ + partial: boolean; + stop_reason?: PreviewStopReason | null; + unreadable_dirs: number; + unreadable_samples: string[]; + symlinks_skipped: number; + missing: number; + outside_roots: number; + unreadable_files: number; + gpu_decode: boolean; } /** `POST /ingest/bulk`: exactly one of `folder`, `manifest`, `s3` and `source_id`. */ diff --git a/monitoring/README.md b/monitoring/README.md index fdad85492..5b6d839c0 100644 --- a/monitoring/README.md +++ b/monitoring/README.md @@ -99,23 +99,28 @@ Rules in `monitoring/alerts/triton-alerts.yml`, evaluated by Prometheus and show ### Grafana unified alerting -Rules in `monitoring/grafana-alerting.yml`, evaluated by Grafana. The shipped contact -points are webhooks to `http://localhost:9999/...` (the Grafana container itself), so -nothing receives them until you set real ones (below). +Rules in `monitoring/grafana-alerting.yml`, evaluated by Grafana and filed under the +`OpenProcessor Alerts` folder. **Nobody is notified until you configure a contact point.** +The two shipped contact points are named `unconfigured-default` and +`unconfigured-critical`: they post to `http://localhost:9999/...` (the Grafana container +itself, where nothing listens), so a firing alert shows in Grafana and its delivery fails +with a logged error. Replace the receivers (below). **Alert Groups:** | Group | Alerts | |-------|--------| -| API Service Health | Error rate, slow response time | -| Triton Inference Server | Server down, high latency, high queue, model not ready (never fires, see below) | -| GPU Resources | High utilization, memory pressure, temperature (never fires, see below) | +| API Service Health | Error rate (5xx share of `http_request_duration_seconds_count`), slow response time | +| Triton Inference Server | Server down, high latency, high queue | +| GPU Resources | High utilization, memory pressure, temperature (DCGM, `DCGM_FI_DEV_GPU_TEMP`) | | System Resources | CPU, memory, load average | -| OpenSearch Health | Cluster status | -Triton does not export `nv_model_ready_state` or `nv_gpu_temperature`, so the Grafana rules -**Triton Model Not Ready** and **High GPU Temperature** can never fire. Use `GET /models/` -for model readiness and `DCGM_FI_DEV_GPU_TEMP` for temperature. +There is no rule for per-model readiness or OpenSearch cluster status: Triton does not +export `nv_model_ready_state` and no OpenSearch exporter ships in this stack, so a rule on +either could never fire. Use `GET /models/` for readiness and `GET /health` or the +`op_opensearch_*` gauges for the datastore. A rule or dashboard panel may only read a series +listed as exported in `tests/test_monitoring_metrics.py`; that test also covers the Grafana +rules, so a rule on a series nothing exports fails the offline suite. ### Alert Thresholds @@ -134,7 +139,9 @@ for model readiness and `DCGM_FI_DEV_GPU_TEMP` for temperature. ### Configuring Notifications -Edit `monitoring/grafana-alerting.yml` to configure notification channels: +Edit the receivers of the two contact points in `monitoring/grafana-alerting.yml` (the +file is provisioned read-only, so the Grafana UI cannot change them), then restart Grafana. +A real webhook needs only a new `url`. Slack and email, replacing the webhook receiver: ```yaml # Slack notifications @@ -144,22 +151,26 @@ Edit `monitoring/grafana-alerting.yml` to configure notification channels: url: "${SLACK_WEBHOOK_URL}" recipient: "#alerts" -# Email notifications +# Email notifications (also needs GF_SMTP_* set on the Grafana container) - uid: email-alerts type: email settings: addresses: "ops@example.com" ``` -Then add environment variables to docker-compose.yml: +`${SLACK_WEBHOOK_URL}` is read from the Grafana container's environment; the shipped +`docker-compose.yml` does not set it, so add it in a `docker-compose.override.yml`: ```yaml -grafana: - environment: - - SLACK_WEBHOOK_URL=https://hooks.slack.com/services/xxx - - ALERT_WEBHOOK_URL=https://your-webhook.example.com/alerts +services: + grafana: + environment: + - SLACK_WEBHOOK_URL=https://hooks.slack.com/services/xxx ``` +Once a receiver can deliver, rename its contact point (and the two names in `policies:`) +away from `unconfigured-`. + ### Viewing Active Alerts 1. **Grafana Alerting**: http://localhost:4605/alerting/list diff --git a/monitoring/grafana-alerting.yml b/monitoring/grafana-alerting.yml index 8eb1b51f1..232bf0c1c 100644 --- a/monitoring/grafana-alerting.yml +++ b/monitoring/grafana-alerting.yml @@ -1,8 +1,20 @@ # Grafana Unified Alerting Configuration # This file configures Grafana's native alerting system with: -# - Alert rules for API, Triton, and system metrics +# - Alert rules for the API, Triton, GPU and host metrics # - Contact points for notifications # - Notification policies for routing +# +# Every metric a rule reads must be one the stack exports: +# tests/test_monitoring_metric_names.py checks them against +# monitoring/exported_metrics.txt. Per-model readiness and OpenSearch cluster status are +# not exported by anything here, so there is no rule for them (see monitoring/README.md). +# +# NOTIFICATIONS ARE OFF UNTIL YOU CONFIGURE A CONTACT POINT. The two contact points +# below are named "unconfigured-*" on purpose: they post to a localhost address nothing +# listens on, so a firing alert shows in Grafana (Alerting > Alert rules, and the +# dashboards' firing-alerts panel) and its delivery fails with a logged error; nobody is +# told. Replace the receivers (a real webhook URL, Slack or email, examples below) and +# restart Grafana; the file is provisioned read-only, so edit it here, not in the UI. apiVersion: 1 @@ -10,20 +22,21 @@ apiVersion: 1 # Contact Points - Where alerts are sent # ============================================================================= contactPoints: - # Default contact point - customize with your notification channels + # Default (warning) alerts. Replace the receiver to notify anyone. - orgId: 1 - name: default-notifications + name: unconfigured-default receivers: - # Webhook for custom integrations (Discord, custom scripts, etc.) - # To use: set ALERT_WEBHOOK_URL in your .env or docker-compose.override.yml - uid: webhook-default type: webhook settings: + # Placeholder: nothing listens here. Set a real webhook URL (Discord, a + # script, an incident tool). url: "http://localhost:9999/alerts" httpMethod: POST disableResolveMessage: false - # Uncomment and configure for Slack notifications + # Slack: set SLACK_WEBHOOK_URL in the Grafana container's environment (for example + # in a docker-compose.override.yml) and uncomment. # - uid: slack-alerts # type: slack # settings: @@ -36,16 +49,16 @@ contactPoints: # {{ .Annotations.summary }} # {{ end }} - # Uncomment and configure for email notifications + # Email needs an SMTP server configured for Grafana (GF_SMTP_* variables). # - uid: email-alerts # type: email # settings: # addresses: "ops@example.com" # singleEmail: true - # Critical alerts - for urgent notifications + # Critical alerts (faster repeat). Replace the receiver the same way. - orgId: 1 - name: critical-notifications + name: unconfigured-critical receivers: - uid: webhook-critical type: webhook @@ -59,7 +72,7 @@ contactPoints: # ============================================================================= policies: - orgId: 1 - receiver: default-notifications + receiver: unconfigured-default group_by: - alertname - severity @@ -68,7 +81,7 @@ policies: repeat_interval: 4h routes: # Critical alerts go to critical channel with faster notification - - receiver: critical-notifications + - receiver: unconfigured-critical matchers: - severity = critical group_wait: 10s @@ -85,7 +98,7 @@ groups: # ------------------------------------------------------------------------- - orgId: 1 name: API Service Health - folder: Visual AI Alerts + folder: OpenProcessor Alerts interval: 30s rules: # API High Error Rate @@ -100,8 +113,8 @@ groups: datasourceUid: prometheus model: expr: | - sum(rate(http_requests_total{status=~"5.."}[5m])) / - sum(rate(http_requests_total[5m])) * 100 + sum(rate(http_request_duration_seconds_count{status=~"5.."}[5m])) / + sum(rate(http_request_duration_seconds_count[5m])) * 100 intervalMs: 1000 maxDataPoints: 43200 - refId: B @@ -197,7 +210,7 @@ groups: # ------------------------------------------------------------------------- - orgId: 1 name: Triton Inference Server - folder: Visual AI Alerts + folder: OpenProcessor Alerts interval: 30s rules: # Triton Server Down @@ -302,56 +315,6 @@ groups: severity: warning component: triton - # Model Not Ready - - uid: triton-model-not-ready - title: Triton Model Not Ready - condition: C - data: - - refId: A - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: prometheus - model: - expr: nv_model_ready_state == 0 - intervalMs: 1000 - maxDataPoints: 43200 - - refId: B - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: __expr__ - model: - conditions: - - evaluator: - params: [0] - type: gt - operator: - type: and - query: - params: - - A - reducer: - type: count - type: classic_conditions - - refId: C - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: __expr__ - model: - expression: B - type: threshold - noDataState: OK - execErrState: OK - for: 1m - annotations: - summary: "Model {{ $labels.model }} is not ready" - description: "Model is in not-ready state - inference requests will fail" - labels: - severity: critical - component: triton - # High Request Queue - uid: triton-high-queue title: Triton High Request Queue @@ -407,7 +370,7 @@ groups: # ------------------------------------------------------------------------- - orgId: 1 name: GPU Resources - folder: Visual AI Alerts + folder: OpenProcessor Alerts interval: 30s rules: # High GPU Utilization @@ -522,7 +485,7 @@ groups: to: 0 datasourceUid: prometheus model: - expr: nv_gpu_temperature + expr: DCGM_FI_DEV_GPU_TEMP intervalMs: 1000 maxDataPoints: 43200 - refId: B @@ -555,8 +518,8 @@ groups: execErrState: OK for: 2m annotations: - summary: "High GPU temperature on {{ $labels.gpu_uuid }}" - description: "GPU temperature above 80°C - check cooling" + summary: "High GPU temperature on GPU {{ $labels.gpu }}" + description: "GPU temperature (DCGM, needs the monitoring profile) above 80°C - check cooling" labels: severity: warning component: gpu @@ -566,7 +529,7 @@ groups: # ------------------------------------------------------------------------- - orgId: 1 name: System Resources - folder: Visual AI Alerts + folder: OpenProcessor Alerts interval: 30s rules: # High System Memory @@ -721,62 +684,3 @@ groups: labels: severity: warning component: system - - # ------------------------------------------------------------------------- - # OpenSearch Alerts - # ------------------------------------------------------------------------- - - orgId: 1 - name: OpenSearch Health - folder: Visual AI Alerts - interval: 60s - rules: - # OpenSearch Cluster Status - - uid: opensearch-cluster-status - title: OpenSearch Cluster Unhealthy - condition: C - data: - - refId: A - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: prometheus - model: - expr: | - opensearch_cluster_status{color!="green"} - intervalMs: 1000 - maxDataPoints: 43200 - - refId: B - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: __expr__ - model: - conditions: - - evaluator: - params: [0] - type: gt - operator: - type: and - query: - params: - - A - reducer: - type: count - type: classic_conditions - - refId: C - relativeTimeRange: - from: 300 - to: 0 - datasourceUid: __expr__ - model: - expression: B - type: threshold - noDataState: OK - execErrState: OK - for: 2m - annotations: - summary: "OpenSearch cluster is not healthy" - description: "Cluster status is not green - check cluster health" - labels: - severity: warning - component: opensearch diff --git a/scripts/codegen/baselines/complexity.txt b/scripts/codegen/baselines/complexity.txt index 4e410bf65..6daaf91ba 100644 --- a/scripts/codegen/baselines/complexity.txt +++ b/scripts/codegen/baselines/complexity.txt @@ -11,7 +11,6 @@ src/routers/curation/review_resolve.py::resolve_new_class_proposal 16 ratchet: m src/routers/curation/vlm.py::vlm_label_batch 23 ratchet: may only go down src/routers/ingest/batch.py::ingest_batch 21 ratchet: may only go down src/routers/ingest/directory.py::ingest_directory 18 ratchet: may only go down -src/services/curation/bulk_ingest/walk.py::iter_image_paths 16 ratchet: may only go down src/services/curation/clustering/auto_promote.py::auto_promote_clusters 16 ratchet: may only go down src/services/curation/clustering/orchestrator.py::cluster_residuals 19 ratchet: may only go down src/services/curation/dataset_import/coco.py::read_coco 20 ratchet: may only go down diff --git a/scripts/curation/bulk_ingest.py b/scripts/curation/bulk_ingest.py index a6889c0cf..5f37aaf4a 100755 --- a/scripts/curation/bulk_ingest.py +++ b/scripts/curation/bulk_ingest.py @@ -43,6 +43,10 @@ docker compose exec api python scripts/curation/bulk_ingest.py --project cars \\ --cancel bulk_20261011T120000_ab12cd34 --api-base http://localhost:8000/curation +``--preview`` with any source walks it read-only and prints the file count, bytes, +extension breakdown, image-size sample, estimated time and index footprint, without +starting a run (the walk is capped by the server and reports ``PARTIAL`` when it was). + Exit status: 0 completed, 1 completed with failed images, 2 failed (the run stopped), 130 cancelled. """ @@ -93,6 +97,12 @@ def build_parser() -> argparse.ArgumentParser: help='Curation API mount (no trailing slash); requests go to ' '/projects/<--project>/...', ) + parser.add_argument( + '--preview', + action='store_true', + help='Only report what the source holds and what ingesting it would cost (file count, ' + 'bytes, extensions, image sizes, estimated time and index size), then exit', + ) parser.add_argument('--source', default='bulk_ingest', help='source tag for ingested images') parser.add_argument( '--extensions', help='Comma-separated file extensions for --folder (default: jpg,jpeg,png)' @@ -133,7 +143,7 @@ def _call(client: httpx.Client, method: str, url: str, **kwargs: Any) -> dict[st def _start_body(args: argparse.Namespace) -> dict[str, Any]: - body: dict[str, Any] = {'source': args.source} + body: dict[str, Any] = {} if args.preview else {'source': args.source} if args.folder: body['folder'] = args.folder elif args.manifest: @@ -147,6 +157,8 @@ def _start_body(args: argparse.Namespace) -> dict[str, Any]: body['source_id'] = args.source_id if args.extensions: body['extensions'] = [e for e in args.extensions.split(',') if e.strip()] + if args.preview: + return body if args.batch_size: body['batch_size'] = args.batch_size if args.concurrency: @@ -162,6 +174,63 @@ def _fmt_eta(seconds: float | None) -> str: return f'{hours}h{minutes:02d}m' if hours else f'{minutes}m{secs:02d}s' +def _fmt_bytes(count: float) -> str: + for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'): + if abs(count) < 1024 or unit == 'TiB': + return f'{count:.0f} {unit}' if unit == 'B' else f'{count:.1f} {unit}' + count /= 1024 + return f'{count:.1f} TiB' + + +def _fmt_duration(low: float, high: float) -> str: + if round(low) == round(high): + return f'about {low:.1f} s' if low < 10 else f'about {_fmt_eta(low)}' + return f'{_fmt_eta(low)} to {_fmt_eta(high)}' + + +def print_preview(preview: dict[str, Any], config: dict[str, Any]) -> None: + """Render a ``POST .../ingest/bulk/preview`` answer; every figure is the server's.""" + print(f'{preview["kind"]} {preview["target"]}') + print(f' files: {preview["files"]}, {_fmt_bytes(preview["total_bytes"])}') + for ext, totals in sorted(preview['by_extension'].items()): + print(f' {ext}: {totals["files"]} files, {_fmt_bytes(totals["bytes"])}') + mp = preview['megapixels'] + if mp['sampled']: + shares = ', '.join(f'{k} MP: {v}' for k, v in mp['histogram'].items()) + gate = mp['share_at_or_above_gate'] * 100 + print( + f' image sizes (header sample of {mp["sampled"]}): {shares}; ' + f'{gate:.0f}% at or above {mp["gate_megapixels"]:g} MP' + ) + skipped = [ + (preview['unreadable_dirs'], 'unreadable directories'), + (preview['symlinks_skipped'], 'symlinks (not followed)'), + (preview['missing'], 'missing manifest entries'), + (preview['outside_roots'], 'entries outside the source roots'), + (preview['unreadable_files'], 'files that could not be read'), + ] + for count, label in skipped: + if count: + print(f' {label}: {count}') + for path in preview['unreadable_samples']: + print(f' unreadable: {path}') + est = preview['estimate'] + print( + f' estimated ingest time: {_fmt_duration(est["duration_s_min"], est["duration_s_max"])}' + f' ({est["basis"]})' + ) + print( + f' estimated index size: {_fmt_bytes(est["opensearch_bytes"])}, ' + f'about {est["vectors"]:.0f} vectors' + ) + print(f' basis: measured figures in {config["preview"]["source_document"]}, not a promise') + if preview['partial']: + print( + f' PARTIAL ({preview["stop_reason"]}): stopped early, so every figure above is a ' + 'lower bound' + ) + + def progress_line(run: dict[str, Any]) -> str: p = run['progress'] seen = f'{p["discovered"]}' + ('' if p['discovery_complete'] else '+') @@ -226,7 +295,10 @@ def _watch(client: httpx.Client, base: str, run: dict[str, Any], interval: float def main(argv: list[str] | None = None) -> int: - args = build_parser().parse_args(argv) + parser = build_parser() + args = parser.parse_args(argv) + if args.preview and (args.status or args.cancel): + parser.error('--preview needs a source (--folder, --manifest, --s3 or --source-id)') bind_script_project(args.project) base = f'{args.api_base.rstrip("/")}/projects/{args.project}' with httpx.Client() as client: @@ -239,6 +311,12 @@ def main(argv: list[str] | None = None) -> int: run = _call(client, 'GET', f'{base}/ingest/bulk/{args.status}') print(progress_line(run)) return EXIT_CODES.get(run['status'], 0) + if args.preview: + preview = _call( + client, 'POST', f'{base}/ingest/bulk/preview', json=_start_body(args) + ) + print_preview(preview, _call(client, 'GET', f'{base}/ingest/bulk/config')) + return 0 run = _call(client, 'POST', f'{base}/ingest/bulk', json=_start_body(args)) verb = 'resumed' if run['resumed'] else 'started' print(f'{verb} run {run["run_id"]} on {run["source"]["kind"]} {run["source"]["path"]}') diff --git a/src/routers/curation/__init__.py b/src/routers/curation/__init__.py index e6193c103..14bc39a73 100644 --- a/src/routers/curation/__init__.py +++ b/src/routers/curation/__init__.py @@ -43,6 +43,7 @@ import src.routers.curation.vlm # isort: skip import src.routers.curation.health import src.routers.curation.ingest_bulk +import src.routers.curation.ingest_bulk_preview import src.routers.curation.ingest_objects import src.routers.curation.ingest_policy import src.routers.curation.ingest_sources diff --git a/src/routers/curation/_ingest_bulk_models.py b/src/routers/curation/_ingest_bulk_models.py index 33dc4a0a3..ba151209c 100644 --- a/src/routers/curation/_ingest_bulk_models.py +++ b/src/routers/curation/_ingest_bulk_models.py @@ -6,6 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator +from src.routers.curation._ingest_bulk_preview_models import BulkPreviewConfig from src.routers.curation._ingest_sources_models import S3Location from src.services.curation.bulk_ingest.limits import ( DEFAULT_BATCH_SIZE, @@ -142,3 +143,4 @@ class BulkIngestConfig(BaseModel): default_concurrency: int max_batch_size: int max_concurrency: int + preview: BulkPreviewConfig diff --git a/src/routers/curation/_ingest_bulk_preview_models.py b/src/routers/curation/_ingest_bulk_preview_models.py new file mode 100644 index 000000000..77ba6cb1a --- /dev/null +++ b/src/routers/curation/_ingest_bulk_preview_models.py @@ -0,0 +1,108 @@ +"""Wire models for ``POST /ingest/bulk/preview`` and the preview block of the bulk config.""" + +from __future__ import annotations + +from typing import Literal, Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from src.routers.curation._ingest_sources_models import S3Location + + +class BulkPreviewRequest(BaseModel): + """Exactly one of ``folder``, ``manifest``, ``s3`` and ``source_id``, as for a start. + Nothing is ingested and nothing is written.""" + + model_config = ConfigDict(extra='forbid') + + folder: str | None = Field(default=None, min_length=1) + manifest: str | None = Field(default=None, min_length=1) + s3: S3Location | None = None + source_id: str | None = Field(default=None, min_length=1, max_length=64) + extensions: list[str] | None = Field(default=None, max_length=32) + + @model_validator(mode='after') + def _exactly_one_source(self) -> Self: + chosen = [self.folder, self.manifest, self.s3, self.source_id] + if sum(x is not None for x in chosen) != 1: + raise ValueError('send exactly one of folder, manifest, s3 and source_id') + return self + + +class PreviewExtensionWire(BaseModel): + files: int + bytes: int + + +class PreviewMegapixelsWire(BaseModel): + # Headers read (no pixel decoded) from a bounded uniform sample of the files; 0 for S3. + sampled: int + unreadable: int + # Bucket label ("<1", "1-3", "3-8", "8-16", "16-24", "24+" megapixels) to sampled count. + histogram: dict[str, int] + # The GPU-decode gate (OP_INGEST_GPU_MIN_MEGAPIXELS) the share is measured against. + gate_megapixels: float + share_at_or_above_gate: float | None = None + + +class PreviewEstimateWire(BaseModel): + # megapixel_sample: one rate per image from the sampled sizes; rate_range: sizes + # unknown, so the span from the slowest to the fastest measured route. + basis: Literal['megapixel_sample', 'rate_range'] + duration_s_min: float + duration_s_max: float + vectors: float + opensearch_bytes: int + + +class BulkPreview(BaseModel): + project: str + kind: Literal['folder', 'manifest', 's3'] + # What was previewed: the folder, manifest path or s3://bucket/prefix. + target: str + extensions: list[str] + files: int + total_bytes: int + by_extension: dict[str, PreviewExtensionWire] + megapixels: PreviewMegapixelsWire + estimate: PreviewEstimateWire + # True when a cap or a cancel ended the walk: every figure is then a lower bound. + partial: bool + stop_reason: Literal['file_cap', 'time_cap', 'cancelled'] | None = None + unreadable_dirs: int + unreadable_samples: list[str] + symlinks_skipped: int + # Manifest entries that do not exist / lie outside the source roots; files not statable. + missing: int + outside_roots: int + unreadable_files: int + gpu_decode: bool + + +class PreviewRateWire(BaseModel): + route: Literal['cpu_small', 'cpu_large', 'gpu_large'] + images_per_s: float + source_section: str + measured_on: str + + +class PreviewMeasuredWire(BaseModel): + value: float + unit: str + source_section: str + measured_on: str + + +class BulkPreviewConfig(BaseModel): + """The bounds of a preview and the measured figures its estimate uses, all from + ``docs/PERFORMANCE.md`` (the section is named per figure).""" + + max_files: int + max_seconds: float + header_sample: int + gpu_decode: bool + gpu_min_megapixels: float + rates: list[PreviewRateWire] + embeddings_per_image: PreviewMeasuredWire + opensearch_bytes_per_image: PreviewMeasuredWire + source_document: str diff --git a/src/routers/curation/ingest_bulk.py b/src/routers/curation/ingest_bulk.py index c8e694223..62298c72b 100644 --- a/src/routers/curation/ingest_bulk.py +++ b/src/routers/curation/ingest_bulk.py @@ -24,6 +24,7 @@ BulkIngestRun, ) from src.routers.curation._ingest_bulk_wire import open_run, own_address, run_wire, slug +from src.routers.curation.ingest_bulk_preview import preview_config from src.routers.curation.ingest_sources import resolve_bulk_source, start_run_now from src.services.curation import image_serving from src.services.curation.bulk_ingest import limits, runner, sender @@ -48,6 +49,7 @@ async def bulk_ingest_config() -> BulkIngestConfig: default_concurrency=limits.DEFAULT_CONCURRENCY, max_batch_size=limits.MAX_BATCH_SIZE, max_concurrency=limits.MAX_CONCURRENCY, + preview=preview_config(), ) diff --git a/src/routers/curation/ingest_bulk_preview.py b/src/routers/curation/ingest_bulk_preview.py new file mode 100644 index 000000000..3c9ae6c2a --- /dev/null +++ b/src/routers/curation/ingest_bulk_preview.py @@ -0,0 +1,190 @@ +"""``POST /ingest/bulk/preview``: what a bulk ingest would touch and cost, with no ingestion. + +A read-only look at a folder, manifest, S3 prefix or registered source, bounded by the +served caps and cancelled when the client disconnects. Nothing is written and no +datastore is read; the source must obey the same rules a start does (inside a source +root, readable, inside ``OP_S3_ALLOWED``). +""" + +from __future__ import annotations + +import asyncio +import threading +from dataclasses import asdict +from typing import TYPE_CHECKING + +from fastapi import HTTPException, Request + +from src.config import get_curation_config +from src.routers.curation._api_error import api_error +from src.routers.curation._common import router +from src.routers.curation._ingest_bulk_preview_models import ( + BulkPreview, + BulkPreviewConfig, + BulkPreviewRequest, + PreviewEstimateWire, + PreviewExtensionWire, + PreviewMeasuredWire, + PreviewMegapixelsWire, + PreviewRateWire, +) +from src.routers.curation._ingest_bulk_wire import slug +from src.services.curation.bulk_ingest import preflight, preflight_rates +from src.services.curation.bulk_ingest.object_store import ( + ObjectStoreError, + S3Config, + S3NotAllowedError, + S3NotConfiguredError, + get_object_store, +) +from src.services.curation.bulk_ingest.source import ( + BulkSource, + InvalidSourceError, + normalise_extensions, +) +from src.services.curation.bulk_ingest.sources import load_source + + +if TYPE_CHECKING: + from collections.abc import Callable + + from src.services.curation.bulk_ingest.preflight import PreviewResult + +PREVIEW_LIMITS = preflight.PreviewLimits() +_POLL_DISCONNECT_S = 0.5 + + +def preview_config() -> BulkPreviewConfig: + """Served inside ``GET /ingest/bulk/config``.""" + config = get_curation_config() + return BulkPreviewConfig( + max_files=PREVIEW_LIMITS.max_files, + max_seconds=PREVIEW_LIMITS.max_seconds, + header_sample=PREVIEW_LIMITS.header_sample, + gpu_decode=config.ingest_gpu_decode, + gpu_min_megapixels=config.ingest_gpu_min_megapixels, + rates=[PreviewRateWire(**asdict(r)) for r in preflight_rates.RATES], + embeddings_per_image=PreviewMeasuredWire(**asdict(preflight_rates.EMBEDDINGS_PER_IMAGE)), + opensearch_bytes_per_image=PreviewMeasuredWire( + **asdict(preflight_rates.OPENSEARCH_BYTES_PER_IMAGE) + ), + source_document='docs/PERFORMANCE.md', + ) + + +def _wire( + result: PreviewResult, kind: str, target: str, extensions: tuple[str, ...] +) -> BulkPreview: + return BulkPreview( + project=slug(), + kind=kind, + target=target, + extensions=list(extensions), + files=result.files, + total_bytes=result.total_bytes, + by_extension={e: PreviewExtensionWire(**asdict(t)) for e, t in result.by_extension.items()}, + megapixels=PreviewMegapixelsWire(**asdict(result.megapixels)), + estimate=PreviewEstimateWire(**asdict(result.estimate)), + partial=result.partial, + stop_reason=result.stop_reason, + unreadable_dirs=result.unreadable_dirs, + unreadable_samples=result.unreadable_samples, + symlinks_skipped=result.symlinks_skipped, + missing=result.missing, + outside_roots=result.outside_roots, + unreadable_files=result.unreadable_files, + gpu_decode=result.gpu_decode, + ) + + +def _plan( + body: BulkPreviewRequest, +) -> tuple[str, str, tuple[str, ...], Callable[..., PreviewResult]]: + """``(kind, target, extensions, run)`` for the request; ``run`` takes ``should_stop``.""" + config = get_curation_config() + gpu_decode = config.ingest_gpu_decode + if body.source_id is not None: + defn = load_source(body.source_id) + if defn is None: + raise api_error( + 404, + 'ingest_source_not_found', + f'no ingest source {body.source_id!r}', + project=slug(), + ) + extensions = tuple(defn.extensions) + if defn.backend == 'folder': + body = BulkPreviewRequest(folder=defn.path, extensions=list(extensions)) + else: + from src.routers.curation._ingest_sources_models import S3Location + + body = BulkPreviewRequest( + s3=S3Location(bucket=defn.bucket, prefix=defn.prefix), extensions=list(extensions) + ) + extensions = normalise_extensions(body.extensions) + if body.s3 is not None: + s3 = body.s3 + prefix = s3.prefix.lstrip('/') + + def run_s3(should_stop: Callable[[], bool]) -> PreviewResult: + s3_config = S3Config.from_env() + s3_config.check(s3.bucket, prefix) + return preflight.preview_s3( + get_object_store(), + s3_config, + s3.bucket, + prefix, + extensions, + limits=PREVIEW_LIMITS, + should_stop=should_stop, + gpu_decode=gpu_decode, + ) + + return 's3', f's3://{s3.bucket}/{prefix}', extensions, run_s3 + kind = 'folder' if body.folder is not None else 'manifest' + path = body.folder or body.manifest or '' + source = BulkSource(kind, path, extensions) # type: ignore[arg-type] + + def run_path(should_stop: Callable[[], bool]) -> PreviewResult: + return preflight.preview_source( + source, + limits=PREVIEW_LIMITS, + should_stop=should_stop, + gpu_decode=gpu_decode, + gpu_min_megapixels=config.ingest_gpu_min_megapixels, + ) + + return kind, path, extensions, run_path + + +@router.post('/ingest/bulk/preview', response_model=BulkPreview) +async def preview_bulk_ingest(body: BulkPreviewRequest, request: Request) -> BulkPreview: + """Count the files, bytes and extensions of a source, sample image sizes, and estimate + duration and index footprint from measured figures (``GET /ingest/bulk/config``). + Capped by file count and seconds (``partial: true`` then); a client disconnect stops + the walk.""" + kind, target, extensions, run = _plan(body) + cancelled = threading.Event() + + async def watch_disconnect() -> None: + while not cancelled.is_set(): + if await request.is_disconnected(): + cancelled.set() + return + await asyncio.sleep(_POLL_DISCONNECT_S) + + watcher = asyncio.create_task(watch_disconnect()) + try: + result = await asyncio.to_thread(run, cancelled.is_set) + except InvalidSourceError as exc: + raise api_error(422, 'bulk_ingest_invalid_source', str(exc), project=slug()) from None + except S3NotConfiguredError as exc: + raise api_error(503, 's3_not_configured', str(exc), project=slug()) from None + except S3NotAllowedError as exc: + raise api_error(422, 's3_not_allowed', str(exc), project=slug()) from None + except ObjectStoreError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from None + finally: + cancelled.set() + watcher.cancel() + return _wire(result, kind, target, extensions) diff --git a/src/services/curation/bulk_ingest/preflight.py b/src/services/curation/bulk_ingest/preflight.py new file mode 100644 index 000000000..2e4788c61 --- /dev/null +++ b/src/services/curation/bulk_ingest/preflight.py @@ -0,0 +1,328 @@ +"""Preflight for a bulk ingest: what a folder, manifest or S3 prefix holds and roughly +what ingesting it would cost, with no ingestion and no write of any kind. + +The walk is bounded (a file cap and a time cap, both served by ``GET /ingest/bulk/config``) +and stoppable: a source too large to finish inside the caps answers with what was seen +and ``partial: true`` plus the reason, never a silent truncation. An unreadable root +refuses. Image sizes come from a bounded sample of file headers (no pixel is decoded); +an S3 prefix is only listed, so its estimate is a range across the measured routes. +""" + +from __future__ import annotations + +import random +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Literal + +from PIL import Image + +from src.services.curation.bulk_ingest import preflight_rates +from src.services.curation.bulk_ingest.s3_listing import wanted_extension +from src.services.curation.bulk_ingest.source import ( + BulkSource, + InvalidSourceError, + iter_manifest_paths, + validate_source, +) +from src.services.curation.bulk_ingest.walk import WalkStats, iter_image_paths +from src.services.curation.image_serving import is_servable_image_path + + +if TYPE_CHECKING: + from collections.abc import Callable, Iterable + + from src.services.curation.bulk_ingest.object_store import ObjectStore, S3Config + +StopReason = Literal['file_cap', 'time_cap', 'cancelled'] +# Upper edges (megapixels) of the histogram buckets; the last bucket is open. +HISTOGRAM_EDGES_MP: tuple[float, ...] = (1.0, 3.0, 8.0, 16.0, 24.0) +HEADER_SAMPLE_SEED = 20261010 + + +@dataclass(frozen=True) +class PreviewLimits: + """Bounds on one preview, served to clients as part of the bulk-ingest config.""" + + max_files: int = 200_000 + max_seconds: float = 60.0 + header_sample: int = 300 + + +@dataclass +class ExtensionTotals: + files: int = 0 + bytes: int = 0 + + +@dataclass(frozen=True) +class MegapixelSample: + sampled: int = 0 + unreadable: int = 0 + histogram: dict[str, int] = field(default_factory=dict) + gate_megapixels: float = 0.0 + # Share of the sampled images at or above the GPU-decode gate; None without a sample. + share_at_or_above_gate: float | None = None + + +@dataclass +class PreviewResult: + files: int = 0 + total_bytes: int = 0 + by_extension: dict[str, ExtensionTotals] = field(default_factory=dict) + megapixels: MegapixelSample = field(default_factory=MegapixelSample) + estimate: preflight_rates.Estimate = field( + default_factory=lambda: preflight_rates.estimate( + files=0, share_large=None, gpu_decode=False + ) + ) + partial: bool = False + stop_reason: StopReason | None = None + unreadable_dirs: int = 0 + unreadable_samples: list[str] = field(default_factory=list) + symlinks_skipped: int = 0 + # Manifest entries only. + missing: int = 0 + outside_roots: int = 0 + # Files that could not be statted. + unreadable_files: int = 0 + gpu_decode: bool = False + + +class _Budget: + """The caps and the cancel flag of one preview; remembers which one tripped.""" + + def __init__( + self, + limits: PreviewLimits, + should_stop: Callable[[], bool], + clock: Callable[[], float], + ) -> None: + self.limits = limits + self._should_stop = should_stop + self._clock = clock + self._started = clock() + self.reason: StopReason | None = None + + def stopped(self) -> bool: + """Cancel or time cap (also the walk's stop poll).""" + if self.reason is not None: + return True + if self._should_stop(): + self.reason = 'cancelled' + elif self._clock() - self._started > self.limits.max_seconds: + self.reason = 'time_cap' + return self.reason is not None + + def full(self, files: int) -> bool: + if files >= self.limits.max_files: + self.reason = 'file_cap' + return self.reason is not None + + +class _Reservoir: + """A fixed-size uniform sample of the paths seen (algorithm R, seeded).""" + + def __init__(self, size: int) -> None: + self.size = size + self.items: list[str] = [] + self._seen = 0 + self._rng = random.Random(HEADER_SAMPLE_SEED) + + def add(self, item: str) -> None: + self._seen += 1 + if len(self.items) < self.size: + self.items.append(item) + return + slot = self._rng.randrange(self._seen) + if slot < self.size: + self.items[slot] = item + + +def _bucket_label(megapixels: float) -> str: + previous = 0.0 + for edge in HISTOGRAM_EDGES_MP: + if megapixels < edge: + return f'<{edge:g}' if previous == 0.0 else f'{previous:g}-{edge:g}' + previous = edge + return f'{previous:g}+' + + +def _sample_headers( + paths: Iterable[str], gate_megapixels: float, stopped: Callable[[], bool] +) -> MegapixelSample: + histogram: dict[str, int] = {} + sampled = unreadable = at_gate = 0 + for path in paths: + if stopped(): + break + try: + with Image.open(path) as image: + width, height = image.size + except (OSError, ValueError, Image.DecompressionBombError): + unreadable += 1 + continue + megapixels = width * height / 1e6 + label = _bucket_label(megapixels) + histogram[label] = histogram.get(label, 0) + 1 + sampled += 1 + at_gate += megapixels >= gate_megapixels + return MegapixelSample( + sampled, + unreadable, + histogram, + gate_megapixels, + at_gate / sampled if sampled else None, + ) + + +def _finish( + result: PreviewResult, + budget: _Budget, + sample: MegapixelSample, + *, + gpu_decode: bool, +) -> PreviewResult: + result.megapixels = sample + result.gpu_decode = gpu_decode + result.partial = budget.reason is not None + result.stop_reason = budget.reason + result.estimate = preflight_rates.estimate( + files=result.files, share_large=sample.share_at_or_above_gate, gpu_decode=gpu_decode + ) + return result + + +def _count(result: PreviewResult, suffix: str, size: int) -> None: + result.files += 1 + result.total_bytes += size + totals = result.by_extension.setdefault(suffix, ExtensionTotals()) + totals.files += 1 + totals.bytes += size + + +def preview_source( + source: BulkSource, + *, + limits: PreviewLimits = PreviewLimits(), # noqa: B008 - frozen defaults + should_stop: Callable[[], bool] = lambda: False, + clock: Callable[[], float] = time.monotonic, + gpu_decode: bool = True, + gpu_min_megapixels: float = 3.0, + walk_workers: int = 8, +) -> PreviewResult: + """Preview a folder or manifest. Blocking: run it off the event loop. Raises + :class:`InvalidSourceError` for a path outside the source roots, a missing or + unreadable root.""" + if source.kind not in {'folder', 'manifest'}: + raise InvalidSourceError(f'cannot preview a {source.kind} source') + validate_source(source) + budget = _Budget(limits, should_stop, clock) + result = PreviewResult() + reservoir = _Reservoir(limits.header_sample) + if source.kind == 'folder': + _preview_folder(source, result, budget, reservoir, walk_workers) + else: + _preview_manifest(source, result, budget, reservoir) + sample = _sample_headers(reservoir.items, gpu_min_megapixels, should_stop) + return _finish(result, budget, sample, gpu_decode=gpu_decode) + + +def _preview_folder( + source: BulkSource, + result: PreviewResult, + budget: _Budget, + reservoir: _Reservoir, + walk_workers: int, +) -> None: + stats = WalkStats() + walk = iter_image_paths( + Path(source.path), + frozenset(source.extensions), + walk_workers, + stats=stats, + should_stop=budget.stopped, + ) + try: + for path in walk: + if budget.stopped() or budget.full(result.files): + break + try: + size = path.stat().st_size + except OSError: + result.unreadable_files += 1 + continue + _count(result, path.suffix.lower(), size) + reservoir.add(str(path)) + finally: + walk.close() + result.unreadable_dirs = stats.unreadable_dirs + result.unreadable_samples = list(stats.unreadable_samples) + result.symlinks_skipped = stats.symlinks + + +def _preview_manifest( + source: BulkSource, result: PreviewResult, budget: _Budget, reservoir: _Reservoir +) -> None: + for entry in iter_manifest_paths(Path(source.path)): + if budget.stopped() or budget.full(result.files + result.missing + result.outside_roots): + break + if not is_servable_image_path(entry): + result.outside_roots += 1 + continue + try: + size = Path(entry).stat().st_size + except FileNotFoundError: + result.missing += 1 + continue + except OSError: + result.unreadable_files += 1 + continue + _count(result, Path(entry).suffix.lower(), size) + reservoir.add(entry) + + +def preview_s3( + store: ObjectStore, + config: S3Config, + bucket: str, + prefix: str, + extensions: Iterable[str], + *, + limits: PreviewLimits = PreviewLimits(), # noqa: B008 - frozen defaults + should_stop: Callable[[], bool] = lambda: False, + clock: Callable[[], float] = time.monotonic, + gpu_decode: bool = True, +) -> PreviewResult: + """Preview a bucket/prefix by listing it (sizes and keys only, no object is read). + Raises the allow-list errors of :meth:`S3Config.check` and the store's classified + listing errors. Blocking.""" + config.check(bucket, prefix) + wanted = frozenset(e.lower() for e in extensions) + budget = _Budget(limits, should_stop, clock) + result = PreviewResult() + token: str | None = None + while not budget.stopped(): + page = store.list_page(bucket, prefix, token=token, delimiter=None) + for listed in page.objects: + if not wanted_extension(listed.ref.key, wanted): + continue + if budget.full(result.files): + break + dot = listed.ref.key.rfind('.') + _count(result, listed.ref.key[dot:].lower(), listed.ref.size or 0) + token = page.next_token + if token is None or budget.reason is not None: + break + return _finish(result, budget, MegapixelSample(), gpu_decode=gpu_decode) + + +__all__ = [ + 'ExtensionTotals', + 'MegapixelSample', + 'PreviewLimits', + 'PreviewResult', + 'preview_s3', + 'preview_source', +] diff --git a/src/services/curation/bulk_ingest/preflight_rates.py b/src/services/curation/bulk_ingest/preflight_rates.py new file mode 100644 index 000000000..fa16b4721 --- /dev/null +++ b/src/services/curation/bulk_ingest/preflight_rates.py @@ -0,0 +1,99 @@ +"""The measured figures a bulk-ingest preview estimates from, and the estimate itself. + +Nothing here is a guess: every number is copied from ``docs/PERFORMANCE.md`` (the +section is named beside it, and a test checks the figure and the section are still in +that file). They are served to clients by ``GET /ingest/bulk/config`` so a screen shows +the basis of an estimate instead of an unexplained number. They describe the measured +hardware (one RTX A6000, policy ``all`` on public COCO and Commons photos); another +host or detector profile will differ, which is why the answer is labelled an estimate. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + + +DECODE_SECTION = 'Decode once and the GPU decode (#217)' +STORAGE_SECTION = 'Storage per image and per vector' +BASELINE_SECTION = 'v0.5.0 baseline (set A, public COCO, 2,000 images)' + + +@dataclass(frozen=True) +class Rate: + route: Literal['cpu_small', 'cpu_large', 'gpu_large'] + images_per_s: float + source_section: str + measured_on: str + + +@dataclass(frozen=True) +class Measured: + value: float + unit: str + source_section: str + measured_on: str + + +RATES: tuple[Rate, ...] = ( + Rate( + 'cpu_small', + 29.0, + DECODE_SECTION, + 'COCO frames (0.3 MP), CPU decode, detect + embed, /ingest/batch, 1 x A6000', + ), + Rate( + 'cpu_large', + 27.0, + DECODE_SECTION, + 'Commons photos (12 to 24 MP), CPU decode, detect + embed, /ingest/batch, 1 x A6000', + ), + Rate( + 'gpu_large', + 35.0, + DECODE_SECTION, + 'Commons photos (12 to 24 MP), GPU decode, detect + embed, /ingest/batch, 1 x A6000', + ), +) +EMBEDDINGS_PER_IMAGE = Measured( + 7.4, 'embeddings per image', BASELINE_SECTION, 'COCO, policy all (detector + PE encoder)' +) +OPENSEARCH_BYTES_PER_IMAGE = Measured( + 65298, + 'bytes per image', + STORAGE_SECTION, + 'COCO, policy all, /ingest/batch, merged until the size repeats (images + items indexes)', +) + + +@dataclass(frozen=True) +class Estimate: + # ``megapixel_sample``: one rate per image from the sampled sizes. ``rate_range``: + # sizes unknown, so the span from the slowest to the fastest measured route. + basis: Literal['megapixel_sample', 'rate_range'] + duration_s_min: float + duration_s_max: float + vectors: float + opensearch_bytes: int + + +def estimate(*, files: int, share_large: float | None, gpu_decode: bool) -> Estimate: + """Duration and footprint for ``files`` images. ``share_large`` is the sampled share + of images at or above the GPU-decode gate, or ``None`` when no sizes are known.""" + rates = {r.route: r.images_per_s for r in RATES} + if share_large is None: + fastest, slowest = max(rates.values()), min(rates.values()) + basis: Literal['megapixel_sample', 'rate_range'] = 'rate_range' + low, high = files / fastest, files / slowest + else: + large_rate = rates['gpu_large'] if gpu_decode else rates['cpu_large'] + large = files * share_large + low = high = (files - large) / rates['cpu_small'] + large / large_rate + basis = 'megapixel_sample' + return Estimate( + basis=basis, + duration_s_min=low, + duration_s_max=high, + vectors=files * EMBEDDINGS_PER_IMAGE.value, + opensearch_bytes=round(files * OPENSEARCH_BYTES_PER_IMAGE.value), + ) diff --git a/src/services/curation/bulk_ingest/source.py b/src/services/curation/bulk_ingest/source.py index 529ab41b3..9ab29f1fb 100644 --- a/src/services/curation/bulk_ingest/source.py +++ b/src/services/curation/bulk_ingest/source.py @@ -18,6 +18,7 @@ from __future__ import annotations import json +import os from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Literal @@ -85,6 +86,13 @@ def validate_source(source: BulkSource) -> None: path = Path(source.path) if source.kind == 'folder' and not path.is_dir(): raise InvalidSourceError(f'{source.path}: not a directory') + if source.kind == 'folder': + # A walk lists an unreadable directory as empty, so a run would "complete" with + # nothing discovered; refuse the unreadable root up front instead. + try: + os.scandir(path).close() + except OSError as exc: + raise InvalidSourceError(f'{source.path}: not readable ({exc.strerror})') from None if source.kind == 'manifest' and not path.is_file(): raise InvalidSourceError(f'{source.path}: not a file') @@ -102,7 +110,7 @@ def _manifest_entry(line: str) -> str | None: return text -def _manifest_paths(path: Path) -> Iterator[str]: +def iter_manifest_paths(path: Path) -> Iterator[str]: with path.open(encoding='utf-8') as fh: for line in fh: entry = _manifest_entry(line) @@ -142,7 +150,7 @@ def iter_source_paths( yield from _queue_refs(source, should_stop) return if source.kind == 'manifest': - yield from _manifest_paths(Path(source.path)) + yield from iter_manifest_paths(Path(source.path)) return for found in iter_image_paths(Path(source.path), frozenset(source.extensions), walk_workers): yield str(found) diff --git a/src/services/curation/bulk_ingest/walk.py b/src/services/curation/bulk_ingest/walk.py index e8929d5a5..dc9b41884 100644 --- a/src/services/curation/bulk_ingest/walk.py +++ b/src/services/curation/bulk_ingest/walk.py @@ -18,97 +18,153 @@ import queue import threading from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Callable, Generator -def iter_image_paths( - root: Path, - extensions: frozenset[str] = frozenset({'.jpg', '.jpeg', '.png'}), - workers: int = 8, -) -> Iterator[Path]: - """Yield every file under ``root`` (recursively) matching ``extensions``. +UNREADABLE_SAMPLE_LIMIT = 10 - Case-insensitive on suffix. Directories that raise ``OSError`` while - being listed (permission denied, a broken symlink, a mount that - dropped mid-walk) are skipped, not fatal — a single bad directory - must never abort a multi-hour ingest sweep. - Args: - root: Directory to walk. - extensions: Lowercase suffixes to include, e.g. ``{'.jpg'}``. - workers: Thread-pool size for concurrent ``os.scandir`` calls. +@dataclass +class WalkStats: + """What a walk skipped, so a caller can report it instead of hiding it.""" - Yields: - ``Path`` objects for matching files, in discovery order (not - sorted — callers needing a stable order should sort the result). - """ - exts = frozenset(e.lower() for e in extensions) - out: queue.Queue[Path] = queue.Queue(maxsize=4096) - pending = 0 - pending_lock = threading.Lock() - pending_zero = threading.Event() - # Set when the consumer stops early (a cancelled run): a worker blocked on a - # full queue must give up, or closing the pool would wait on it forever. - stop = threading.Event() - - def _put(path: Path) -> None: - while not stop.is_set(): + unreadable_dirs: int = 0 + symlinks: int = 0 + # The first few unreadable directories, for the message. + unreadable_samples: list[str] = field(default_factory=list) + _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + def add_unreadable(self, path: Path) -> None: + with self._lock: + self.unreadable_dirs += 1 + if len(self.unreadable_samples) < UNREADABLE_SAMPLE_LIMIT: + self.unreadable_samples.append(str(path)) + + def add_symlink(self) -> None: + with self._lock: + self.symlinks += 1 + + +class _Walk: + """One parallel walk: workers scan directories and feed a bounded queue the caller drains.""" + + def __init__( + self, root: Path, extensions: frozenset[str], workers: int, stats: WalkStats | None + ) -> None: + self._root = root + self._exts = frozenset(e.lower() for e in extensions) + self._stats = stats + self._out: queue.Queue[Path] = queue.Queue(maxsize=4096) + self._pending = 0 + self._pending_lock = threading.Lock() + self._pending_zero = threading.Event() + # Set when the consumer stops early (a cancelled run): a worker blocked on a + # full queue must give up, or closing the pool would wait on it forever. + self._stop = threading.Event() + self._executor = ThreadPoolExecutor(max_workers=max(1, workers)) + + def _put(self, path: Path) -> None: + while not self._stop.is_set(): try: - out.put(path, timeout=0.1) + self._out.put(path, timeout=0.1) return except queue.Full: continue - def _scan(dir_path: Path) -> None: - nonlocal pending + def _list(self, dir_path: Path) -> list[os.DirEntry[str]]: + # A closed walk must not keep listing the rest of a huge tree behind it. + if self._stop.is_set(): + return [] try: with os.scandir(dir_path) as it: - entries = list(it) + return list(it) except OSError: - entries = [] - for entry in entries: + if self._stats is not None: + self._stats.add_unreadable(dir_path) + return [] + + def _visit(self, entry: os.DirEntry[str]) -> None: + if entry.is_symlink(): + if self._stats is not None: + self._stats.add_symlink() + elif entry.is_dir(follow_symlinks=False): + with self._pending_lock: + self._pending += 1 + self._executor.submit(self._scan, Path(entry.path)) + elif entry.is_file(follow_symlinks=False) and Path(entry.name).suffix.lower() in self._exts: + self._put(Path(entry.path)) + + def _scan(self, dir_path: Path) -> None: + for entry in self._list(dir_path): try: - if entry.is_dir(follow_symlinks=False): - with pending_lock: - pending += 1 - executor.submit(_scan, Path(entry.path)) - elif entry.is_file(follow_symlinks=False): - suffix = Path(entry.name).suffix.lower() - if suffix in exts: - _put(Path(entry.path)) + self._visit(entry) except OSError: continue - with pending_lock: - pending -= 1 - if pending == 0: - pending_zero.set() - - with ThreadPoolExecutor(max_workers=max(1, workers)) as executor: - with pending_lock: - pending += 1 - executor.submit(_scan, root) - + with self._pending_lock: + self._pending -= 1 + if self._pending == 0: + self._pending_zero.set() + + def run(self, should_stop: Callable[[], bool]) -> Generator[Path, None, None]: + with self._pending_lock: + self._pending += 1 + self._executor.submit(self._scan, self._root) try: - # Drain the queue on the calling thread while workers fan out; - # a sentinel-free design would race the "done" check against - # in-flight puts, so we poll with a short timeout instead. + # Drain the queue on the calling thread while workers fan out; a sentinel-free + # design would race the "done" check against in-flight puts, so we poll with a + # short timeout instead, which is also where a stop request is noticed even + # when the tree yields no matching file. while True: try: - item = out.get(timeout=0.1) - yield item + yield self._out.get(timeout=0.1) except queue.Empty: - if pending_zero.is_set() and out.empty(): + if self._pending_zero.is_set() and self._out.empty(): break + if should_stop(): + return # Drain anything queued between the last empty check and shutdown. - while not out.empty(): - yield out.get_nowait() + while not self._out.empty(): + yield self._out.get_nowait() finally: - stop.set() + self._stop.set() + # Never wait on a listing stuck on a dead mount: a stopped walk returns now + # and the stuck worker thread ends when its syscall does. + self._executor.shutdown(wait=False, cancel_futures=True) + + +def iter_image_paths( + root: Path, + extensions: frozenset[str] = frozenset({'.jpg', '.jpeg', '.png'}), + workers: int = 8, + stats: WalkStats | None = None, + should_stop: Callable[[], bool] = lambda: False, +) -> Generator[Path, None, None]: + """Yield every file under ``root`` (recursively) matching ``extensions``. + + Case-insensitive on suffix. Directories that raise ``OSError`` while + being listed (permission denied, a broken symlink, a mount that + dropped mid-walk) are skipped, not fatal — a single bad directory + must never abort a multi-hour ingest sweep. + + Args: + root: Directory to walk. + extensions: Lowercase suffixes to include, e.g. ``{'.jpg'}``. + workers: Thread-pool size for concurrent ``os.scandir`` calls. + should_stop: Polled while the walk waits; when true the walk ends (a cancelled + or time-boxed caller), even if the tree has yielded nothing for a while. + stats: Filled with the unreadable directories and symlinks the walk skipped. + + Yields: + ``Path`` objects for matching files, in discovery order (not + sorted — callers needing a stable order should sort the result). + """ + yield from _Walk(root, extensions, workers, stats).run(should_stop) -__all__ = ['iter_image_paths'] +__all__ = ['WalkStats', 'iter_image_paths'] diff --git a/tests/curation/cross_project_leak/_route_tables.py b/tests/curation/cross_project_leak/_route_tables.py index cbb141f2c..096fcd17f 100644 --- a/tests/curation/cross_project_leak/_route_tables.py +++ b/tests/curation/cross_project_leak/_route_tables.py @@ -21,6 +21,7 @@ 'the sweep names no folder, so it is refused before a run is claimed; the run ' 'itself is covered in tests/integration/test_bulk_ingest_routes.py' ), + ('POST', '/ingest/bulk/preview'): 'read-only walk of a source; writes nothing', ('POST', '/ingest/bulk/{run_id}/cancel'): 'the sweep has no live run to cancel', ('POST', '/ingest/sources'): 'the sweep names no existing folder, so the source is refused', ('PATCH', '/ingest/sources/{source_id}'): 'the sweep has no such source (404)', diff --git a/tests/curation/test_bulk_ingest_preflight.py b/tests/curation/test_bulk_ingest_preflight.py new file mode 100644 index 000000000..bbd6abe96 --- /dev/null +++ b/tests/curation/test_bulk_ingest_preflight.py @@ -0,0 +1,298 @@ +"""Bulk-ingest preflight: what a folder, manifest or S3 prefix holds and what ingesting it +would cost, with no ingestion. The walk is bounded and cancellable, an unreadable root +refuses, and every estimate figure is a measured one from docs/PERFORMANCE.md.""" + +from __future__ import annotations + +import os +import re +import threading +import time +from pathlib import Path + +import pytest +from _fake_object_store import FakeObjectStore +from PIL import Image + +from src.services.curation import image_serving +from src.services.curation.bulk_ingest import preflight, preflight_rates +from src.services.curation.bulk_ingest.object_store import S3Config, S3NotAllowedError +from src.services.curation.bulk_ingest.source import BulkSource, InvalidSourceError, validate_source +from src.services.curation.bulk_ingest.walk import WalkStats, iter_image_paths + + +REPO = Path(__file__).resolve().parents[2] + + +@pytest.fixture(autouse=True) +def _roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + image_serving, + '_configured_roots', + lambda config=None: (tmp_path.resolve(),), # noqa: ARG005 + ) + + +def _img(path: Path, size: tuple[int, int] = (64, 48)) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + Image.new('RGB', size, (10, 20, 30)).save(path, format='JPEG') + + +def _folder(tmp_path: Path) -> Path: + root = tmp_path / 'photos' + for i in range(3): + _img(root / f'a{i}.jpg') + _img(root / 'sub' / 'b.JPG') + Image.new('RGB', (8, 8)).save(root / 'sub' / 'c.png') + (root / 'notes.txt').write_text('x') + return root + + +def _preview(source: BulkSource, **kw: object) -> preflight.PreviewResult: + return preflight.preview_source(source, **kw) # type: ignore[arg-type] + + +def test_folder_counts_bytes_and_extensions(tmp_path: Path) -> None: + root = _folder(tmp_path) + result = _preview(BulkSource('folder', str(root))) + assert result.files == 5 + assert result.total_bytes == sum( + p.stat().st_size for p in root.rglob('*') if p.suffix.lower() in {'.jpg', '.png'} + ) + assert {e: v.files for e, v in result.by_extension.items()} == {'.jpg': 4, '.png': 1} + assert result.partial is False + assert result.stop_reason is None + + +def test_symlinks_and_unreadable_directories_are_reported_not_hidden(tmp_path: Path) -> None: + root = _folder(tmp_path) + (root / 'link.jpg').symlink_to(root / 'a0.jpg') + locked = root / 'locked' + _img(locked / 'x.jpg') + locked.chmod(0) + try: + result = _preview(BulkSource('folder', str(root))) + finally: + locked.chmod(0o755) + assert result.files == 5 + assert result.symlinks_skipped == 1 + assert result.unreadable_dirs == 1 + assert result.unreadable_samples == [str(locked)] + + +def test_unreadable_root_refuses(tmp_path: Path) -> None: + root = _folder(tmp_path) + root.chmod(0) + try: + with pytest.raises(InvalidSourceError, match='not readable'): + _preview(BulkSource('folder', str(root))) + with pytest.raises(InvalidSourceError, match='not readable'): + validate_source(BulkSource('folder', str(root))) + finally: + root.chmod(0o755) + + +def test_a_root_outside_the_source_roots_refuses(tmp_path: Path) -> None: + with pytest.raises(InvalidSourceError, match='source root'): + _preview(BulkSource('folder', '/etc')) + + +def test_the_file_cap_stops_the_walk_and_says_partial(tmp_path: Path) -> None: + root = _folder(tmp_path) + result = _preview(BulkSource('folder', str(root)), limits=preflight.PreviewLimits(max_files=2)) + assert result.files == 2 + assert (result.partial, result.stop_reason) == (True, 'file_cap') + + +def test_the_time_cap_stops_the_walk(tmp_path: Path) -> None: + root = _folder(tmp_path) + ticks = iter(range(0, 1000, 10)) + result = _preview( + BulkSource('folder', str(root)), + limits=preflight.PreviewLimits(max_seconds=25), + clock=lambda: float(next(ticks)), + ) + assert (result.partial, result.stop_reason) == (True, 'time_cap') + assert result.files < 5 + + +def test_a_cancel_stops_the_walk(tmp_path: Path) -> None: + root = _folder(tmp_path) + result = _preview(BulkSource('folder', str(root)), should_stop=lambda: True) + assert (result.partial, result.stop_reason) == (True, 'cancelled') + + +def test_closing_the_walk_early_does_not_scan_the_rest_of_the_tree( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / 'wide' + for i in range(40): + _img(root / f'd{i}' / 'x.jpg') + scanned: list[str] = [] + real = os.scandir + + def counting(path: str | os.PathLike[str]) -> object: + scanned.append(str(path)) + return real(path) + + monkeypatch.setattr(os, 'scandir', counting) + walk = iter_image_paths(root, frozenset({'.jpg'}), workers=1) + next(walk) + walk.close() + assert len(scanned) < 41 + + +def test_a_stalled_directory_listing_cannot_hold_a_stopped_walk( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = _folder(tmp_path) + gate = threading.Event() + real = os.scandir + + def stalled(path: str | os.PathLike[str]) -> object: + gate.wait(10) + return real(path) + + monkeypatch.setattr(os, 'scandir', stalled) + started = time.monotonic() + try: + found = list(iter_image_paths(root, frozenset({'.jpg'}), 1, should_stop=lambda: True)) + assert found == [] + assert time.monotonic() - started < 3 + finally: + gate.set() + + +def test_walk_stats_count_what_the_walk_skips(tmp_path: Path) -> None: + root = _folder(tmp_path) + (root / 'l.jpg').symlink_to(root / 'a0.jpg') + stats = WalkStats() + list(iter_image_paths(root, frozenset({'.jpg'}), workers=2, stats=stats)) + assert stats.symlinks == 1 + assert stats.unreadable_dirs == 0 + + +def test_the_megapixel_sample_buckets_headers_without_decoding(tmp_path: Path) -> None: + root = tmp_path / 'mix' + for i in range(4): + _img(root / f'small{i}.jpg', (100, 100)) # 0.01 MP + for i in range(2): + _img(root / f'big{i}.jpg', (2400, 1600)) # 3.84 MP, above the 3.0 MP gate + result = _preview(BulkSource('folder', str(root)), gpu_decode=True, gpu_min_megapixels=3.0) + sample = result.megapixels + assert sample.sampled == 6 + assert sample.unreadable == 0 + assert sample.share_at_or_above_gate == pytest.approx(2 / 6) + assert sum(sample.histogram.values()) == 6 + + +def test_the_sample_is_bounded(tmp_path: Path) -> None: + root = tmp_path / 'many' + for i in range(30): + _img(root / f'{i}.jpg') + limits = preflight.PreviewLimits(header_sample=10) + first = _preview(BulkSource('folder', str(root)), limits=limits) + second = _preview(BulkSource('folder', str(root)), limits=limits) + assert first.megapixels.sampled == 10 + assert first.megapixels == second.megapixels + + +def test_estimate_uses_the_measured_rates_and_footprint_only(tmp_path: Path) -> None: + root = tmp_path / 'm' + for i in range(4): + _img(root / f's{i}.jpg', (100, 100)) + rates = {r.route: r for r in preflight_rates.RATES} + result = _preview(BulkSource('folder', str(root)), gpu_decode=True, gpu_min_megapixels=3.0) + est = result.estimate + assert est.basis == 'megapixel_sample' + assert ( + est.duration_s_min + == est.duration_s_max + == pytest.approx(4 / rates['cpu_small'].images_per_s) + ) + assert est.vectors == pytest.approx(4 * preflight_rates.EMBEDDINGS_PER_IMAGE.value) + assert est.opensearch_bytes == round(4 * preflight_rates.OPENSEARCH_BYTES_PER_IMAGE.value) + + +def test_large_photos_use_the_cpu_rate_when_gpu_decode_is_off(tmp_path: Path) -> None: + root = tmp_path / 'm' + _img(root / 'big.jpg', (2400, 1600)) + rates = {r.route: r for r in preflight_rates.RATES} + on = _preview(BulkSource('folder', str(root)), gpu_decode=True, gpu_min_megapixels=3.0) + off = _preview(BulkSource('folder', str(root)), gpu_decode=False, gpu_min_megapixels=3.0) + assert on.estimate.duration_s_min == pytest.approx(1 / rates['gpu_large'].images_per_s) + assert off.estimate.duration_s_min == pytest.approx(1 / rates['cpu_large'].images_per_s) + + +def test_without_headers_the_estimate_is_a_range_across_the_measured_routes() -> None: + est = preflight_rates.estimate(files=100, share_large=None, gpu_decode=True) + rates = [r.images_per_s for r in preflight_rates.RATES] + assert est.basis == 'rate_range' + assert est.duration_s_min == pytest.approx(100 / max(rates)) + assert est.duration_s_max == pytest.approx(100 / min(rates)) + + +def test_every_rate_and_footprint_figure_is_in_performance_md() -> None: + text = (REPO / 'docs' / 'PERFORMANCE.md').read_text(encoding='utf-8') + figures = [r.images_per_s for r in preflight_rates.RATES] + [ + preflight_rates.EMBEDDINGS_PER_IMAGE.value, + preflight_rates.OPENSEARCH_BYTES_PER_IMAGE.value, + ] + for value in figures: + shown = str(value) if isinstance(value, int) else f'{value:.1f}' + assert re.search(rf'(? None: + root = _folder(tmp_path) + manifest = tmp_path / 'list.txt' + manifest.write_text( + '\n'.join( + [ + '# comment', + str(root / 'a0.jpg'), + str(root / 'a1.jpg'), + str(root / 'gone.jpg'), + '/etc/passwd.jpg', + '', + ] + ) + ) + result = _preview(BulkSource('manifest', str(manifest))) + assert result.files == 2 + assert result.missing == 1 + assert result.outside_roots == 1 + + +def _s3_store() -> FakeObjectStore: + store = FakeObjectStore(page_size=2) + for i in range(7): + store.put('b', f'p/{i}.jpg', b'x' * 10) + store.put('b', 'p/readme.txt', b'no') + return store + + +def test_s3_prefix_is_counted_by_listing_only() -> None: + store = _s3_store() + config = S3Config(allowed=(('b', 'p/'),)) + result = preflight.preview_s3(store, config, 'b', 'p/', ('.jpg',)) + assert (result.files, result.total_bytes) == (7, 70) + assert result.megapixels.sampled == 0 + assert result.estimate.basis == 'rate_range' + assert store.get_calls == [] + + +def test_s3_listing_is_capped_and_outside_the_allow_list_refuses() -> None: + config = S3Config(allowed=(('b', 'p/'),)) + capped = preflight.preview_s3( + _s3_store(), config, 'b', 'p/', ('.jpg',), limits=preflight.PreviewLimits(max_files=3) + ) + assert (capped.partial, capped.stop_reason) == (True, 'file_cap') + with pytest.raises(S3NotAllowedError): + preflight.preview_s3(_s3_store(), config, 'other', 'p/', ('.jpg',)) diff --git a/tests/integration/test_bulk_ingest_cli.py b/tests/integration/test_bulk_ingest_cli.py index 07b0d2fb6..8e10f598d 100644 --- a/tests/integration/test_bulk_ingest_cli.py +++ b/tests/integration/test_bulk_ingest_cli.py @@ -131,3 +131,45 @@ def test_status_and_cancel_address_a_run_by_id( assert capsys.readouterr().out.strip() assert cli.main(_args('--status', 'bulk_20260101T000000_deadbeef')) == 2 assert '404' in capsys.readouterr().err + + +def test_preview_prints_the_counts_and_the_estimate_and_ingests_nothing( + cli: Any, client: TestClient, served_root: Path, capsys: pytest.CaptureFixture[str] +) -> None: + _write(served_root / 'imgs', [f'{i}.jpg' for i in range(6)]) + + code = cli.main(_args('--folder', str(served_root / 'imgs'), '--preview')) + + out = capsys.readouterr().out + assert code == 0 + assert 'files: 6' in out + assert '.jpg' in out + assert 'estimated ingest time' in out + assert 'docs/PERFORMANCE.md' in out + assert client.get('/curation/projects/default/ingest/bulk').json()['total'] == 0 + + +def test_preview_of_a_refused_source_exits_two( + cli: Any, served_root: Path, capsys: pytest.CaptureFixture[str] +) -> None: + assert cli.main(_args('--folder', str(served_root / 'missing'), '--preview')) == 2 + assert 'not a directory' in capsys.readouterr().err + + +def test_preview_cannot_be_combined_with_status(cli: Any) -> None: + with pytest.raises(SystemExit): + cli.main(_args('--status', 'bulk_x', '--preview')) + + +@pytest.mark.parametrize( + ('low', 'high', 'shown'), + [ + (0.34, 0.34, 'about 0.3 s'), + (69.0, 69.0, 'about 1m09s'), + (3600.0, 7200.0, '1h00m to 2h00m'), + ], +) +def test_the_estimate_is_shown_in_readable_units( + cli: Any, low: float, high: float, shown: str +) -> None: + assert cli._fmt_duration(low, high) == shown diff --git a/tests/integration/test_bulk_ingest_routes.py b/tests/integration/test_bulk_ingest_routes.py index a3ca01c36..88976c309 100644 --- a/tests/integration/test_bulk_ingest_routes.py +++ b/tests/integration/test_bulk_ingest_routes.py @@ -257,3 +257,55 @@ def test_an_unknown_project_is_refused_before_anything_runs( ) assert resp.status_code == 404 assert resp.json()['detail']['error'] == 'project_not_found' + + +def test_config_serves_the_preview_bounds_and_the_measured_figures(client: TestClient) -> None: + preview = client.get(f'{BASE}/config').json()['preview'] + assert preview['max_files'] > 0 + assert preview['max_seconds'] > 0 + assert preview['header_sample'] > 0 + assert {r['route'] for r in preview['rates']} == {'cpu_small', 'cpu_large', 'gpu_large'} + assert all(r['source_section'] for r in preview['rates']) + assert preview['embeddings_per_image']['value'] == 7.4 + assert preview['source_document'] == 'docs/PERFORMANCE.md' + + +def test_preview_reports_a_folder_and_writes_nothing( + client: TestClient, fake_os: FakeOpenSearch, served_root: Path +) -> None: + _write_images(served_root / 'a', [f'{i}.jpg' for i in range(4)]) + (served_root / 'a' / 'notes.txt').write_text('x') + res = client.post(f'{BASE}/preview', json={'folder': str(served_root / 'a')}) + assert res.status_code == 200, res.text + body = res.json() + assert (body['kind'], body['files'], body['partial']) == ('folder', 4, False) + assert body['by_extension']['.jpg']['files'] == 4 + assert body['megapixels']['sampled'] == 4 + assert body['estimate']['basis'] == 'megapixel_sample' + assert body['estimate']['opensearch_bytes'] > 0 + assert fake_os.images == {} + assert client.get(BASE).json()['total'] == 0 + + +def test_preview_refuses_a_folder_outside_the_source_roots(client: TestClient) -> None: + res = client.post(f'{BASE}/preview', json={'folder': '/etc'}) + assert res.status_code == 422 + assert res.json()['detail']['error'] == 'bulk_ingest_invalid_source' + + +def test_preview_needs_exactly_one_source(client: TestClient) -> None: + assert client.post(f'{BASE}/preview', json={}).status_code == 422 + two = {'folder': '/a', 'manifest': '/b'} + assert client.post(f'{BASE}/preview', json=two).status_code == 422 + + +def test_preview_of_s3_is_refused_while_s3_is_off(client: TestClient) -> None: + res = client.post(f'{BASE}/preview', json={'s3': {'bucket': 'b', 'prefix': 'p/'}}) + assert res.status_code == 503 + assert res.json()['detail']['error'] == 's3_not_configured' + + +def test_preview_of_an_unknown_source_is_a_404(client: TestClient) -> None: + res = client.post(f'{BASE}/preview', json={'source_id': 'nope_nope'}) + assert res.status_code == 404 + assert res.json()['detail']['error'] == 'ingest_source_not_found' diff --git a/tests/test_monitoring_metrics.py b/tests/test_monitoring_metrics.py index 4ed8add35..7cf783d3c 100644 --- a/tests/test_monitoring_metrics.py +++ b/tests/test_monitoring_metrics.py @@ -246,6 +246,19 @@ def _alert_exprs(path: Path) -> list[str]: return exprs +def _grafana_alert_exprs(path: Path) -> list[tuple[str, str]]: + """``(rule title, expr)`` of every query of every Grafana-provisioned rule (the + queries sit under ``data[].model.expr``, not at the rule's top level).""" + data = yaml.safe_load(path.read_text()) + return [ + (rule['title'], query['model']['expr']) + for group in data.get('groups', []) + for rule in group.get('rules', []) + for query in rule.get('data', []) + if query.get('model', {}).get('expr') + ] + + def test_fixture_is_nonempty_and_disjoint_from_promql_keywords() -> None: assert REAL_METRIC_NAMES assert not (REAL_METRIC_NAMES & _PROMQL_KEYWORDS) @@ -313,3 +326,46 @@ def test_every_dashboard_is_provisionable_and_linked() -> None: assert data['panels'], path.name assert 'openprocessor' in data.get('tags', []), path.name assert any(link.get('type') == 'dashboards' for link in data.get('links', [])), path.name + + +GRAFANA_ALERTING_PATH = REPO_ROOT / 'monitoring' / 'grafana-alerting.yml' + + +def test_grafana_alert_rules_query_only_real_metrics() -> None: + """The Grafana rules were never checked: Triton Model Not Ready, High GPU + Temperature, the error-rate rule (``http_requests_total``) and the OpenSearch rule + (``opensearch_cluster_status``) all read series nothing exports, so none could fire.""" + rules = _grafana_alert_exprs(GRAFANA_ALERTING_PATH) + assert rules + unknown = { + f'{title}: {expr.strip()}': bad + for title, expr in rules + if (bad := _metric_names_in_expr(expr) - REAL_METRIC_NAMES) + } + assert not unknown, f'Grafana rule(s) query metric(s) nothing exports: {unknown}' + + +def test_the_gpu_temperature_rule_reads_dcgm() -> None: + rules = dict(_grafana_alert_exprs(GRAFANA_ALERTING_PATH)) + assert 'DCGM_FI_DEV_GPU_TEMP' in rules['High GPU Temperature'] + assert 'Triton Model Not Ready' not in rules + + +def test_the_grafana_folder_is_named_for_this_product() -> None: + data = yaml.safe_load(GRAFANA_ALERTING_PATH.read_text()) + assert {g['folder'] for g in data['groups']} == {'OpenProcessor Alerts'} + + +def test_a_contact_point_that_cannot_deliver_says_so_in_its_name() -> None: + """The shipped receivers post to a localhost address nothing listens on, so nobody is + notified until the operator replaces them; the name must not suggest otherwise.""" + data = yaml.safe_load(GRAFANA_ALERTING_PATH.read_text()) + points = {p['name']: p for p in data['contactPoints']} + for name, point in points.items(): + for receiver in point['receivers']: + url = receiver.get('settings', {}).get('url', '') + if 'localhost' in url or '127.0.0.1' in url: + assert name.startswith('unconfigured-'), name + policy = data['policies'][0] + routed = {policy['receiver'], *(r['receiver'] for r in policy['routes'])} + assert routed <= set(points)