diff --git a/docker-compose.publish.yml b/docker-compose.publish.yml index c54bace2..ead6fb3b 100644 --- a/docker-compose.publish.yml +++ b/docker-compose.publish.yml @@ -54,6 +54,7 @@ services: restart: unless-stopped volumes: - staging-data:/usr/src/app/staging_data:ro + - fswatch-state:/var/lib/tagbase-fswatch nginx: image: ghcr.io/tagbase/tagbase-nginx:${TAGBASE_VERSION:-latest} @@ -317,5 +318,6 @@ volumes: lego-data: postgis-data: staging-data: + fswatch-state: dbbackups: tagbase-logs: diff --git a/docker-compose.test.yml b/docker-compose.test.yml index c7e639cd..380f6fb2 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -100,6 +100,7 @@ services: - TAGBASE_INGEST_BASE=tagbase_server:5433/tagbase/api/v0 volumes: - ./staging_data:/usr/src/app/staging_data:ro + - fswatch-state:/var/lib/tagbase-fswatch networks: - test-network @@ -195,6 +196,7 @@ networks: test-network: volumes: + fswatch-state: tagbase-test-pgdata: tagbase-test-alloy: tagbase-test-prometheus: diff --git a/docker-compose.yml b/docker-compose.yml index 0c7527a3..3356a693 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -59,6 +59,7 @@ services: restart: unless-stopped volumes: - ./staging_data:/usr/src/app/staging_data:ro + - fswatch-state:/var/lib/tagbase-fswatch nginx: build: context: ./services/nginx @@ -375,6 +376,7 @@ networks: internal-network: volumes: + fswatch-state: alloy-data: prometheus-data: loki-data: diff --git a/openapi.yaml b/openapi.yaml index 7e3b1146..fbe111a7 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -70,20 +70,24 @@ paths: - $ref: "#/components/parameters/version" requestBody: content: + application/gzip: + schema: + $ref: "#/components/schemas/SyncUpload" application/octet-stream: schema: - description: Compressed binary e.g. zip file containing one or more eTUFF files - format: binary - maxLength: 1000000000 - minLength: 1 - type: string + $ref: "#/components/schemas/SyncUpload" + application/x-gzip: + schema: + $ref: "#/components/schemas/SyncUpload" + application/x-tar: + schema: + $ref: "#/components/schemas/SyncUpload" + application/zip: + schema: + $ref: "#/components/schemas/SyncUpload" text/plain: schema: - description: Plain text eTUFF file - format: binary - maxLength: 1000000000 - minLength: 1 - type: string + $ref: "#/components/schemas/SyncUpload" required: true x-body-name: body responses: @@ -95,6 +99,8 @@ paths: description: A success message confirming ingestion. "400": $ref: "#/components/responses/BadRequest" + "413": + $ref: "#/components/responses/PayloadTooLarge" "500": $ref: "#/components/responses/InternalError" summary: Post a local file and perform a ingest operation @@ -102,6 +108,79 @@ paths: - ingest x-codegen-request-body-name: body x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs: + post: + description: > + Accept a .txt, .zip, .tar.gz, or .tgz upload up to 2 GiB, store it, + and enqueue ingest. Poll GET /ingest/jobs/{job_id} until the job + succeeds or fails. + operationId: create_ingest_job + parameters: + - $ref: "#/components/parameters/filename" + - $ref: "#/components/parameters/notes" + - $ref: "#/components/parameters/type" + - $ref: "#/components/parameters/version" + requestBody: + content: + application/gzip: + schema: + $ref: "#/components/schemas/JobUpload" + application/octet-stream: + schema: + $ref: "#/components/schemas/JobUpload" + application/x-gzip: + schema: + $ref: "#/components/schemas/JobUpload" + application/x-tar: + schema: + $ref: "#/components/schemas/JobUpload" + application/zip: + schema: + $ref: "#/components/schemas/JobUpload" + text/plain: + schema: + $ref: "#/components/schemas/JobUpload" + required: true + x-body-name: body + responses: + "202": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestAccepted" + description: Request accepted. Ingest job enqueued + "400": + $ref: "#/components/responses/BadRequest" + "413": + $ref: "#/components/responses/PayloadTooLarge" + "500": + $ref: "#/components/responses/InternalError" + summary: Enqueue a large file for ingest + tags: + - ingest + x-codegen-request-body-name: body + x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs/{job_id}: + parameters: + - $ref: "#/components/parameters/jobId" + get: + description: Read the status of one ingest job. + operationId: get_ingest_job + responses: + "200": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestJob" + description: The current ingest job. + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + summary: Read one ingest job + tags: + - ingest + x-openapi-router-controller: tagbase_server.controllers.ingest_controller /tags: delete: description: Delete all tags @@ -205,6 +284,18 @@ components: application/problem+json: schema: $ref: "#/components/schemas/Problem" + NotFound: + description: The requested resource does not exist. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" + PayloadTooLarge: + description: The request body exceeds the limit for this operation. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" InternalError: description: Unexpected server error. Contact admin detailed in openapi.yaml. content: @@ -212,6 +303,19 @@ components: schema: $ref: "#/components/schemas/Problem" parameters: + jobId: + description: Ingest job id + explode: false + in: path + name: job_id + required: true + schema: + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + type: string + style: simple filename: description: Free-form text field to explicitly define the name of the file to be persisted explode: true @@ -398,6 +502,91 @@ components: example: 4bf92f3577b34da6a3ce929d0e0e4736 title: Ingest200 type: object + SyncUpload: + description: Synchronous ingest body, at most 100 MiB + format: binary + maxLength: 104857600 + minLength: 1 + type: string + JobUpload: + description: Ingest job body, at most 2 GiB + format: binary + maxLength: 2147483648 + minLength: 1 + type: string + IngestAccepted: + description: HTTP 202 response after an ingest job is queued + required: + - id + - status + example: + id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: queued + properties: + id: + description: Ingest job id + format: uuid + type: string + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: + description: Job status at accept time + type: string + minLength: 6 + maxLength: 9 + pattern: ^queued$ + example: queued + title: IngestAccepted + type: object + IngestJob: + description: Status of one ingest job + required: + - id + - status + - files_done + example: + id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: succeeded + files_done: 1 + files_total: 1 + properties: + id: + description: Ingest job id + format: uuid + type: string + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: + description: queued, running, succeeded, or failed + type: string + minLength: 6 + maxLength: 9 + pattern: ^(queued|running|succeeded|failed)$ + example: succeeded + error: + description: Present when the job failed + type: string + minLength: 1 + maxLength: 10000 + pattern: ^(?!\s*$).+$ + files_total: + description: eTUFF members in the upload + type: integer + format: int32 + minimum: 0 + maximum: 10000 + files_done: + description: eTUFF members already processed + type: integer + format: int32 + minimum: 0 + maximum: 10000 + title: IngestJob + type: object Metadata: description: Contains the ingested tag metadata consistent with the eTUFF specification example: diff --git a/services/docker-cron/crontab b/services/docker-cron/crontab index 89ade87e..17773a8f 100755 --- a/services/docker-cron/crontab +++ b/services/docker-cron/crontab @@ -1 +1,2 @@ */1 * * * * tagbase . /usr/src/app/.env && /usr/local/bin/python3 /usr/src/app/poll_metadata_and_obs_types.py >/proc/1/fd/1 2>/proc/1/fd/2 +15 3 * * * tagbase . /usr/src/app/.env && /usr/local/bin/python3 /usr/src/app/purge_ingest_jobs.py >/proc/1/fd/1 2>/proc/1/fd/2 diff --git a/services/docker-cron/entrypoint.sh b/services/docker-cron/entrypoint.sh index 3b72c8f1..8acec710 100755 --- a/services/docker-cron/entrypoint.sh +++ b/services/docker-cron/entrypoint.sh @@ -8,6 +8,7 @@ POSTGRES_PORT="${POSTGRES_PORT:-5432}" cat >/usr/src/app/.env <"$DEDUPE_STATE" 2>/dev/null; then - DEDUPE_STATE="/tmp/fswatch-ingest-dedupe-$$" - : >"$DEDUPE_STATE" +if [ -n "${DEDUPE_STATE:-}" ]; then + STATE_DIR=$(dirname "$DEDUPE_STATE") + DONE_FILE=$DEDUPE_STATE +else + STATE_DIR=${FSWATCH_STATE_DIR:-/var/lib/tagbase-fswatch} + DONE_FILE=$STATE_DIR/done.tsv fi +mkdir -p "$STATE_DIR" +PENDING_FILE=$STATE_DIR/pending.tsv +touch "$PENDING_FILE" "$DONE_FILE" -echo "Watching $PATH_TO_CHECK (dedupe=$DEDUPE_STATE)" +echo "Watching $PATH_TO_CHECK (state=$STATE_DIR)" -# Detect GNU vs BSD stat once (not per-file — missing files must not flip dialects). if stat -c%s / >/dev/null 2>&1; then file_size() { [ -f "$1" ] || return 1 @@ -27,12 +32,7 @@ else fi urlencode() { - # Prefer python3 (available in fswatch image); fall back to raw name. - if command -v python3 >/dev/null 2>&1; then - python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=""))' "$1" - else - printf '%s' "$1" - fi + python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=""))' "$1" } should_ingest() { @@ -72,79 +72,149 @@ wait_stable() { return 1 } -already_ingested() { +already_done() { + path=$1 + size=$2 + awk -F '\t' -v path="$path" -v size="$size" '$1 == path && $2 == size { found = 1 } END { exit !found }' "$DONE_FILE" +} + +already_pending() { + path=$1 + awk -F '\t' -v path="$path" '$1 == path { found = 1 } END { exit !found }' "$PENDING_FILE" +} + +record_pending() { + path=$1 + size=$2 + job_id=$3 + awk -F '\t' -v path="$path" '$1 != path { print }' "$PENDING_FILE" >"$STATE_DIR/.pending.next" + printf '%s\t%s\t%s\n' "$path" "$size" "$job_id" >>"$STATE_DIR/.pending.next" + mv "$STATE_DIR/.pending.next" "$PENDING_FILE" +} + +drop_pending() { + path=$1 + awk -F '\t' -v path="$path" '$1 != path { print }' "$PENDING_FILE" >"$STATE_DIR/.pending.next" + mv "$STATE_DIR/.pending.next" "$PENDING_FILE" +} + +record_done() { path=$1 size=$2 - [ -f "$DEDUPE_STATE" ] || return 1 - { - read -r last_path - read -r last_size - } <"$DEDUPE_STATE" || return 1 - [ "$path" = "$last_path" ] && [ "$size" = "$last_size" ] + drop_pending "$path" + awk -F '\t' -v path="$path" '$1 != path { print }' "$DONE_FILE" >"$STATE_DIR/.done.next" + printf '%s\t%s\n' "$path" "$size" >>"$STATE_DIR/.done.next" + mv "$STATE_DIR/.done.next" "$DONE_FILE" +} + +json_field() { + python3 -c 'import json,sys; print(json.load(sys.stdin).get(sys.argv[1], "") or "")' "$1" } -ingest_file() { +submit_file() { line=$1 filename=${line##*/} if ! should_ingest "$filename"; then echo "Skipping non-ingestible: $filename" return 0 fi - echo "Contents of $PATH_TO_CHECK changed; Checking: $filename" if ! wait_stable "$line"; then echo "Gave up waiting for stable file: $filename" return 0 fi size=$(file_size "$line") || return 0 - if already_ingested "$line" "$size"; then - echo "Already ingested at this size, skipping: $filename" + if already_pending "$line" || already_done "$line" "$size"; then + echo "Already tracked, skipping: $filename" return 0 fi - echo "Processing: $filename" + echo "Submitting: $filename" enc=$(urlencode "$filename") ctype=application/octet-stream case "$filename" in *.txt) ctype=text/plain ;; esac - if ! curl -sS -f -X POST \ + body_file=$(mktemp "$STATE_DIR/.body.XXXXXX") + code=$(curl -sS --max-time "$CURL_MAX_TIME" -o "$body_file" -w '%{http_code}' -X POST \ -H "accept: application/json" \ -H "Content-Type: ${ctype}" \ -T "$line" \ - "${TAGBASE_INGEST_BASE}/ingest?filename=${enc}&type=etuff"; then - echo "Ingest failed for $filename (see tagbase_server logs)" - else - printf '%s\n%s\n' "$line" "$size" >"$DEDUPE_STATE" - echo - echo "Ingest OK: $filename" + "${TAGBASE_INGEST_BASE}/ingest/jobs?filename=${enc}&type=etuff" || true) + if [ "$code" != "202" ]; then + echo "Ingest accept failed for $filename (HTTP ${code})" + rm -f "$body_file" + return 0 + fi + job_id=$(json_field id <"$body_file" || true) + rm -f "$body_file" + if [ -z "$job_id" ]; then + echo "Ingest accept for $filename did not return a job id" + return 0 fi + record_pending "$line" "$size" "$job_id" + echo "Queued $filename as $job_id" +} + +reconcile_pending() { + [ -s "$PENDING_FILE" ] || return 0 + snapshot=$(mktemp "$STATE_DIR/.pending.XXXXXX") + cp "$PENDING_FILE" "$snapshot" + while IFS=$'\t' read -r path size job_id; do + [ -n "${job_id:-}" ] || continue + body_file=$(mktemp "$STATE_DIR/.status.XXXXXX") + code=$(curl -sS --max-time "$CURL_MAX_TIME" -o "$body_file" -w '%{http_code}' \ + -H "accept: application/json" \ + "${TAGBASE_INGEST_BASE}/ingest/jobs/${job_id}" || true) + if [ "$code" != "200" ]; then + echo "Job status request failed for $job_id (HTTP ${code}); keeping pending" + rm -f "$body_file" + continue + fi + status=$(json_field status <"$body_file" || true) + if [ "$status" = "succeeded" ]; then + record_done "$path" "$size" + echo "Ingest OK: ${path##*/}" + elif [ "$status" = "failed" ]; then + echo "Job $job_id failed: $(cat "$body_file")" + drop_pending "$path" + fi + rm -f "$body_file" + done <"$snapshot" + rm -f "$snapshot" } scan_existing() { - # Process files already present (startup / leftover from a prior blind window). - # shellcheck disable=SC2039 for path in "$PATH_TO_CHECK"*; do [ -f "$path" ] || continue - ingest_file "$path" + submit_file "$path" done } scan_existing -# Continuous watch: keep receiving events while wait_stable/curl run for one file. -# Line-buffer fswatch stdout when piped (otherwise events can sit in libc buffer). -# Outer loop only restarts fswatch if the binary exits unexpectedly. -FSWATCH_CMD="fswatch" +FSWATCH_CMD=(fswatch) if command -v stdbuf >/dev/null 2>&1; then - FSWATCH_CMD="stdbuf -oL fswatch" + FSWATCH_CMD=(stdbuf -oL fswatch) fi +FIFO=$(mktemp -u "$STATE_DIR/.events.XXXXXX") +mkfifo "$FIFO" +"${FSWATCH_CMD[@]}" "$PATH_TO_CHECK" \ + --event Created --event MovedTo --event IsFile --event Updated >"$FIFO" & +watch_pid=$! +cleanup() { + kill "$watch_pid" >/dev/null 2>&1 || true + rm -f "$FIFO" +} +trap cleanup EXIT + +exec 3<>"$FIFO" while true; do - # shellcheck disable=SC2086 - $FSWATCH_CMD "$PATH_TO_CHECK" \ - --event Created --event MovedTo --event IsFile --event Updated | - while IFS= read -r line; do - ingest_file "$line" + reconcile_pending + if IFS= read -r -t "$RECONCILE_SECONDS" -u 3 line; then + submit_file "$line" + # bash 5 `read -t 0` repeats the previous line and never consumes the fifo. + while IFS= read -r -t 1 -u 3 line; do + submit_file "$line" done - echo "fswatch exited; restarting watch on $PATH_TO_CHECK" - sleep 1 + fi done diff --git a/services/nginx/config/nginx.conf b/services/nginx/config/nginx.conf index 95fc13fc..ef86a631 100644 --- a/services/nginx/config/nginx.conf +++ b/services/nginx/config/nginx.conf @@ -76,6 +76,42 @@ http { proxy_redirect off; } + error_page 413 @payload_too_large; + location @payload_too_large { + internal; + default_type application/problem+json; + return 413 '{"type":"urn:tagbase:problem:http-error","title":"Payload Too Large","status":413,"detail":"Request body exceeds the configured limit.","trace_id":"00000000000000000000000000000000"}'; + } + + # Longer prefix than /tagbase/api/v0/ingest. Stream the body; idle + # gaps still die after 60s, while an active upload does not. + location /tagbase/api/v0/ingest/jobs { + client_max_body_size 2048m; + client_body_timeout 60s; + proxy_http_version 1.1; + proxy_request_buffering off; + proxy_read_timeout 60s; + proxy_send_timeout 60s; + proxy_pass http://tagbase_server; + proxy_set_header Host $host; + proxy_set_header Connection ""; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Scheme $scheme; + proxy_set_header X-Forwarded-Path /tagbase; + } + + location /tagbase/api/v0/ingest { + client_max_body_size 100m; + client_body_timeout 60s; + proxy_read_timeout 60s; + proxy_send_timeout 60s; + proxy_pass http://tagbase_server; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Scheme $scheme; + proxy_set_header X-Forwarded-Path /tagbase; + } + location /tagbase { client_max_body_size 1000M; proxy_read_timeout 60s; diff --git a/services/nginx/config/nginx.test.conf b/services/nginx/config/nginx.test.conf index aca35f3e..6c7c3c73 100644 --- a/services/nginx/config/nginx.test.conf +++ b/services/nginx/config/nginx.test.conf @@ -63,6 +63,40 @@ http { add_header Access-Control-Allow-Origin $http_origin; } + error_page 413 @payload_too_large; + location @payload_too_large { + internal; + default_type application/problem+json; + return 413 '{"type":"urn:tagbase:problem:http-error","title":"Payload Too Large","status":413,"detail":"Request body exceeds the configured limit.","trace_id":"00000000000000000000000000000000"}'; + } + + location /tagbase/api/v0/ingest/jobs { + client_max_body_size 2048m; + client_body_timeout 60s; + proxy_http_version 1.1; + proxy_request_buffering off; + proxy_read_timeout 60s; + proxy_send_timeout 60s; + proxy_pass http://tagbase_server; + proxy_set_header Host $host; + proxy_set_header Connection ""; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Scheme $scheme; + proxy_set_header X-Forwarded-Path /tagbase; + } + + location /tagbase/api/v0/ingest { + client_max_body_size 100m; + client_body_timeout 60s; + proxy_read_timeout 60s; + proxy_send_timeout 60s; + proxy_pass http://tagbase_server; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Scheme $scheme; + proxy_set_header X-Forwarded-Path /tagbase; + } + location /tagbase { client_max_body_size 1000M; proxy_read_timeout 60s; diff --git a/services/nginx/proxy/index.html b/services/nginx/proxy/index.html index 97af1206..e7e22efc 100644 --- a/services/nginx/proxy/index.html +++ b/services/nginx/proxy/index.html @@ -26,8 +26,11 @@

Navigation

instances.
  • - /tagbase/api/v0: REST API base path. + /tagbase/api/v0/ui/: interactive REST API.
  • /upload.html

    ← Home

    Upload data

    -

    Select eTUFF text files or supported compressed files to ingest.

    +

    + Select eTUFF .txt, .zip, .tar.gz, + or .tgz files. Large files are queued and this page follows + them until the server finishes. +

    - +
    + + + +
    +
    +
    +
    Upload data const fileName = document.querySelector("#file-name"); const status = document.querySelector("#status"); const submit = document.querySelector("#submit"); + const abortButton = document.querySelector("#abort"); + const stopMonitor = document.querySelector("#stop-monitor"); + const progressWrap = document.querySelector("#progress-wrap"); + const progress = document.querySelector("#progress"); + const STORAGE_KEY = "tagbase-ingest-jobs"; + const POLL_MS = 3000; + let activeXhr = null; + let pollTimer = null; function showStatus(type, message) { status.className = `alert alert-${type} mt-4`; @@ -136,6 +178,116 @@

    Upload data

    ); fileInput.addEventListener("change", updateFileListDisplay); + function loadJobs() { + try { + return JSON.parse(localStorage.getItem(STORAGE_KEY) || "[]"); + } catch { + return []; + } + } + + function saveJobs(jobs) { + localStorage.setItem(STORAGE_KEY, JSON.stringify(jobs)); + stopMonitor.classList.toggle("d-none", jobs.length === 0); + } + + function rememberJob(job) { + const jobs = loadJobs().filter((item) => item.id !== job.id); + jobs.push(job); + saveJobs(jobs); + } + + function forgetJob(id) { + saveJobs(loadJobs().filter((item) => item.id !== id)); + } + + function postFile(file, params) { + return new Promise((resolve, reject) => { + const xhr = new XMLHttpRequest(); + activeXhr = xhr; + xhr.open("POST", `/tagbase/api/v0/ingest/jobs?${params}`); + xhr.setRequestHeader( + "Content-Type", + file.type || "application/octet-stream", + ); + xhr.setRequestHeader("Accept", "application/json"); + xhr.upload.onprogress = (event) => { + if (!event.lengthComputable) return; + const pct = Math.round((event.loaded / event.total) * 100); + progress.style.width = `${pct}%`; + progress.textContent = `${pct}%`; + }; + xhr.onload = () => { + activeXhr = null; + if (xhr.status === 202) { + resolve(JSON.parse(xhr.responseText)); + return; + } + reject(new Error(`${xhr.status}\n${xhr.responseText}`)); + }; + xhr.onerror = () => { + activeXhr = null; + reject(new Error("upload failed")); + }; + xhr.onabort = () => { + activeXhr = null; + reject(new Error("upload aborted")); + }; + progressWrap.classList.remove("d-none"); + abortButton.classList.remove("d-none"); + xhr.send(file); + }); + } + + async function pollOnce() { + const jobs = loadJobs(); + if (!jobs.length) { + stopMonitor.classList.add("d-none"); + return; + } + const lines = []; + for (const job of jobs) { + try { + const response = await fetch( + `/tagbase/api/v0/ingest/jobs/${job.id}`, + { headers: { Accept: "application/json" } }, + ); + const body = await response.json(); + if (body.status === "succeeded") { + forgetJob(job.id); + lines.push(`✓ ${job.filename}: succeeded`); + } else if (body.status === "failed") { + forgetJob(job.id); + lines.push(`✗ ${job.filename}: ${body.error || "failed"}`); + } else { + lines.push(`… ${job.filename}: ${body.status}`); + } + } catch (error) { + lines.push(`… ${job.filename}: ${error.message}`); + } + } + const remaining = loadJobs(); + showStatus( + remaining.length ? "info" : "success", + lines.join("\n") || "No jobs are being monitored.", + ); + } + + function ensurePolling() { + if (pollTimer) return; + pollTimer = window.setInterval(pollOnce, POLL_MS); + pollOnce(); + } + + abortButton.addEventListener("click", () => { + if (activeXhr) activeXhr.abort(); + }); + + stopMonitor.addEventListener("click", () => { + saveJobs([]); + showStatus("info", "Stopped monitoring. The server job keeps running."); + }); + form.addEventListener("submit", async (event) => { event.preventDefault(); const files = Array.from(fileInput.files); @@ -143,69 +295,39 @@

    Upload data

    const version = document.querySelector("#version").value.trim(); const notes = document.querySelector("#notes").value.trim(); - submit.disabled = true; - showStatus("info", `Uploading ${files.length} file(s)…`); - const results = []; - let successCount = 0; for (let i = 0; i < files.length; i++) { const file = files[i]; const params = new URLSearchParams({ filename: file.name }); if (version) params.set("version", version); if (notes) params.set("notes", notes); - showStatus( "info", `Uploading file ${i + 1} of ${files.length}: ${file.name}…`, ); - try { - const response = await fetch(`/tagbase/api/v0/ingest?${params}`, { - method: "POST", - headers: { - "Content-Type": file.type || "application/octet-stream", - }, - body: file, - }); - const text = await response.text(); - let result; - try { - result = JSON.stringify(JSON.parse(text), null, 2); - } catch { - result = text; - } - - if (!response.ok) - throw new Error( - `${response.status} ${response.statusText}\n${result}`, - ); - - successCount++; - results.push(`✓ ${file.name}: ${result}`); + const accepted = await postFile(file, params); + rememberJob({ id: accepted.id, filename: file.name }); + results.push(`queued ${file.name} (${accepted.id})`); } catch (error) { results.push(`✗ ${file.name}: ${error.message}`); + } finally { + abortButton.classList.add("d-none"); + progressWrap.classList.add("d-none"); + progress.style.width = "0%"; } } - if (successCount === files.length) { - showStatus( - "success", - `All ${files.length} uploads completed.\n\n` + results.join("\n\n"), - ); - form.reset(); - fileName.textContent = ""; - } else { - showStatus( - "warning", - `Completed with errors (${successCount}/${files.length} succeeded):\n\n` + - results.join("\n\n"), - ); - } - + showStatus("info", results.join("\n")); submit.disabled = false; + form.reset(); + fileName.textContent = ""; + ensurePolling(); }); + + if (loadJobs().length) ensurePolling(); diff --git a/services/postgis/tagbase_schema.sql b/services/postgis/tagbase_schema.sql index b24a19a7..e48d8cc6 100644 --- a/services/postgis/tagbase_schema.sql +++ b/services/postgis/tagbase_schema.sql @@ -759,7 +759,7 @@ COMMENT ON COLUMN submission.date_time IS 'Local datetime stamp at the time of e -- Name: COLUMN submission.filename; Type: COMMENT; Schema: public; Owner: postgres -- -COMMENT ON COLUMN submission.filename IS 'Full path, name and extension of the ingested eTUFF tag data file'; +COMMENT ON COLUMN submission.filename IS 'Name of the ingested eTUFF file. Archive members keep their path inside the archive'; -- @@ -1161,8 +1161,9 @@ $$; -- The following PROCEDURE ensures that data migration (from the proc_observation's -- table to other target tables) is executed. f CREATE PROCEDURE sp_execute_data_migration(submission_id_param integer, is_reference_track_param boolean) -LANGUAGE SQL +LANGUAGE plpgsql AS $BODY$ +BEGIN --\connect tagbase -- data_time_series WITH moved_rows AS @@ -1205,7 +1206,7 @@ AS $BODY$ variable_value, submission_id, tag_id, - is_reference_track_param + moved_rows.is_reference_track_param FROM moved_rows; WITH moved_rows AS ( DELETE @@ -1293,8 +1294,7 @@ AS $BODY$ WITH moved_rows AS ( DELETE FROM proc_observations a USING observation_types b, - submission c, - data_time_series d + submission c WHERE a.variable_id = b.variable_id AND b.variable_name LIKE 'TimeAt%' AND c.submission_id = submission_id_param @@ -1304,7 +1304,7 @@ AS $BODY$ cast(substring(variable_name, '(\d+)') AS int) AS bin_class, a.date_time, a.variable_value, - d.position_date_time, + cast(NULL AS timestamp(6) with time zone), a.variable_id) INSERT INTO data_histogram_bin_data SELECT * @@ -1368,32 +1368,70 @@ AS $BODY$ WHERE data_profile.date_time = moved_rows.date_time AND data_profile.submission_id = moved_rows.submission_id; -- SQL update statements to link measurement date time with position date time - UPDATE data_time_series - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_time_series.submission_id = data_position.submission_id - AND data_time_series.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; - UPDATE data_histogram_bin_data - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_histogram_bin_data.submission_id = data_position.submission_id - AND data_histogram_bin_data.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; - UPDATE data_profile - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_profile.submission_id = data_position.submission_id - AND data_profile.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; + DROP TABLE IF EXISTS migration_position_match; + CREATE TEMP TABLE migration_position_match ON COMMIT DROP AS + SELECT m.submission_id, + m.tag_id, + m.date_time, + p.position_date_time + FROM ( + SELECT submission_id, tag_id, date_time + FROM data_time_series + WHERE position_date_time IS NULL + UNION + SELECT submission_id, tag_id, date_time + FROM data_histogram_bin_data + WHERE position_date_time IS NULL + UNION + SELECT submission_id, tag_id, date_time + FROM data_profile + WHERE position_date_time IS NULL + ) AS m + CROSS JOIN LATERAL ( + SELECT dp.date_time AS position_date_time + FROM data_position AS dp + WHERE dp.submission_id = m.submission_id + AND dp.tag_id = m.tag_id + AND dp.date_time <= m.date_time + ORDER BY dp.date_time DESC + LIMIT 1 + ) AS p; + UPDATE data_time_series AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; + UPDATE data_histogram_bin_data AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; + UPDATE data_profile AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; +END; $BODY$; +CREATE TABLE ingest_job ( + id uuid PRIMARY KEY, + status text NOT NULL + CHECK (status IN ('queued', 'running', 'succeeded', 'failed')), + source_path text NOT NULL, + filename text, + version text, + notes text, + error text, + files_total integer, + files_done integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + diff --git a/services/postgis/tagbase_schema_tables.sql b/services/postgis/tagbase_schema_tables.sql index 72ef0795..46a556ed 100644 --- a/services/postgis/tagbase_schema_tables.sql +++ b/services/postgis/tagbase_schema_tables.sql @@ -1,6 +1,6 @@ --- Regenerated for test/CI PostGIS init (no CREATE ROLE/DATABASE; OWNER=tagbase). --- Source: tagbase_schema.sql (tail -n +9, strip \connect, OWNER TO tagbase). -CREATE EXTENSION IF NOT EXISTS postgis; +-- Regenerated for test/CI PostGIS init (no CREATE ROLE/DATABASE; OWNER=tagbase). +-- Source: tagbase_schema.sql (tail -n +9, strip \connect, OWNER TO tagbase). +CREATE EXTENSION IF NOT EXISTS postgis; SET statement_timeout = 0; SET lock_timeout = 0; SET idle_in_transaction_session_timeout = 0; @@ -754,7 +754,7 @@ COMMENT ON COLUMN submission.date_time IS 'Local datetime stamp at the time of e -- Name: COLUMN submission.filename; Type: COMMENT; Schema: public; Owner: postgres -- -COMMENT ON COLUMN submission.filename IS 'Full path, name and extension of the ingested eTUFF tag data file'; +COMMENT ON COLUMN submission.filename IS 'Name of the ingested eTUFF file. Archive members keep their path inside the archive'; -- @@ -1156,8 +1156,9 @@ $$; -- The following PROCEDURE ensures that data migration (from the proc_observation's -- table to other target tables) is executed. f CREATE PROCEDURE sp_execute_data_migration(submission_id_param integer, is_reference_track_param boolean) -LANGUAGE SQL +LANGUAGE plpgsql AS $BODY$ +BEGIN --\connect tagbase -- data_time_series WITH moved_rows AS @@ -1200,7 +1201,7 @@ AS $BODY$ variable_value, submission_id, tag_id, - is_reference_track_param + moved_rows.is_reference_track_param FROM moved_rows; WITH moved_rows AS ( DELETE @@ -1288,8 +1289,7 @@ AS $BODY$ WITH moved_rows AS ( DELETE FROM proc_observations a USING observation_types b, - submission c, - data_time_series d + submission c WHERE a.variable_id = b.variable_id AND b.variable_name LIKE 'TimeAt%' AND c.submission_id = submission_id_param @@ -1299,7 +1299,7 @@ AS $BODY$ cast(substring(variable_name, '(\d+)') AS int) AS bin_class, a.date_time, a.variable_value, - d.position_date_time, + cast(NULL AS timestamp(6) with time zone), a.variable_id) INSERT INTO data_histogram_bin_data SELECT * @@ -1363,32 +1363,70 @@ AS $BODY$ WHERE data_profile.date_time = moved_rows.date_time AND data_profile.submission_id = moved_rows.submission_id; -- SQL update statements to link measurement date time with position date time - UPDATE data_time_series - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_time_series.submission_id = data_position.submission_id - AND data_time_series.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; - UPDATE data_histogram_bin_data - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_histogram_bin_data.submission_id = data_position.submission_id - AND data_histogram_bin_data.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; - UPDATE data_profile - SET position_date_time = - (SELECT date_time - FROM data_position - WHERE data_profile.submission_id = data_position.submission_id - AND data_profile.date_time >= data_position.date_time - ORDER BY data_position.date_time DESC - LIMIT 1) - WHERE position_date_time IS NULL; + DROP TABLE IF EXISTS migration_position_match; + CREATE TEMP TABLE migration_position_match ON COMMIT DROP AS + SELECT m.submission_id, + m.tag_id, + m.date_time, + p.position_date_time + FROM ( + SELECT submission_id, tag_id, date_time + FROM data_time_series + WHERE position_date_time IS NULL + UNION + SELECT submission_id, tag_id, date_time + FROM data_histogram_bin_data + WHERE position_date_time IS NULL + UNION + SELECT submission_id, tag_id, date_time + FROM data_profile + WHERE position_date_time IS NULL + ) AS m + CROSS JOIN LATERAL ( + SELECT dp.date_time AS position_date_time + FROM data_position AS dp + WHERE dp.submission_id = m.submission_id + AND dp.tag_id = m.tag_id + AND dp.date_time <= m.date_time + ORDER BY dp.date_time DESC + LIMIT 1 + ) AS p; + UPDATE data_time_series AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; + UPDATE data_histogram_bin_data AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; + UPDATE data_profile AS t + SET position_date_time = m.position_date_time + FROM migration_position_match AS m + WHERE t.position_date_time IS NULL + AND t.submission_id = m.submission_id + AND t.tag_id = m.tag_id + AND t.date_time = m.date_time; +END; $BODY$; + +CREATE TABLE ingest_job ( + id uuid PRIMARY KEY, + status text NOT NULL + CHECK (status IN ('queued', 'running', 'succeeded', 'failed')), + source_path text NOT NULL, + filename text, + version text, + notes text, + error text, + files_total integer, + files_done integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); diff --git a/tagbase_server/Dockerfile b/tagbase_server/Dockerfile index 04889bd6..ac44ec38 100644 --- a/tagbase_server/Dockerfile +++ b/tagbase_server/Dockerfile @@ -18,6 +18,6 @@ COPY . /usr/src/app EXPOSE 5433 -ENTRYPOINT ["gunicorn"] +RUN chmod 0555 /usr/src/app/docker-entrypoint.sh -CMD ["-c", "gunicorn.conf.py", "tagbase_server.__main__:app"] +ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"] diff --git a/tagbase_server/docker-entrypoint.sh b/tagbase_server/docker-entrypoint.sh new file mode 100755 index 00000000..ad5cf985 --- /dev/null +++ b/tagbase_server/docker-entrypoint.sh @@ -0,0 +1,31 @@ +#!/bin/sh +# Gunicorn accepts uploads. One ingest process beside it runs the queue. +set -eu + +python -m tagbase_server.ingest_worker & +worker_pid=$! + +gunicorn -c gunicorn.conf.py tagbase_server.__main__:app & +gunicorn_pid=$! + +stop() { + kill "$worker_pid" "$gunicorn_pid" 2>/dev/null || true +} +trap stop INT TERM + +status=0 +while true; do + if ! kill -0 "$gunicorn_pid" 2>/dev/null; then + wait "$gunicorn_pid" || status=$? + break + fi + if ! kill -0 "$worker_pid" 2>/dev/null; then + wait "$worker_pid" || status=$? + break + fi + sleep 1 +done +stop +wait "$worker_pid" 2>/dev/null || true +wait "$gunicorn_pid" 2>/dev/null || true +exit "$status" diff --git a/tagbase_server/gunicorn.conf.py b/tagbase_server/gunicorn.conf.py index 3d96ef6f..d72abbe9 100644 --- a/tagbase_server/gunicorn.conf.py +++ b/tagbase_server/gunicorn.conf.py @@ -20,8 +20,10 @@ def post_fork(server, worker): from tagbase_server.__main__ import app from tagbase_server.telemetry import setup_telemetry + from tagbase_server.upload_stream import register_upload_heartbeat from tagbase_server.utils.db_utils import assert_schema_ready, connect + register_upload_heartbeat(worker.notify) setup_telemetry(flask_app=app.app) server.log.info("OpenTelemetry setup complete in worker pid=%s", worker.pid) diff --git a/tagbase_server/pylock.toml b/tagbase_server/pylock.toml index 366f11d3..6a03fd33 100644 --- a/tagbase_server/pylock.toml +++ b/tagbase_server/pylock.toml @@ -100,28 +100,6 @@ url = "https://files.pythonhosted.org/packages/9d/3b/d9cbe13627ba18bc8c9fb3575aa [packages.wheels.hashes] sha256 = "4ed0b1a90f5a39f58e984b270b3e945c5038a1d064748a23478833027428102c" -[[packages]] -name = "easyprocess" -version = "1.1" - -[[packages.wheels]] -name = "EasyProcess-1.1-py3-none-any.whl" -url = "https://files.pythonhosted.org/packages/9c/cf/27d1f4b3bae5e566f94fc716e048120128cf603d5163638d22bcd0fc92d8/EasyProcess-1.1-py3-none-any.whl" - -[packages.wheels.hashes] -sha256 = "82eed523a0a5eb12a81fa4eacd9f342caeb3f900eb4b798740e6696ad07e63f9" - -[[packages]] -name = "entrypoint2" -version = "1.1" - -[[packages.wheels]] -name = "entrypoint2-1.1-py2.py3-none-any.whl" -url = "https://files.pythonhosted.org/packages/42/ee/84c8990b08efa0265bd10fc8781ef26e3157715bf0dfa47ee3c056b513d4/entrypoint2-1.1-py2.py3-none-any.whl" - -[packages.wheels.hashes] -sha256 = "eeb8c327bdb65cdd1668c023a6b110b7e3d1a046fb05e043861ebd9264b3a257" - [[packages]] name = "flask" version = "3.1.3" @@ -320,17 +298,6 @@ url = "https://files.pythonhosted.org/packages/2d/39/f820a78508ec19de9209eaaf5d1 [packages.wheels.hashes] sha256 = "4953c7092442dec9560f9b25f9ff184006acd467980c00ab798f1644d432a595" -[[packages]] -name = "patool" -version = "4.0.5" - -[[packages.wheels]] -name = "patool-4.0.5-py2.py3-none-any.whl" -url = "https://files.pythonhosted.org/packages/61/50/cb792aaf25a8f42f5738dafb3030a1243d484ab366ff321ebcd45dfcfc12/patool-4.0.5-py2.py3-none-any.whl" - -[packages.wheels.hashes] -sha256 = "9413a114c7aedf2263f151a56798055a593de671ef0a9ef94870284ad21818fb" - [[packages]] name = "psycopg2-binary" version = "2.9.12" @@ -386,17 +353,6 @@ url = "https://files.pythonhosted.org/packages/ec/dd/96da98f892250475bdf2328112d [packages.wheels.hashes] sha256 = "04156e608bee23d3792fd45c94ae47fae1036688e75032eea2e3bf0323d1f126" -[[packages]] -name = "pyunpack" -version = "0.3" - -[[packages.wheels]] -name = "pyunpack-0.3-py2.py3-none-any.whl" -url = "https://files.pythonhosted.org/packages/72/20/5a6dcb0d28529ce6efe850755994c80817279eecf08620003775fda3b914/pyunpack-0.3-py2.py3-none-any.whl" - -[packages.wheels.hashes] -sha256 = "8f517cfc71215f37f74cf3a7668028828c68dc76f4d02e7a69f227ce978d51a3" - [[packages]] name = "pyyaml" version = "6.0.3" diff --git a/tagbase_server/requirements.txt b/tagbase_server/requirements.txt index cc73487e..955126f5 100644 --- a/tagbase_server/requirements.txt +++ b/tagbase_server/requirements.txt @@ -10,9 +10,7 @@ opentelemetry-instrumentation-logging==0.65b0 opentelemetry-instrumentation-psycopg2==0.65b0 pandas==3.0.6 parmap==1.7.0 -patool==4.1.0 psycopg2-binary==2.9.13 python_dateutil==2.9.0.post0 -pyunpack==0.3 tqdm==4.70.1 tzlocal==5.4.4 diff --git a/tagbase_server/tagbase_server/__main__.py b/tagbase_server/tagbase_server/__main__.py index 5b881131..53620139 100644 --- a/tagbase_server/tagbase_server/__main__.py +++ b/tagbase_server/tagbase_server/__main__.py @@ -11,6 +11,7 @@ from tagbase_server import encoder from tagbase_server.problem import register_problem_handlers from tagbase_server.telemetry import setup_telemetry +from tagbase_server.upload_stream import UploadGuardMiddleware LOGGER_NAME = "tagbase_server" @@ -62,6 +63,7 @@ def configure_cors(flask_app, cors_factory=CORS): swagger_ui_options=swagger_ui_options, jsonifier=encoder.create_jsonifier(), ) +app.add_middleware(UploadGuardMiddleware) app.add_api( "openapi.yaml", arguments={"title": "tagbase-server API"}, diff --git a/tagbase_server/tagbase_server/archives.py b/tagbase_server/tagbase_server/archives.py new file mode 100644 index 00000000..4cd46ab0 --- /dev/null +++ b/tagbase_server/tagbase_server/archives.py @@ -0,0 +1,285 @@ +"""Shared eTUFF archive handling for synchronous ingest and the job worker. + +Allowed inputs are .txt, .zip, .tar.gz, and .tgz. The logical filename and +the file's magic bytes decide the format. A temporary path suffix does not. +""" + +from __future__ import annotations + +import os +import shutil +import stat +import tarfile +import tempfile +import zipfile +from dataclasses import dataclass +from pathlib import PurePosixPath + +from tagbase_server.utils.io_utils import TEMP_DIR + +_ZIP_MAGIC = b"PK" +_GZIP_MAGIC = b"\x1f\x8b" +_DEFAULT_MAX_MEMBERS = 10_000 +_DEFAULT_MAX_UNCOMPRESSED = 10 * 1024 * 1024 * 1024 +_READ_CHUNK = 1024 * 1024 +_INVALID_FILENAME = "invalid filename" +_NO_ETUFF_MEMBERS = "archive contained no eTUFF files" +_CORRUPT_ARCHIVE = "archive is corrupt or unreadable" + + +class ArchiveError(ValueError): + """The upload is empty, the wrong type, or an unsafe archive.""" + + +@dataclass +class PreparedIngest: + """eTUFF members ready to parse, plus an extraction directory to delete.""" + + files: list[str] + workspace: str | None + names: list[str] | None = None + + def submission_names(self) -> list[str]: + """Names stored on submission rows. + + Archive members keep their path inside the archive. + """ + if self.names is None: + return [os.path.basename(path) for path in self.files] + if len(self.names) != len(self.files): + raise RuntimeError("ingest names do not match extracted files") + return list(self.names) + + def cleanup(self) -> None: + if self.workspace: + shutil.rmtree(self.workspace) + + +def archive_limits() -> tuple[int, int]: + members = int(os.environ.get("TAGBASE_ARCHIVE_MAX_MEMBERS", _DEFAULT_MAX_MEMBERS)) + expanded = int( + os.environ.get( + "TAGBASE_ARCHIVE_MAX_UNCOMPRESSED", + str(_DEFAULT_MAX_UNCOMPRESSED), + ) + ) + return members, expanded + + +def logical_filename(name: str) -> str: + """Reject blank names and any path, keeping a single basename.""" + if name is None: + raise ArchiveError(_INVALID_FILENAME) + text = str(name).strip() + if not text or text in {".", ".."} or "/" in text or "\\" in text: + raise ArchiveError(_INVALID_FILENAME) + base = os.path.basename(text) + if base != text: + raise ArchiveError(_INVALID_FILENAME) + return base + + +def classify(path: str, filename: str) -> str: + """Return etuff, zip, or tar after the name and magic bytes agree.""" + logical = logical_filename(filename) + lowered = logical.lower() + if lowered.endswith((".tar.gz", ".tgz")): + expected = "tar" + elif lowered.endswith(".zip"): + expected = "zip" + elif lowered.endswith(".txt"): + expected = "etuff" + else: + raise ArchiveError( + "unsupported file type; allowed types are .txt, .zip, .tar.gz, and .tgz" + ) + sniffed = _sniff(path) + if expected == "etuff" and sniffed != "text": + raise ArchiveError("file contents do not match the .txt filename") + if expected == "zip" and sniffed != "zip": + raise ArchiveError("file contents do not match the .zip filename") + if expected == "tar" and sniffed != "gzip": + raise ArchiveError("file contents do not match the archive filename") + return expected + + +def preflight(path: str, filename: str) -> str: + """Reject an empty, unsupported, or unsafe upload without extracting it.""" + kind = classify(path, filename) + if os.path.getsize(path) == 0: + raise ArchiveError("request body is empty") + if kind == "zip": + _inspect_zip(path) + elif kind == "tar": + _inspect_tar(path) + return kind + + +def materialize(path: str, filename: str) -> PreparedIngest: + """Validate the upload and return the eTUFF files to parse.""" + kind = preflight(path, filename) + if kind == "etuff": + return PreparedIngest( + files=[path], + workspace=None, + names=[logical_filename(filename)], + ) + workspace = tempfile.mkdtemp(dir=TEMP_DIR, prefix="tagbase-archive-") + try: + if kind == "zip": + members = _extract_zip(path, workspace) + else: + members = _extract_tar(path, workspace) + except Exception: + shutil.rmtree(workspace, ignore_errors=True) + raise + if not members: + shutil.rmtree(workspace, ignore_errors=True) + raise ArchiveError(_NO_ETUFF_MEMBERS) + return PreparedIngest( + files=[member_path for member_path, _name in members], + workspace=workspace, + names=[name for _member_path, name in members], + ) + + +def _sniff(path: str) -> str: + with open(path, "rb") as handle: + header = handle.read(8) + if not header: + raise ArchiveError("request body is empty") + if header.startswith(_ZIP_MAGIC): + return "zip" + if header.startswith(_GZIP_MAGIC): + return "gzip" + return "text" + + +def _member_name(name: str) -> str: + parts = PurePosixPath(name).parts + if ( + not name + or name.startswith("/") + or name.startswith("\\") + or ".." in parts + or "\\" in name + ): + raise ArchiveError(f"archive member path is not allowed: {name}") + return str(PurePosixPath(*parts)) + + +def _limits_exceeded(count: int, declared: int) -> None: + max_members, max_uncompressed = archive_limits() + if count > max_members: + raise ArchiveError("archive has too many members") + if declared > max_uncompressed: + raise ArchiveError("archive uncompressed size exceeds the configured cap") + + +def _inspect_zip(path: str) -> None: + try: + with zipfile.ZipFile(path) as archive: + infos = archive.infolist() + declared = 0 + txts = 0 + for index, info in enumerate(infos, start=1): + _reject_zip_member(info) + if not info.is_dir(): + declared += info.file_size + if info.filename.endswith(".txt"): + txts += 1 + _limits_exceeded(index, declared) + except zipfile.BadZipFile as exc: + raise ArchiveError(_CORRUPT_ARCHIVE) from exc + if txts == 0: + raise ArchiveError(_NO_ETUFF_MEMBERS) + + +def _reject_zip_member(info: zipfile.ZipInfo) -> None: + if info.flag_bits & 0x1: + raise ArchiveError("encrypted ZIP members are not allowed") + _member_name(info.filename) + mode = info.external_attr >> 16 + if stat.S_ISLNK(mode): + raise ArchiveError("symlink archive members are not allowed") + + +def _extract_zip(path: str, workspace: str) -> list[tuple[str, str]]: + files: list[tuple[str, str]] = [] + written = 0 + _, max_uncompressed = archive_limits() + with zipfile.ZipFile(path) as archive: + for info in archive.infolist(): + _reject_zip_member(info) + if info.is_dir(): + continue + relative = _member_name(info.filename) + target = os.path.join(workspace, relative) + os.makedirs(os.path.dirname(target), exist_ok=True) + with archive.open(info) as source, open(target, "wb") as dest: + while True: + chunk = source.read(_READ_CHUNK) + if not chunk: + break + written += len(chunk) + if written > max_uncompressed: + raise ArchiveError("extracted bytes exceed the configured cap") + dest.write(chunk) + if relative.endswith(".txt"): + files.append((target, relative)) + return files + + +def _inspect_tar(path: str) -> None: + try: + with tarfile.open(path, "r:gz") as archive: + declared = 0 + txts = 0 + count = 0 + for member in archive: + count += 1 + _reject_tar_member(member) + if member.isfile(): + declared += member.size + if member.name.endswith(".txt"): + txts += 1 + _limits_exceeded(count, declared) + except (tarfile.TarError, OSError) as exc: + raise ArchiveError(_CORRUPT_ARCHIVE) from exc + if txts == 0: + raise ArchiveError(_NO_ETUFF_MEMBERS) + + +def _reject_tar_member(member: tarfile.TarInfo) -> None: + if member.issym() or member.islnk(): + raise ArchiveError("symlink archive members are not allowed") + _member_name(member.name) + + +def _extract_tar(path: str, workspace: str) -> list[tuple[str, str]]: + files: list[tuple[str, str]] = [] + written = 0 + _, max_uncompressed = archive_limits() + with tarfile.open(path, "r:gz") as archive: + for member in archive: + _reject_tar_member(member) + if not member.isfile(): + continue + relative = _member_name(member.name) + target = os.path.join(workspace, relative) + os.makedirs(os.path.dirname(target), exist_ok=True) + source = archive.extractfile(member) + if source is None: + raise ArchiveError(_CORRUPT_ARCHIVE) + with source, open(target, "wb") as dest: + while True: + chunk = source.read(_READ_CHUNK) + if not chunk: + break + written += len(chunk) + if written > max_uncompressed: + raise ArchiveError("extracted bytes exceed the configured cap") + dest.write(chunk) + if relative.endswith(".txt"): + files.append((target, relative)) + return files diff --git a/tagbase_server/tagbase_server/controllers/ingest_controller.py b/tagbase_server/tagbase_server/controllers/ingest_controller.py index 0efb540a..1f3f3122 100644 --- a/tagbase_server/tagbase_server/controllers/ingest_controller.py +++ b/tagbase_server/tagbase_server/controllers/ingest_controller.py @@ -1,18 +1,35 @@ import logging +import os +import tempfile import time +import uuid from functools import partial from multiprocessing import cpu_count +from urllib.parse import urlparse import parmap from tqdm import tqdm as std_tqdm +from tagbase_server.archives import ( + ArchiveError, + logical_filename, + materialize, + preflight, +) +from tagbase_server.ingest_jobs import IngestJobError, enqueue_job, fetch_job from tagbase_server.models.ingest200 import Ingest200 # noqa: E501 -from tagbase_server.problem import TagbaseClientError, as_json, current_trace_id +from tagbase_server.problem import ( + TYPE_HTTP, + TagbaseClientError, + as_json, + current_trace_id, +) from tagbase_server.telemetry import get_tracer, record_ingest_request +from tagbase_server.upload_stream import job_max_bytes, staged_upload, sync_max_bytes from tagbase_server.utils.io_utils import ( + TEMP_DIR, process_get_input_data, process_post_input_data, - unpack_compressed_binary, ) from tagbase_server.utils.processing_utils import process_etuff_file @@ -31,14 +48,20 @@ def _resolve_ingest_file_type(type): def _ingest_progress_bar(data_file): - """Create a standard stdout progress bar for ingestio]n.""" + """Create a standard stdout progress bar for ingestion.""" return partial(std_tqdm, desc=f"Ingesting: {data_file}") -def _run_ingest_queue(etuff_files, version, notes, data_file): +def _ingest_member(item, version=None, notes=None): + path, name = item + return process_etuff_file(path, version=version, notes=notes, submission_name=name) + + +def _run_ingest_queue(prepared, version, notes, data_file): + members = list(zip(prepared.files, prepared.submission_names(), strict=True)) parmap.map( - process_etuff_file, - etuff_files, + _ingest_member, + members, version=version, notes=notes, pm_parallel=False, @@ -68,52 +91,81 @@ def _ingest_with_telemetry(operation_name, work): record_ingest_request(outcome, time.perf_counter() - start) -def ingest_get(file, notes=None, type=None, type_=None, version=None): # noqa: E501 - """Get network accessible file and execute ingestion +def _logical_from_url(file_url: str) -> str: + path = urlparse(file_url).path or file_url + return os.path.basename(path.rstrip("/")) - Get network accessible file and execute ingestion # noqa: E501 - :param file: Location of a network accessible (file, ftp, http, https) file e.g. 'file:///usr/src/app/data/eTUFF-sailfish-117259.txt'. - :type file: str - :param notes: Free-form text field where details of submitted eTUFF file for ingest can be provided e.g. submitter name, etuff data contents (tag metadata and measurements + primary position data, or just secondary solution-positional meta/data) - :type notes: str - :param type: Type of file to be ingested, defaults to 'etuff' - :type type: str - :param type_: Connexion pythonic alias for OpenAPI parameter ``type`` - :type type_: str - :param version: Version identifier for the eTUFF tag data file ingested - :type version: str +def _safe_remove(path: str | None) -> None: + if not path: + return + try: + if os.path.isfile(path): + os.remove(path) + except OSError: + logger.exception("Failed to delete ingest file %s", path) - :rtype: Union[Ingest200, Tuple[Ingest200, int], Tuple[Ingest200, int, Dict[str, str]] - """ - def _work(): - ingest_file_type = _resolve_ingest_file_type( - type_ if type_ is not None else type - ) - logger.info("Ingest file type: %s", ingest_file_type) - start = time.perf_counter() - data_file = process_get_input_data(file) - etuff_files = [] - if not data_file.endswith(".txt"): - etuff_files = unpack_compressed_binary(data_file) - else: - etuff_files.append(data_file) - logger.info("eTUFF ingestion queue: %s", etuff_files) - _run_ingest_queue(etuff_files, version, notes, data_file) - finish = time.perf_counter() - elapsed = round(finish - start, 2) +def _process_members(data_file, logical_name, version, notes, started): + prepared = materialize(data_file, logical_name) + try: + logger.info("eTUFF ingestion queue: %s", prepared.submission_names()) + _run_ingest_queue(prepared, version, notes, data_file) + elapsed = round(time.perf_counter() - started, 2) return as_json( Ingest200.from_dict( { "code": "200", "elapsed": elapsed, "trace_id": current_trace_id(), - "message": f"Processing %s file(s) - {etuff_files}" - % len(etuff_files), + "message": f"Processing {len(prepared.files)} file(s) - {prepared.files}", } ) ) + finally: + try: + prepared.cleanup() + except OSError: + logger.exception("Failed to delete extracted workspace for %s", data_file) + + +def _reject_sync_body(body) -> bytes: + payload = body if isinstance(body, (bytes, bytearray)) else bytes(body) + if len(payload) == 0: + raise TagbaseClientError("Request body is empty.") + if len(payload) > sync_max_bytes(): + raise TagbaseClientError( + "Synchronous ingest accepts at most 100 MiB. Use POST /ingest/jobs.", + title="Payload Too Large", + type_=TYPE_HTTP, + status=413, + ) + return payload + + +def ingest_get(file, notes=None, type=None, type_=None, version=None): # noqa: E501 + """Get network accessible file and execute ingestion.""" + + def _work(): + ingest_file_type = _resolve_ingest_file_type( + type_ if type_ is not None else type + ) + logger.info("Ingest file type: %s", ingest_file_type) + started = time.perf_counter() + data_file = process_get_input_data(file) + local = urlparse(file).path + owned = data_file != local + try: + return _process_members( + data_file, + _logical_from_url(file), + version, + notes, + started, + ) + finally: + if owned: + _safe_remove(data_file) return _ingest_with_telemetry("get", _work) @@ -121,52 +173,114 @@ def _work(): def ingest_post( filename, body, notes=None, type=None, type_=None, version=None ): # noqa: E501 - """Post a local file and perform a ingest operation - - Post a local file and perform a ingest operation # noqa: E501 - - :param filename: Free-form text field to explicitly define the name of the file to be persisted - :type filename: str - :param body: - :type body: str - :param notes: Free-form text field where details of submitted eTUFF file for ingest can be provided e.g. submitter name, etuff data contents (tag metadata and measurements + primary position data, or just secondary solution-positional meta/data) - :type notes: str - :param type: Type of file to be ingested, defaults to 'etuff' - :type type: str - :param type_: Connexion pythonic alias for OpenAPI parameter ``type`` - :type type_: str - :param version: Version identifier for the eTUFF tag data file ingested - :type version: str - - :rtype: Union[Ingest200, Tuple[Ingest200, int], Tuple[Ingest200, int, Dict[str, str]] - """ + """Post a local file and perform a ingest operation.""" def _work(): ingest_file_type = _resolve_ingest_file_type( type_ if type_ is not None else type ) logger.info("Ingest file type: %s", ingest_file_type) - start = time.perf_counter() - data_file = process_post_input_data(filename, body) - etuff_files = [] - if not data_file.endswith(".txt"): - etuff_files = unpack_compressed_binary(data_file) - else: - etuff_files.append(data_file) - logger.info("eTUFF ingestion queue: %s", etuff_files) - _run_ingest_queue(etuff_files, version, notes, data_file) - finish = time.perf_counter() - elapsed = round(finish - start, 2) - return as_json( - Ingest200.from_dict( - { - "code": "200", - "elapsed": elapsed, - "trace_id": current_trace_id(), - "message": f"Processing %s file(s) - {etuff_files}." - % len(etuff_files), - } - ) - ) + started = time.perf_counter() + payload = _reject_sync_body(body) + safe_name = logical_filename(filename) + data_file = process_post_input_data(safe_name, payload) + try: + return _process_members(data_file, safe_name, version, notes, started) + finally: + _safe_remove(data_file) return _ingest_with_telemetry("post", _work) + + +def _stage_job_body(body) -> str: + staged = staged_upload.get() + if staged: + return staged + payload = body if isinstance(body, (bytes, bytearray)) else bytes(body) + if len(payload) == 0: + raise TagbaseClientError("Request body is empty.") + if len(payload) > job_max_bytes(): + raise TagbaseClientError( + "Ingest jobs accept at most 2 GiB.", + title="Payload Too Large", + type_=TYPE_HTTP, + status=413, + ) + handle = tempfile.NamedTemporaryFile( + dir=TEMP_DIR, + prefix="tagbase-upload-", + delete=False, + ) + try: + view = memoryview(payload) + for offset in range(0, len(view), 1024 * 1024): + handle.write(view[offset : offset + 1024 * 1024]) + finally: + handle.close() + return handle.name + + +def _isolate_upload(source_path): + """Keep a POST body off the caller-supplied name so a later post cannot overwrite it.""" + unique = os.path.join( + TEMP_DIR, f"{uuid.uuid4().hex}-{os.path.basename(source_path)}" + ) + os.replace(source_path, unique) + return unique + + +def create_ingest_job( + filename, body, notes=None, type=None, type_=None, version=None +): # noqa: E501 + """Stream a large upload into the ingest queue and return 202.""" + started = time.perf_counter() + outcome = "accepted" + stored = None + try: + _resolve_ingest_file_type(type_ if type_ is not None else type) + staged = _stage_job_body(body) + try: + preflight(staged, filename) + except ArchiveError as exc: + _safe_remove(staged) + raise TagbaseClientError(str(exc)) from exc + stored = _isolate_upload(staged) + job_id = enqueue_job( + source_path=stored, + filename=filename, + version=version, + notes=notes, + ) + return as_json({"id": job_id, "status": "queued"}, 202) + except TagbaseClientError: + outcome = "client_error" + _safe_remove(stored) + raise + except IngestJobError: + outcome = "server_error" + _safe_remove(stored) + logger.exception("Failed to queue ingest for %s", filename) + raise + except Exception: + outcome = "server_error" + _safe_remove(stored) + raise + finally: + record_ingest_request(outcome, time.perf_counter() - started) + + +def get_ingest_job(job_id): # noqa: E501 + """Return one ingest job.""" + try: + job = fetch_job(job_id) + except IngestJobError: + logger.exception("Failed to read ingest job %s", job_id) + raise + if job is None: + raise TagbaseClientError( + f"Ingest job '{job_id}' was not found.", + title="Not Found", + type_=TYPE_HTTP, + status=404, + ) + return as_json(job) diff --git a/tagbase_server/tagbase_server/ingest_jobs.py b/tagbase_server/tagbase_server/ingest_jobs.py new file mode 100644 index 00000000..e7cb7ec9 --- /dev/null +++ b/tagbase_server/tagbase_server/ingest_jobs.py @@ -0,0 +1,189 @@ +"""Queued ingest jobs. + +A request stores the upload and inserts one row. The ingest process claims +the oldest queued row and updates it until the job is terminal. +""" + +from __future__ import annotations + +import os +import uuid +from typing import Any + +from psycopg2.extras import RealDictCursor + +from tagbase_server.utils.db_utils import connect + +RESTART_ERROR = "restarted before ingest finished" +MISSING_FILE_ERROR = "upload file missing after restart" +DEFAULT_RETENTION_DAYS = 30 + +_CLAIM_SQL = """ +UPDATE ingest_job +SET status = 'running', updated_at = now() +WHERE id = ( + SELECT id + FROM ingest_job + WHERE status = 'queued' + ORDER BY created_at, updated_at + FOR UPDATE SKIP LOCKED + LIMIT 1 +) +RETURNING * +""" + + +class IngestJobError(Exception): + """The job table could not be read or written.""" + + +def stale_failures(jobs: list[dict[str, Any]], path_exists) -> list[tuple[str, str]]: + """Jobs a restart must fail. + + A running job was interrupted. A queued job whose upload is gone cannot + run. A queued job whose file is still present stays queued. + """ + failed: list[tuple[str, str]] = [] + for job in jobs: + status = job["status"] + job_id = str(job["id"]) + if status == "running": + failed.append((job_id, RESTART_ERROR)) + elif status == "queued" and not path_exists(job["source_path"]): + failed.append((job_id, MISSING_FILE_ERROR)) + return failed + + +def public_job(row: dict[str, Any]) -> dict[str, Any]: + job: dict[str, Any] = { + "id": str(row["id"]), + "status": row["status"], + "files_done": row["files_done"], + } + if row["error"] is not None: + job["error"] = row["error"] + if row["files_total"] is not None: + job["files_total"] = row["files_total"] + return job + + +def enqueue_job( + *, + source_path: str, + filename: str | None, + version: str | None, + notes: str | None, +) -> str: + job_id = str(uuid.uuid4()) + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO ingest_job (id, status, source_path, filename, version, notes) + VALUES (%s, 'queued', %s, %s, %s, %s) + """, + (job_id, source_path, filename, version, notes), + ) + conn.commit() + return job_id + + +def fetch_job(job_id: str) -> dict[str, Any] | None: + with _connection() as conn: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute("SELECT * FROM ingest_job WHERE id = %s", (job_id,)) + row = cur.fetchone() + conn.commit() + if row is None: + return None + return public_job(dict(row)) + + +def recover_stale_jobs() -> None: + with _connection() as conn: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(""" + SELECT id, status, source_path + FROM ingest_job + WHERE status IN ('queued', 'running') + """) + rows = [dict(row) for row in cur.fetchall()] + for job_id, error in stale_failures(rows, os.path.isfile): + cur.execute( + """ + UPDATE ingest_job + SET status = 'failed', error = %s, updated_at = now() + WHERE id = %s AND status IN ('queued', 'running') + """, + (error, job_id), + ) + conn.commit() + + +def claim_next_job() -> dict[str, Any] | None: + with _connection() as conn: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute(_CLAIM_SQL) + row = cur.fetchone() + conn.commit() + if row is None: + return None + return dict(row) + + +def retention_days() -> int: + return int(os.environ.get("INGEST_JOB_RETENTION_DAYS", DEFAULT_RETENTION_DAYS)) + + +def purge_expired_jobs(days: int | None = None) -> int: + """Delete terminal jobs older than the retention window.""" + window = DEFAULT_RETENTION_DAYS if days is None else days + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM ingest_job + WHERE status IN ('succeeded', 'failed') + AND updated_at < now() - (%s * interval '1 day') + """, + (window,), + ) + deleted = cur.rowcount + conn.commit() + return deleted + + +def set_progress(job_id: str, *, files_total: int, files_done: int) -> None: + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE ingest_job + SET files_total = %s, files_done = %s, updated_at = now() + WHERE id = %s + """, + (files_total, files_done, job_id), + ) + conn.commit() + + +def finish_job(job_id: str, *, status: str, error: str | None) -> None: + with _connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + UPDATE ingest_job + SET status = %s, error = %s, updated_at = now() + WHERE id = %s + """, + (status, error, job_id), + ) + conn.commit() + + +def _connection(): + conn = connect() + if not hasattr(conn, "cursor"): + raise IngestJobError("database unavailable") + conn.autocommit = False + return conn diff --git a/tagbase_server/tagbase_server/ingest_worker.py b/tagbase_server/tagbase_server/ingest_worker.py new file mode 100644 index 00000000..31a96475 --- /dev/null +++ b/tagbase_server/tagbase_server/ingest_worker.py @@ -0,0 +1,107 @@ +"""One ingest process. It runs a single upload at a time.""" + +from __future__ import annotations + +import logging +import os +import time + +from tagbase_server.archives import PreparedIngest, materialize +from tagbase_server.ingest_jobs import ( + claim_next_job, + finish_job, + recover_stale_jobs, + set_progress, +) +from tagbase_server.utils.processing_utils import process_etuff_file + +logger = logging.getLogger(__name__) + +_POLL_SECONDS = 3 + + +def run_job(job: dict) -> None: + """Ingest every member. The first error fails the job and stops the queue.""" + job_id = str(job["id"]) + logical = job.get("filename") or os.path.basename(job["source_path"]) + prepared = materialize(job["source_path"], logical) + job["_workspace"] = prepared.workspace + members = prepared.files + if not members: + raise RuntimeError("archive contained no eTUFF files") + set_progress(job_id, files_total=len(members), files_done=0) + for done, (member, name) in enumerate( + zip(members, prepared.submission_names(), strict=True), start=1 + ): + process_etuff_file( + member, + version=job.get("version"), + notes=job.get("notes"), + count_request=False, + submission_name=name, + ) + set_progress(job_id, files_total=len(members), files_done=done) + + +def _discard_job_files(job: dict) -> None: + workspace = job.get("_workspace") + if workspace: + try: + PreparedIngest(files=[], workspace=workspace).cleanup() + except OSError: + logger.exception( + "Failed to delete extracted workspace for job %s", job.get("id") + ) + source = job.get("source_path") + if not source: + return + try: + if os.path.isfile(source): + os.remove(source) + except OSError: + logger.exception("Failed to delete staged upload %s", source) + + +def process_next() -> bool: + """Claim and run the oldest queued job. Return False when the queue is empty.""" + job = claim_next_job() + if job is None: + return False + job_id = str(job["id"]) + try: + run_job(job) + except Exception as exc: + logger.exception("Ingest job %s failed", job_id) + finish_job(job_id, status="failed", error=str(exc)) + else: + finish_job(job_id, status="succeeded", error=None) + finally: + _discard_job_files(job) + return True + + +def process_available() -> int: + """Run every job that is already queued. Used by tests in this process.""" + recover_stale_jobs() + ran = 0 + while process_next(): + ran += 1 + return ran + + +def main() -> None: + from tagbase_server.telemetry import setup_telemetry + + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + setup_telemetry(flask_app=None) + logger.info("Ingest worker starting") + recover_stale_jobs() + while True: + if not process_next(): + time.sleep(_POLL_SECONDS) + + +if __name__ == "__main__": + main() diff --git a/tagbase_server/tagbase_server/models/__init__.py b/tagbase_server/tagbase_server/models/__init__.py index 65a21d02..090fc3b0 100644 --- a/tagbase_server/tagbase_server/models/__init__.py +++ b/tagbase_server/tagbase_server/models/__init__.py @@ -3,9 +3,10 @@ # flake8: noqa from __future__ import absolute_import -from tagbase_server.models.ingest200 import Ingest200 - # import models into model package +from tagbase_server.models.ingest200 import Ingest200 +from tagbase_server.models.ingest_accepted import IngestAccepted +from tagbase_server.models.ingest_job import IngestJob from tagbase_server.models.problem import Problem from tagbase_server.models.tag200 import Tag200 from tagbase_server.models.tag200_collection import Tag200Collection diff --git a/tagbase_server/tagbase_server/models/ingest_accepted.py b/tagbase_server/tagbase_server/models/ingest_accepted.py new file mode 100644 index 00000000..f08c82b1 --- /dev/null +++ b/tagbase_server/tagbase_server/models/ingest_accepted.py @@ -0,0 +1,113 @@ +# coding: utf-8 + +from __future__ import absolute_import + +import re # noqa: E501 +from datetime import date, datetime # noqa: F401 +from typing import Dict, List # noqa: F401 + +from tagbase_server import util +from tagbase_server.models.base_model_ import Model + + +class IngestAccepted(Model): + """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + + Do not edit the class manually. + """ + + def __init__(self, id=None, status=None): # noqa: E501 + """IngestAccepted - a model defined in OpenAPI + + :param id: The id of this IngestAccepted. # noqa: E501 + :type id: str + :param status: The status of this IngestAccepted. # noqa: E501 + :type status: str + """ + self.openapi_types = {"id": str, "status": str} + + self.attribute_map = {"id": "id", "status": "status"} + + self._id = id + self._status = status + + @classmethod + def from_dict(cls, dikt) -> "IngestAccepted": + """Returns the dict as a model + + :param dikt: A dict. + :type: dict + :return: The IngestAccepted of this IngestAccepted. # noqa: E501 + :rtype: IngestAccepted + """ + return util.deserialize_model(dikt, cls) + + @property + def id(self) -> str: + """Gets the id of this IngestAccepted. + + Job id # noqa: E501 + + :return: The id of this IngestAccepted. + :rtype: str + """ + return self._id + + @id.setter + def id(self, id: str): + """Sets the id of this IngestAccepted. + + Job id # noqa: E501 + + :param id: The id of this IngestAccepted. + :type id: str + """ + if id is None: + raise ValueError("Invalid value for `id`, must not be `None`") # noqa: E501 + if id is not None and len(id) > 36: + raise ValueError( + "Invalid value for `id`, length must be less than or equal to `36`" + ) # noqa: E501 + if id is not None and len(id) < 36: + raise ValueError( + "Invalid value for `id`, length must be greater than or equal to `36`" + ) # noqa: E501 + if id is not None and not re.search( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$", + id, + ): # noqa: E501 + raise ValueError( + r"Invalid value for `id`, must be a follow pattern or equal to `/^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/`" + ) # noqa: E501 + + self._id = id + + @property + def status(self) -> str: + """Gets the status of this IngestAccepted. + + Job status at accept time # noqa: E501 + + :return: The status of this IngestAccepted. + :rtype: str + """ + return self._status + + @status.setter + def status(self, status: str): + """Sets the status of this IngestAccepted. + + Job status at accept time # noqa: E501 + + :param status: The status of this IngestAccepted. + :type status: str + """ + allowed_values = ["queued"] # noqa: E501 + if status not in allowed_values: + raise ValueError( + "Invalid value for `status` ({0}), must be one of {1}".format( + status, allowed_values + ) + ) + + self._status = status diff --git a/tagbase_server/tagbase_server/models/ingest_job.py b/tagbase_server/tagbase_server/models/ingest_job.py new file mode 100644 index 00000000..64fd5563 --- /dev/null +++ b/tagbase_server/tagbase_server/models/ingest_job.py @@ -0,0 +1,237 @@ +# coding: utf-8 + +from __future__ import absolute_import + +import re # noqa: E501 +from datetime import date, datetime # noqa: F401 +from typing import Dict, List # noqa: F401 + +from tagbase_server import util +from tagbase_server.models.base_model_ import Model + + +class IngestJob(Model): + """NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + + Do not edit the class manually. + """ + + def __init__( + self, id=None, status=None, error=None, files_total=None, files_done=None + ): # noqa: E501 + """IngestJob - a model defined in OpenAPI + + :param id: The id of this IngestJob. # noqa: E501 + :type id: str + :param status: The status of this IngestJob. # noqa: E501 + :type status: str + :param error: The error of this IngestJob. # noqa: E501 + :type error: str + :param files_total: The files_total of this IngestJob. # noqa: E501 + :type files_total: int + :param files_done: The files_done of this IngestJob. # noqa: E501 + :type files_done: int + """ + self.openapi_types = { + "id": str, + "status": str, + "error": str, + "files_total": int, + "files_done": int, + } + + self.attribute_map = { + "id": "id", + "status": "status", + "error": "error", + "files_total": "files_total", + "files_done": "files_done", + } + + self._id = id + self._status = status + self._error = error + self._files_total = files_total + self._files_done = files_done + + @classmethod + def from_dict(cls, dikt) -> "IngestJob": + """Returns the dict as a model + + :param dikt: A dict. + :type: dict + :return: The IngestJob of this IngestJob. # noqa: E501 + :rtype: IngestJob + """ + return util.deserialize_model(dikt, cls) + + @property + def id(self) -> str: + """Gets the id of this IngestJob. + + Job id # noqa: E501 + + :return: The id of this IngestJob. + :rtype: str + """ + return self._id + + @id.setter + def id(self, id: str): + """Sets the id of this IngestJob. + + Job id # noqa: E501 + + :param id: The id of this IngestJob. + :type id: str + """ + if id is None: + raise ValueError("Invalid value for `id`, must not be `None`") # noqa: E501 + if id is not None and len(id) > 36: + raise ValueError( + "Invalid value for `id`, length must be less than or equal to `36`" + ) # noqa: E501 + if id is not None and len(id) < 36: + raise ValueError( + "Invalid value for `id`, length must be greater than or equal to `36`" + ) # noqa: E501 + if id is not None and not re.search( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$", + id, + ): # noqa: E501 + raise ValueError( + r"Invalid value for `id`, must be a follow pattern or equal to `/^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/`" + ) # noqa: E501 + + self._id = id + + @property + def status(self) -> str: + """Gets the status of this IngestJob. + + queued, running, succeeded, or failed # noqa: E501 + + :return: The status of this IngestJob. + :rtype: str + """ + return self._status + + @status.setter + def status(self, status: str): + """Sets the status of this IngestJob. + + queued, running, succeeded, or failed # noqa: E501 + + :param status: The status of this IngestJob. + :type status: str + """ + allowed_values = ["queued", "running", "succeeded", "failed"] # noqa: E501 + if status not in allowed_values: + raise ValueError( + "Invalid value for `status` ({0}), must be one of {1}".format( + status, allowed_values + ) + ) + + self._status = status + + @property + def error(self) -> str: + """Gets the error of this IngestJob. + + Present when the job failed # noqa: E501 + + :return: The error of this IngestJob. + :rtype: str + """ + return self._error + + @error.setter + def error(self, error: str): + """Sets the error of this IngestJob. + + Present when the job failed # noqa: E501 + + :param error: The error of this IngestJob. + :type error: str + """ + if error is not None and len(error) > 10000: + raise ValueError( + "Invalid value for `error`, length must be less than or equal to `10000`" + ) # noqa: E501 + if error is not None and len(error) < 1: + raise ValueError( + "Invalid value for `error`, length must be greater than or equal to `1`" + ) # noqa: E501 + if error is not None and not re.search(r"^(?!\s*$).+$", error): # noqa: E501 + raise ValueError( + r"Invalid value for `error`, must be a follow pattern or equal to `/^(?!\s*$).+$/`" + ) # noqa: E501 + + self._error = error + + @property + def files_total(self) -> int: + """Gets the files_total of this IngestJob. + + Members in the archive, once unpacking has started # noqa: E501 + + :return: The files_total of this IngestJob. + :rtype: int + """ + return self._files_total + + @files_total.setter + def files_total(self, files_total: int): + """Sets the files_total of this IngestJob. + + Members in the archive, once unpacking has started # noqa: E501 + + :param files_total: The files_total of this IngestJob. + :type files_total: int + """ + if files_total is not None and files_total > 2147483647: # noqa: E501 + raise ValueError( + "Invalid value for `files_total`, must be a value less than or equal to `2147483647`" + ) # noqa: E501 + if files_total is not None and files_total < 0: # noqa: E501 + raise ValueError( + "Invalid value for `files_total`, must be a value greater than or equal to `0`" + ) # noqa: E501 + + self._files_total = files_total + + @property + def files_done(self) -> int: + """Gets the files_done of this IngestJob. + + Members ingested so far # noqa: E501 + + :return: The files_done of this IngestJob. + :rtype: int + """ + return self._files_done + + @files_done.setter + def files_done(self, files_done: int): + """Sets the files_done of this IngestJob. + + Members ingested so far # noqa: E501 + + :param files_done: The files_done of this IngestJob. + :type files_done: int + """ + if files_done is None: + raise ValueError( + "Invalid value for `files_done`, must not be `None`" + ) # noqa: E501 + if files_done is not None and files_done > 2147483647: # noqa: E501 + raise ValueError( + "Invalid value for `files_done`, must be a value less than or equal to `2147483647`" + ) # noqa: E501 + if files_done is not None and files_done < 0: # noqa: E501 + raise ValueError( + "Invalid value for `files_done`, must be a value greater than or equal to `0`" + ) # noqa: E501 + + self._files_done = files_done diff --git a/tagbase_server/tagbase_server/openapi/openapi.yaml b/tagbase_server/tagbase_server/openapi/openapi.yaml index 7e3b1146..fbe111a7 100644 --- a/tagbase_server/tagbase_server/openapi/openapi.yaml +++ b/tagbase_server/tagbase_server/openapi/openapi.yaml @@ -70,20 +70,24 @@ paths: - $ref: "#/components/parameters/version" requestBody: content: + application/gzip: + schema: + $ref: "#/components/schemas/SyncUpload" application/octet-stream: schema: - description: Compressed binary e.g. zip file containing one or more eTUFF files - format: binary - maxLength: 1000000000 - minLength: 1 - type: string + $ref: "#/components/schemas/SyncUpload" + application/x-gzip: + schema: + $ref: "#/components/schemas/SyncUpload" + application/x-tar: + schema: + $ref: "#/components/schemas/SyncUpload" + application/zip: + schema: + $ref: "#/components/schemas/SyncUpload" text/plain: schema: - description: Plain text eTUFF file - format: binary - maxLength: 1000000000 - minLength: 1 - type: string + $ref: "#/components/schemas/SyncUpload" required: true x-body-name: body responses: @@ -95,6 +99,8 @@ paths: description: A success message confirming ingestion. "400": $ref: "#/components/responses/BadRequest" + "413": + $ref: "#/components/responses/PayloadTooLarge" "500": $ref: "#/components/responses/InternalError" summary: Post a local file and perform a ingest operation @@ -102,6 +108,79 @@ paths: - ingest x-codegen-request-body-name: body x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs: + post: + description: > + Accept a .txt, .zip, .tar.gz, or .tgz upload up to 2 GiB, store it, + and enqueue ingest. Poll GET /ingest/jobs/{job_id} until the job + succeeds or fails. + operationId: create_ingest_job + parameters: + - $ref: "#/components/parameters/filename" + - $ref: "#/components/parameters/notes" + - $ref: "#/components/parameters/type" + - $ref: "#/components/parameters/version" + requestBody: + content: + application/gzip: + schema: + $ref: "#/components/schemas/JobUpload" + application/octet-stream: + schema: + $ref: "#/components/schemas/JobUpload" + application/x-gzip: + schema: + $ref: "#/components/schemas/JobUpload" + application/x-tar: + schema: + $ref: "#/components/schemas/JobUpload" + application/zip: + schema: + $ref: "#/components/schemas/JobUpload" + text/plain: + schema: + $ref: "#/components/schemas/JobUpload" + required: true + x-body-name: body + responses: + "202": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestAccepted" + description: Request accepted. Ingest job enqueued + "400": + $ref: "#/components/responses/BadRequest" + "413": + $ref: "#/components/responses/PayloadTooLarge" + "500": + $ref: "#/components/responses/InternalError" + summary: Enqueue a large file for ingest + tags: + - ingest + x-codegen-request-body-name: body + x-openapi-router-controller: tagbase_server.controllers.ingest_controller + /ingest/jobs/{job_id}: + parameters: + - $ref: "#/components/parameters/jobId" + get: + description: Read the status of one ingest job. + operationId: get_ingest_job + responses: + "200": + content: + application/json: + schema: + $ref: "#/components/schemas/IngestJob" + description: The current ingest job. + "404": + $ref: "#/components/responses/NotFound" + "500": + $ref: "#/components/responses/InternalError" + summary: Read one ingest job + tags: + - ingest + x-openapi-router-controller: tagbase_server.controllers.ingest_controller /tags: delete: description: Delete all tags @@ -205,6 +284,18 @@ components: application/problem+json: schema: $ref: "#/components/schemas/Problem" + NotFound: + description: The requested resource does not exist. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" + PayloadTooLarge: + description: The request body exceeds the limit for this operation. + content: + application/problem+json: + schema: + $ref: "#/components/schemas/Problem" InternalError: description: Unexpected server error. Contact admin detailed in openapi.yaml. content: @@ -212,6 +303,19 @@ components: schema: $ref: "#/components/schemas/Problem" parameters: + jobId: + description: Ingest job id + explode: false + in: path + name: job_id + required: true + schema: + format: uuid + maxLength: 36 + minLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + type: string + style: simple filename: description: Free-form text field to explicitly define the name of the file to be persisted explode: true @@ -398,6 +502,91 @@ components: example: 4bf92f3577b34da6a3ce929d0e0e4736 title: Ingest200 type: object + SyncUpload: + description: Synchronous ingest body, at most 100 MiB + format: binary + maxLength: 104857600 + minLength: 1 + type: string + JobUpload: + description: Ingest job body, at most 2 GiB + format: binary + maxLength: 2147483648 + minLength: 1 + type: string + IngestAccepted: + description: HTTP 202 response after an ingest job is queued + required: + - id + - status + example: + id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: queued + properties: + id: + description: Ingest job id + format: uuid + type: string + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: + description: Job status at accept time + type: string + minLength: 6 + maxLength: 9 + pattern: ^queued$ + example: queued + title: IngestAccepted + type: object + IngestJob: + description: Status of one ingest job + required: + - id + - status + - files_done + example: + id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: succeeded + files_done: 1 + files_total: 1 + properties: + id: + description: Ingest job id + format: uuid + type: string + minLength: 36 + maxLength: 36 + pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$ + example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736 + status: + description: queued, running, succeeded, or failed + type: string + minLength: 6 + maxLength: 9 + pattern: ^(queued|running|succeeded|failed)$ + example: succeeded + error: + description: Present when the job failed + type: string + minLength: 1 + maxLength: 10000 + pattern: ^(?!\s*$).+$ + files_total: + description: eTUFF members in the upload + type: integer + format: int32 + minimum: 0 + maximum: 10000 + files_done: + description: eTUFF members already processed + type: integer + format: int32 + minimum: 0 + maximum: 10000 + title: IngestJob + type: object Metadata: description: Contains the ingested tag metadata consistent with the eTUFF specification example: diff --git a/tagbase_server/tagbase_server/test/conftest.py b/tagbase_server/tagbase_server/test/conftest.py index 701ac5d5..632da5a1 100644 --- a/tagbase_server/tagbase_server/test/conftest.py +++ b/tagbase_server/tagbase_server/test/conftest.py @@ -11,6 +11,7 @@ from tagbase_server import encoder from tagbase_server.problem import register_problem_handlers +from tagbase_server.upload_stream import UploadGuardMiddleware OPENAPI_DIR = Path(__file__).resolve().parents[1] / "openapi" _FIXTURES = Path(__file__).resolve().parent / "fixtures" @@ -41,6 +42,7 @@ def create_test_app(): swagger_ui_options=swagger_ui_options, jsonifier=encoder.create_jsonifier(), ) + app.add_middleware(UploadGuardMiddleware) app.add_api("openapi.yaml", pythonic_params=True, strict_validation=True) register_problem_handlers(app.app) return app @@ -117,6 +119,14 @@ def clean_db(db_conn): with db_conn.cursor() as cur: cur.execute("TRUNCATE submission CASCADE") cur.execute("TRUNCATE dataset CASCADE") + cur.execute(""" + DO $$ + BEGIN + IF to_regclass('public.ingest_job') IS NOT NULL THEN + TRUNCATE ingest_job; + END IF; + END $$; + """) for attribute_id, category, name, description, necessity in METADATA_TYPE_SEED: cur.execute( """ diff --git a/tagbase_server/tagbase_server/test/helpers.py b/tagbase_server/tagbase_server/test/helpers.py index 7280e7a1..dcbf9079 100644 --- a/tagbase_server/tagbase_server/test/helpers.py +++ b/tagbase_server/tagbase_server/test/helpers.py @@ -20,3 +20,31 @@ def response_json(response): return json.loads(response.content) + + +def assert_sync_ingest(response): + """Synchronous /ingest returns 200 and a trace id in the request.""" + assert response.status_code == 200, response.content + body = response_json(response) + assert body["code"] == "200" + assert len(body["trace_id"]) == 32 + return body + + +def finish_accepted_ingest(client, response): + """Assert the upload was accepted, then run queued jobs in this process.""" + from tagbase_server.ingest_worker import process_available + + assert response.status_code == 202 + accepted = response_json(response) + assert accepted["status"] == "queued" + assert accepted["id"] + process_available() + job = client.get( + f"{API_PREFIX}/ingest/jobs/{accepted['id']}", + headers={"Accept": "application/json"}, + ) + assert job.status_code == 200 + body = response_json(job) + assert body["status"] == "succeeded", body + return body diff --git a/tagbase_server/tagbase_server/test/integration/test_ingest_api.py b/tagbase_server/tagbase_server/test/integration/test_ingest_api.py index 70f25d22..95118555 100644 --- a/tagbase_server/tagbase_server/test/integration/test_ingest_api.py +++ b/tagbase_server/tagbase_server/test/integration/test_ingest_api.py @@ -5,7 +5,13 @@ import pytest -from tagbase_server.test.helpers import API_PREFIX, ETUFF_FIXTURE, response_json +from tagbase_server.test.helpers import ( + API_PREFIX, + ETUFF_FIXTURE, + assert_sync_ingest, + finish_accepted_ingest, + response_json, +) pytestmark = pytest.mark.integration @@ -34,10 +40,7 @@ def test_post_ingest_etuff_makes_tag_listable_and_gettable( }, data=etuff_bytes, ) - assert response.status_code == 200 - body = response_json(response) - assert body["code"] == "200" - assert "minimal-etuff.txt" in body["message"] + assert_sync_ingest(response) listed = client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) assert listed.status_code == 200 @@ -52,7 +55,7 @@ def test_post_ingest_etuff_makes_tag_listable_and_gettable( assert detail.status_code == 200 tag = response_json(detail)["tag"] assert len(tag) == 1 - assert tag[0]["filename"].endswith("minimal-etuff.txt") + assert tag[0]["filename"] == "minimal-etuff.txt" assert tag[0]["notes"] == "integration ingest notes" assert tag[0]["version"] == "1" assert tag[0]["metadata"]["ptt"] == "117464" @@ -70,8 +73,8 @@ def test_post_ingest_duplicate_etuff_keeps_single_submission( second = client.post( f"{API_PREFIX}/ingest", params=params, headers=headers, data=etuff_bytes ) - assert first.status_code == 200 - assert second.status_code == 200 + assert_sync_ingest(first) + assert_sync_ingest(second) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) @@ -97,7 +100,7 @@ def test_post_ingest_zip_etuff_makes_tag_retrievable(client, clean_db, etuff_zip }, data=etuff_zip_bytes, ) - assert response.status_code == 200 + assert_sync_ingest(response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) @@ -114,7 +117,7 @@ def test_get_ingest_file_url_makes_tag_listable(client, clean_db, short_etuff_fi }, headers={"Accept": "application/json"}, ) - assert response.status_code == 200 + assert_sync_ingest(response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) @@ -162,8 +165,25 @@ def test_ingest_defaults_omitted_type_to_etuff(client, clean_db, etuff_bytes): }, data=etuff_bytes, ) - assert response.status_code == 200 + assert_sync_ingest(response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) assert listed["count"] == 1 + + +def test_post_ingest_job_zip_reaches_succeeded(client, clean_db, etuff_zip_bytes): + response = client.post( + f"{API_PREFIX}/ingest/jobs", + params={"filename": "minimal-etuff.zip", "version": "1"}, + headers={ + "Accept": "application/json", + "Content-Type": "application/zip", + }, + data=etuff_zip_bytes, + ) + finish_accepted_ingest(client, response) + listed = response_json( + client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) + ) + assert listed["tags"][0]["filename"] == "minimal-etuff.txt" diff --git a/tagbase_server/tagbase_server/test/integration/test_tags_api.py b/tagbase_server/tagbase_server/test/integration/test_tags_api.py index 18037d70..f3aa06b0 100644 --- a/tagbase_server/tagbase_server/test/integration/test_tags_api.py +++ b/tagbase_server/tagbase_server/test/integration/test_tags_api.py @@ -2,7 +2,11 @@ import pytest -from tagbase_server.test.helpers import API_PREFIX, response_json +from tagbase_server.test.helpers import ( + API_PREFIX, + assert_sync_ingest, + response_json, +) pytestmark = pytest.mark.integration @@ -21,7 +25,7 @@ def _ingest(client, etuff_bytes, notes="notes", version="1"): }, data=etuff_bytes, ) - assert response.status_code == 200 + assert_sync_ingest(response) listed = response_json( client.get(f"{API_PREFIX}/tags", headers={"Accept": "application/json"}) ) diff --git a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py index 4fd6d5bc..f607082b 100644 --- a/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py +++ b/tagbase_server/tagbase_server/test/stack/test_fswatch_contract.py @@ -69,6 +69,8 @@ class _IngestHandler(BaseHTTPRequestHandler): requests = [] + job_status = "succeeded" + get_status = 200 def do_POST(self): # noqa: N802 length = int(self.headers.get("Content-Length", "0")) @@ -76,15 +78,40 @@ def do_POST(self): # noqa: N802 parsed = urlparse(self.path) self.requests.append( { + "method": "POST", "path": parsed.path, "query": parse_qs(parsed.query), "body": body, } ) - self.send_response(200) + self.send_response(202) self.send_header("Content-Type", "application/json") self.end_headers() - self.wfile.write(b'{"code":"200","message":"ok","elapsed":"0"}') + self.wfile.write( + b'{"id":"11111111-1111-1111-1111-111111111111","status":"queued"}' + ) + + def do_GET(self): # noqa: N802 + parsed = urlparse(self.path) + if not parsed.path.endswith("/11111111-1111-1111-1111-111111111111"): + self.send_response(404) + self.end_headers() + return + status = type(self).job_status + code = type(self).get_status + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.end_headers() + if code != 200: + self.wfile.write(b'{"title":"unavailable"}') + return + error = "" if status != "failed" else "parse failed" + payload = ( + '{"id":"11111111-1111-1111-1111-111111111111",' + f'"status":"{status}","error":"{error}",' + '"files_total":1,"files_done":0}' + ) + self.wfile.write(payload.encode()) def log_message(self, format, *args): # noqa: A003 return @@ -93,6 +120,8 @@ def log_message(self, format, *args): # noqa: A003 @pytest.fixture def mock_ingest_server(): _IngestHandler.requests = [] + _IngestHandler.job_status = "succeeded" + _IngestHandler.get_status = 200 server = HTTPServer(("127.0.0.1", 0), _IngestHandler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() @@ -121,7 +150,7 @@ def _install_fake_fswatch(work): def _run_fswatch(env, log_path): log_f = open(log_path, "w", encoding="utf-8") proc = subprocess.Popen( - ["sh", str(POST_SH)], + ["bash", str(POST_SH)], env=env, stdout=log_f, stderr=subprocess.STDOUT, @@ -212,7 +241,7 @@ def test_fswatch_posts_basename_and_body_after_size_stable(mock_ingest_server): f"fswatch did not POST to ingest. post.sh output:\n{_read_log(log_path)}" ) req = handler.requests[0] - assert req["path"].endswith("/ingest") + assert req["path"].endswith("/ingest/jobs") assert req["query"]["filename"] == ["drop-etuff.txt"] assert req["query"]["type"] == ["etuff"] assert req["body"] == b"partial-complete" @@ -287,3 +316,76 @@ def test_fswatch_ingests_multiple_files_dropped_quickly(mock_ingest_server): finally: _stop_fswatch(proc) shutil.rmtree(work, ignore_errors=True) + + +def test_fswatch_keeps_submitting_while_a_job_is_running(mock_ingest_server): + ingest_base, handler = mock_ingest_server + handler.job_status = "running" + work = _workspace_tmpdir("running") + staging = work / "staging_data" + staging.mkdir() + bin_dir = _install_fake_fswatch(work) + log_path = work / "fswatch.log" + env = _clean_env(os.environ, staging, work / "dedupe", ingest_base, bin_dir) + proc = _run_fswatch(env, log_path) + try: + time.sleep(0.4) + (staging / "first.txt").write_bytes(b"one") + (staging / "second.txt").write_bytes(b"two") + # record_pending runs after the POST returns, so wait for the file too. + deadline = time.time() + 30 + pending_path = work / "pending.tsv" + pending = "" + while time.time() < deadline: + names = { + (r["query"].get("filename") or [None])[0] for r in handler.requests + } + if pending_path.is_file(): + pending = pending_path.read_text(encoding="utf-8") + if ( + names >= {"first.txt", "second.txt"} + and "first.txt" in pending + and "second.txt" in pending + ): + break + time.sleep(0.25) + else: + pytest.fail( + "expected both posts to stay pending while running. " + f"pending:\n{pending}\nlog:\n{_read_log(log_path)}" + ) + finally: + _stop_fswatch(proc) + shutil.rmtree(work, ignore_errors=True) + + +def test_fswatch_reconciles_pending_across_restart(mock_ingest_server): + ingest_base, handler = mock_ingest_server + work = _workspace_tmpdir("restart") + staging = work / "staging_data" + staging.mkdir() + target = staging / "already.txt" + target.write_bytes(b"kept") + pending = work / "pending.tsv" + pending.write_text( + f"{target}\t{target.stat().st_size}\t11111111-1111-1111-1111-111111111111\n", + encoding="utf-8", + ) + bin_dir = _install_fake_fswatch(work) + log_path = work / "fswatch.log" + env = _clean_env(os.environ, staging, work / "dedupe", ingest_base, bin_dir) + proc = _run_fswatch(env, log_path) + try: + deadline = time.time() + 20 + done = work / "dedupe" + while time.time() < deadline: + if done.is_file() and "already.txt" in done.read_text(encoding="utf-8"): + break + time.sleep(0.25) + else: + pytest.fail(f"pending job was not reconciled. log:\n{_read_log(log_path)}") + assert handler.requests == [] + assert "already.txt" not in pending.read_text(encoding="utf-8") + finally: + _stop_fswatch(proc) + shutil.rmtree(work, ignore_errors=True) diff --git a/tagbase_server/tagbase_server/test/test_archives.py b/tagbase_server/tagbase_server/test/test_archives.py new file mode 100644 index 00000000..7379a3ec --- /dev/null +++ b/tagbase_server/tagbase_server/test/test_archives.py @@ -0,0 +1,171 @@ +# coding: utf-8 + +import io +import os +import stat +import tarfile +import zipfile + +import pytest + +from tagbase_server.archives import ( + ArchiveError, + archive_limits, + materialize, + preflight, +) + + +def _zip(path, members): + with zipfile.ZipFile(path, "w") as archive: + for info, payload in members: + archive.writestr(info, payload) + + +def test_archive_limits_default_to_ten_thousand_and_ten_gib(): + members, expanded = archive_limits() + assert members == 10_000 + assert expanded == 10 * 1024 * 1024 * 1024 + + +def test_plain_text_materialize_returns_the_file(tmp_path): + path = tmp_path / "tag.txt" + path.write_text("etuff", encoding="utf-8") + prepared = materialize(str(path), "tag.txt") + assert prepared.files == [str(path)] + assert prepared.submission_names() == ["tag.txt"] + assert prepared.workspace is None + + +def test_staged_text_keeps_the_logical_name(tmp_path): + path = tmp_path / "5fb7423d15d9452dbfdc81d9a1c2c995-tagbase-upload-cjz5__3y" + path.write_text("etuff", encoding="utf-8") + prepared = materialize(str(path), "2024BFT07799A.txt") + assert prepared.files == [str(path)] + assert prepared.submission_names() == ["2024BFT07799A.txt"] + + +def test_zip_extracts_txt_and_cleanup_removes_workspace(tmp_path): + archive = tmp_path / "bundle.zip" + _zip(archive, [(zipfile.ZipInfo("inner.txt"), b"etuff")]) + prepared = materialize(str(archive), "bundle.zip") + try: + assert len(prepared.files) == 1 + assert prepared.files[0].endswith("inner.txt") + assert prepared.submission_names() == ["inner.txt"] + assert os.path.isdir(prepared.workspace) + finally: + workspace = prepared.workspace + prepared.cleanup() + assert not os.path.exists(workspace) + + +def test_zip_member_name_keeps_its_directory(tmp_path): + archive = tmp_path / "bundle.zip" + _zip(archive, [(zipfile.ZipInfo("cruise/inner.txt"), b"etuff")]) + prepared = materialize(str(archive), "bundle.zip") + try: + assert prepared.submission_names() == ["cruise/inner.txt"] + finally: + prepared.cleanup() + + +def test_tar_gz_and_tgz_extract_txt(tmp_path): + for name in ("bundle.tar.gz", "bundle.tgz"): + archive = tmp_path / name + with tarfile.open(archive, "w:gz") as tar: + payload = b"etuff" + info = tarfile.TarInfo("inner.txt") + info.size = len(payload) + tar.addfile(info, io.BytesIO(payload)) + prepared = materialize(str(archive), name) + try: + assert prepared.files[0].endswith("inner.txt") + assert prepared.submission_names() == ["inner.txt"] + finally: + prepared.cleanup() + + +def test_rejects_path_traversal_symlink_and_encrypted_zip(tmp_path): + traversal = tmp_path / "traversal.zip" + info = zipfile.ZipInfo("../evil.txt") + _zip(traversal, [(info, b"x")]) + with pytest.raises(ArchiveError, match="path"): + preflight(str(traversal), "traversal.zip") + + link = tmp_path / "link.zip" + info = zipfile.ZipInfo("link.txt") + info.external_attr = stat.S_IFLNK << 16 + _zip(link, [(info, b"target")]) + with pytest.raises(ArchiveError, match="symlink"): + preflight(str(link), "link.zip") + + encrypted = tmp_path / "secret.zip" + _zip(encrypted, [(zipfile.ZipInfo("secret.txt"), b"x")]) + raw = bytearray(encrypted.read_bytes()) + raw[6] |= 0x1 + central = raw.find(b"PK\x01\x02") + raw[central + 8] |= 0x1 + encrypted.write_bytes(raw) + with pytest.raises(ArchiveError, match="encrypted"): + preflight(str(encrypted), "secret.zip") + + +def test_rejects_tar_traversal_and_symlink(tmp_path): + traversal = tmp_path / "traversal.tar.gz" + with tarfile.open(traversal, "w:gz") as tar: + info = tarfile.TarInfo("../evil.txt") + info.size = 1 + tar.addfile(info, io.BytesIO(b"x")) + with pytest.raises(ArchiveError, match="path"): + preflight(str(traversal), "traversal.tar.gz") + + link = tmp_path / "link.tgz" + with tarfile.open(link, "w:gz") as tar: + info = tarfile.TarInfo("link.txt") + info.type = tarfile.SYMTYPE + info.linkname = "inner.txt" + tar.addfile(info) + with pytest.raises(ArchiveError, match="symlink"): + preflight(str(link), "link.tgz") + + +def test_member_and_expanded_size_caps(tmp_path, monkeypatch): + monkeypatch.setenv("TAGBASE_ARCHIVE_MAX_MEMBERS", "1") + archive = tmp_path / "many.zip" + _zip( + archive, + [ + (zipfile.ZipInfo("a.txt"), b"a"), + (zipfile.ZipInfo("b.txt"), b"b"), + ], + ) + with pytest.raises(ArchiveError, match="too many members"): + preflight(str(archive), "many.zip") + + monkeypatch.setenv("TAGBASE_ARCHIVE_MAX_MEMBERS", "10") + monkeypatch.setenv("TAGBASE_ARCHIVE_MAX_UNCOMPRESSED", "4") + large = tmp_path / "large.zip" + _zip(large, [(zipfile.ZipInfo("big.txt"), b"12345")]) + with pytest.raises(ArchiveError, match="uncompressed"): + preflight(str(large), "large.zip") + + +def test_rejects_corrupt_empty_and_mismatched_types(tmp_path): + corrupt = tmp_path / "bad.zip" + corrupt.write_bytes(b"not-a-zip") + with pytest.raises(ArchiveError): + preflight(str(corrupt), "bad.zip") + + empty = tmp_path / "empty.txt" + empty.write_bytes(b"") + with pytest.raises(ArchiveError, match="empty"): + preflight(str(empty), "empty.txt") + + disguised = tmp_path / "notes.txt" + disguised.write_bytes(b"PK\x03\x04nope") + with pytest.raises(ArchiveError, match="do not match"): + preflight(str(disguised), "notes.txt") + + with pytest.raises(ArchiveError, match="unsupported"): + preflight(str(disguised), "notes.csv") diff --git a/tagbase_server/tagbase_server/test/test_ingest_jobs.py b/tagbase_server/tagbase_server/test/test_ingest_jobs.py new file mode 100644 index 00000000..cc94682a --- /dev/null +++ b/tagbase_server/tagbase_server/test/test_ingest_jobs.py @@ -0,0 +1,141 @@ +# coding: utf-8 + +from unittest.mock import patch + +import pytest + +from tagbase_server.archives import PreparedIngest +from tagbase_server.ingest_jobs import ( + _CLAIM_SQL, + DEFAULT_RETENTION_DAYS, + MISSING_FILE_ERROR, + RESTART_ERROR, + retention_days, + stale_failures, +) +from tagbase_server.ingest_worker import process_available, process_next, run_job + + +def test_stale_failures_marks_running_and_missing_queued_files(): + jobs = [ + {"id": "running-1", "status": "running", "source_path": "/tmp/a.zip"}, + {"id": "queued-gone", "status": "queued", "source_path": "/tmp/missing.zip"}, + {"id": "queued-kept", "status": "queued", "source_path": "/tmp/kept.zip"}, + {"id": "done", "status": "succeeded", "source_path": "/tmp/done.zip"}, + ] + + failed = stale_failures(jobs, lambda path: path == "/tmp/kept.zip") + + assert failed == [ + ("running-1", RESTART_ERROR), + ("queued-gone", MISSING_FILE_ERROR), + ] + + +def test_run_job_stops_on_the_first_member_error(): + job = { + "id": "job-1", + "source_path": "/tmp/archive.zip", + "version": "1", + "notes": None, + } + seen = [] + + def _ingest( + path, version=None, notes=None, count_request=True, submission_name=None + ): + seen.append((path, count_request, submission_name)) + if submission_name == "b.txt": + raise RuntimeError("bad member") + + prepared = PreparedIngest( + files=["/tmp/staged-a", "/tmp/staged-b", "/tmp/staged-c"], + workspace=None, + names=["a.txt", "b.txt", "c.txt"], + ) + with ( + patch("tagbase_server.ingest_worker.materialize", return_value=prepared), + patch("tagbase_server.ingest_worker.process_etuff_file", side_effect=_ingest), + patch("tagbase_server.ingest_worker.set_progress") as progress, + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + with pytest.raises(RuntimeError, match="bad member"): + run_job(job) + + assert [path for path, _counted, _name in seen] == [ + "/tmp/staged-a", + "/tmp/staged-b", + ] + assert [name for _path, _counted, name in seen] == ["a.txt", "b.txt"] + assert all(counted is False for _path, counted, _name in seen) + finish.assert_not_called() + assert progress.call_args_list[-1].kwargs["files_done"] == 1 + + +def test_process_next_fails_the_job_and_keeps_the_queue_moving(): + job = {"id": "job-2", "source_path": "/tmp/one.txt", "version": None, "notes": None} + with ( + patch("tagbase_server.ingest_worker.claim_next_job", return_value=job), + patch( + "tagbase_server.ingest_worker.run_job", + side_effect=RuntimeError("stopped"), + ), + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + assert process_next() is True + + finish.assert_called_once_with("job-2", status="failed", error="stopped") + + +def test_claim_orders_by_created_at_then_updated_at(): + assert "ORDER BY created_at, updated_at" in _CLAIM_SQL + assert "FOR UPDATE SKIP LOCKED" in _CLAIM_SQL + + +def test_retention_defaults_to_thirty_days(monkeypatch): + monkeypatch.delenv("INGEST_JOB_RETENTION_DAYS", raising=False) + assert DEFAULT_RETENTION_DAYS == 30 + assert retention_days() == 30 + + +def test_process_available_runs_two_queued_jobs(): + jobs = iter( + [ + {"id": "job-a", "source_path": "/tmp/a.txt", "filename": "a.txt"}, + {"id": "job-b", "source_path": "/tmp/b.txt", "filename": "b.txt"}, + ] + ) + + def claim(): + return next(jobs, None) + + with ( + patch("tagbase_server.ingest_worker.recover_stale_jobs"), + patch("tagbase_server.ingest_worker.claim_next_job", side_effect=claim), + patch("tagbase_server.ingest_worker.run_job"), + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + assert process_available() == 2 + + assert [call.kwargs["status"] for call in finish.call_args_list] == [ + "succeeded", + "succeeded", + ] + + +def test_process_next_deletes_the_staged_file_after_success(tmp_path): + source = tmp_path / "one.txt" + source.write_text("etuff", encoding="utf-8") + job = {"id": "job-3", "source_path": str(source), "filename": "one.txt"} + claims = iter([job]) + with ( + patch( + "tagbase_server.ingest_worker.claim_next_job", + side_effect=lambda: next(claims, None), + ), + patch("tagbase_server.ingest_worker.run_job"), + patch("tagbase_server.ingest_worker.finish_job") as finish, + ): + assert process_next() is True + finish.assert_called_once_with("job-3", status="succeeded", error=None) + assert not source.exists() diff --git a/tagbase_server/tagbase_server/test/test_sonar_cleanup.py b/tagbase_server/tagbase_server/test/test_sonar_cleanup.py index 848e9d6a..19f547ab 100644 --- a/tagbase_server/tagbase_server/test/test_sonar_cleanup.py +++ b/tagbase_server/tagbase_server/test/test_sonar_cleanup.py @@ -12,7 +12,7 @@ from tagbase_server.controllers import ingest_controller from tagbase_server.controllers.ingest_controller import _resolve_ingest_file_type from tagbase_server.models.base_model_ import Model -from tagbase_server.models.ingest200 import Ingest200 +from tagbase_server.models.tag_put200 import TagPut200 from tagbase_server.utils import io_utils from tagbase_server.utils import processing_utils as pu @@ -86,23 +86,6 @@ def test_process_get_input_data_local_file(tmp_path): assert result == str(local) -@mock.patch("tagbase_server.utils.io_utils.Archive") -def test_unpack_compressed_binary_uses_temp_dir(mock_archive, tmp_path): - archive_path = tmp_path / "bundle.zip" - archive_path.write_bytes(b"x") - - def extractall(target): - assert target.startswith(tempfile.gettempdir()) - nested = os.path.join(target, "inner.txt") - with open(nested, "w", encoding="utf-8") as handle: - handle.write("etuff") - - mock_archive.return_value.extractall.side_effect = extractall - files = io_utils.unpack_compressed_binary(str(archive_path)) - assert len(files) == 1 - assert files[0].endswith("inner.txt") - - @mock.patch("tagbase_server.utils.io_utils.urlopen") def test_process_get_input_data_downloads_remote(mock_urlopen, tmp_path): mock_urlopen.return_value.read.side_effect = [b"chunk", b""] @@ -125,34 +108,48 @@ def test_resolve_ingest_file_type_defaults_and_rejects(): _resolve_ingest_file_type("csv") -@mock.patch("tagbase_server.controllers.ingest_controller.parmap.map") +@mock.patch("tagbase_server.controllers.ingest_controller.record_ingest_request") +@mock.patch("tagbase_server.controllers.ingest_controller._run_ingest_queue") +@mock.patch("tagbase_server.controllers.ingest_controller.materialize") @mock.patch("tagbase_server.controllers.ingest_controller.process_get_input_data") -def test_ingest_get_happy_path(mock_get, mock_map): +def test_ingest_get_happy_path(mock_get, mock_materialize, mock_queue, mock_record): + from tagbase_server.archives import PreparedIngest + mock_get.return_value = "/data/file.txt" - mock_map.return_value = [0] + mock_materialize.return_value = PreparedIngest( + files=["/data/file.txt"], workspace=None + ) result, status, headers = ingest_controller.ingest_get( "file:///data/file.txt", type="etuff" ) assert status == 200 assert headers["Content-Type"] == "application/json" assert result.code == "200" - mock_map.assert_called_once() + assert len(result.trace_id) == 32 + mock_queue.assert_called_once() + mock_record.assert_called_once() + assert mock_record.call_args.args[0] == "success" -@mock.patch("tagbase_server.controllers.ingest_controller.parmap.map") -@mock.patch("tagbase_server.controllers.ingest_controller.unpack_compressed_binary") +@mock.patch("tagbase_server.controllers.ingest_controller.record_ingest_request") +@mock.patch("tagbase_server.controllers.ingest_controller._run_ingest_queue") +@mock.patch("tagbase_server.controllers.ingest_controller.materialize") @mock.patch("tagbase_server.controllers.ingest_controller.process_post_input_data") -def test_ingest_post_archive_path(mock_post, mock_unpack, mock_map): - mock_post.return_value = "/data/bundle.zip" - mock_unpack.return_value = ["/data/a.txt"] - mock_map.return_value = [0] +def test_ingest_post_archive_path(mock_post, mock_materialize, mock_queue, mock_record): + from tagbase_server.archives import PreparedIngest + + mock_post.return_value = "/tmp/bundle.zip" + mock_materialize.return_value = PreparedIngest( + files=["/tmp/inner.txt"], workspace=None + ) result, status, headers = ingest_controller.ingest_post( "bundle.zip", b"x", type=None ) assert status == 200 assert headers["Content-Type"] == "application/json" assert result.code == "200" - mock_unpack.assert_called_once() + mock_queue.assert_called_once() + mock_record.assert_called_once() def test_processing_utils_has_no_pytz(): @@ -280,7 +277,15 @@ def test_process_etuff_file_full_path( mock_build.return_value = ([[timezone.utc, 1, "1", 3, 2]], 1) mock_buffer.return_value = StringIO("x") mock_migrate.return_value = 0.1 - assert pu.process_etuff_file("f.txt") is None + assert ( + pu.process_etuff_file( + "/tmp/5fb7423d-tagbase-upload-cjz5", + submission_name="2024BFT07799A.txt", + ) + is None + ) + assert mock_props.call_args.args[0] == "/tmp/5fb7423d-tagbase-upload-cjz5" + assert mock_insert.call_args.args[2] == "2024BFT07799A.txt" @mock.patch("tagbase_server.utils.processing_utils.connect") @@ -333,7 +338,7 @@ def __init__(self): assert as_dict["items"][0]["name"] == "a" assert as_dict["nested"]["name"] == "b" assert as_dict["mapping"]["k"]["name"] == "c" - a = Ingest200.from_dict({"code": "200", "elapsed": 1.0, "message": "a"}) - b = Ingest200.from_dict({"code": "200", "elapsed": 1.0, "message": "b"}) + a = TagPut200.from_dict({"code": "200", "message": "a"}) + b = TagPut200.from_dict({"code": "200", "message": "b"}) assert a != b assert not (a != a) diff --git a/tagbase_server/tagbase_server/test/test_upload_stream.py b/tagbase_server/tagbase_server/test/test_upload_stream.py new file mode 100644 index 00000000..51492ad0 --- /dev/null +++ b/tagbase_server/tagbase_server/test/test_upload_stream.py @@ -0,0 +1,87 @@ +# coding: utf-8 + +import asyncio +import os + +from tagbase_server.api_prefix import API_PREFIX +from tagbase_server.test.conftest import create_test_app +from tagbase_server.upload_stream import ( + JOB_MAX_BYTES, + SYNC_MAX_BYTES, + UploadGuardMiddleware, + staged_upload, +) + + +def test_default_upload_caps(): + assert SYNC_MAX_BYTES == 100 * 1024 * 1024 + assert JOB_MAX_BYTES == 2 * 1024 * 1024 * 1024 + + +def test_sync_post_above_cap_is_413(monkeypatch): + monkeypatch.setenv("TAGBASE_SYNC_MAX_BYTES", "4") + app = create_test_app() + with app.test_client() as client: + response = client.post( + f"{API_PREFIX}/ingest", + params={"filename": "a.txt"}, + headers={"Content-Type": "text/plain", "Accept": "application/json"}, + data=b"12345", + ) + assert response.status_code == 413 + assert response.headers["content-type"].startswith("application/problem+json") + + +def _run(middleware, body, limit_env="8"): + captured = {} + + async def app(scope, receive, send): + message = await receive() + captured["body"] = message.get("body") + captured["staged"] = staged_upload.get() + if captured["staged"]: + with open(captured["staged"], "rb") as handle: + captured["file"] = handle.read() + await send( + { + "type": "http.response.start", + "status": 202, + "headers": [(b"content-type", b"application/json")], + } + ) + await send({"type": "http.response.body", "body": b'{"status":"queued"}'}) + + scope = { + "type": "http", + "method": "POST", + "path": "/tagbase/api/v0/ingest/jobs", + "headers": [(b"content-length", str(len(body)).encode("ascii"))], + } + + async def receive(): + return {"type": "http.request", "body": body, "more_body": False} + + sent = [] + + async def send(message): + sent.append(message) + + asyncio.run(UploadGuardMiddleware(app)(scope, receive, send)) + return captured, sent + + +def test_job_upload_streams_without_buffering_the_body(monkeypatch, tmp_path): + monkeypatch.setenv("TAGBASE_JOB_MAX_BYTES", "8") + monkeypatch.setenv("TMPDIR", str(tmp_path)) + captured, sent = _run(None, b"12345678") + assert captured["body"] == b"staged" + assert captured["file"] == b"12345678" + assert sent[0]["status"] == 202 + os.remove(captured["staged"]) + + +def test_job_upload_one_byte_over_is_413(monkeypatch): + monkeypatch.setenv("TAGBASE_JOB_MAX_BYTES", "8") + captured, sent = _run(None, b"123456789") + assert captured == {} + assert sent[0]["status"] == 413 diff --git a/tagbase_server/tagbase_server/upload_stream.py b/tagbase_server/tagbase_server/upload_stream.py new file mode 100644 index 00000000..4f325d5c --- /dev/null +++ b/tagbase_server/tagbase_server/upload_stream.py @@ -0,0 +1,289 @@ +"""Stream ingest uploads before Connexion buffers the request body. + +Synchronous POST /ingest is rejected above 100 MiB. POST /ingest/jobs is +written to a staging file in chunks, with a 2 GiB cap, and the application +then sees a short substitute body. +""" + +from __future__ import annotations + +import contextvars +import json +import os +import tempfile +from collections.abc import Callable + +from tagbase_server.problem import TYPE_HTTP, current_trace_id +from tagbase_server.utils.io_utils import TEMP_DIR + +SYNC_MAX_BYTES = 100 * 1024 * 1024 +JOB_MAX_BYTES = 2 * 1024 * 1024 * 1024 +_STAGED_BODY = b"staged" +_CHUNK = 1024 * 1024 +_HTTP_REQUEST = "http.request" +_TITLE_BAD_REQUEST = "Bad Request" +_TITLE_TOO_LARGE = "Payload Too Large" +_EMPTY_BODY = "Request body is empty." + +staged_upload: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "staged_upload", + default=None, +) +_heartbeat: Callable[[], None] | None = None + + +def register_upload_heartbeat(callback: Callable[[], None] | None) -> None: + """Keep the Gunicorn worker alive while a chunked upload is in progress.""" + global _heartbeat + _heartbeat = callback + + +def sync_max_bytes() -> int: + return int(os.environ.get("TAGBASE_SYNC_MAX_BYTES", SYNC_MAX_BYTES)) + + +def job_max_bytes() -> int: + return int(os.environ.get("TAGBASE_JOB_MAX_BYTES", JOB_MAX_BYTES)) + + +def upload_route(path: str) -> str | None: + trimmed = path.rstrip("/") or "/" + if trimmed.endswith("/ingest/jobs"): + return "jobs" + if trimmed.endswith("/ingest"): + return "sync" + return None + + +class UploadGuardMiddleware: + """ASGI middleware installed outside Connexion.""" + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope.get("type") != "http" or scope.get("method") != "POST": + await self.app(scope, receive, send) + return + route = upload_route(scope.get("path", "")) + if route == "jobs": + await self._stream_job(scope, receive, send) + return + if route == "sync": + await self._guard_sync(scope, receive, send) + return + await self.app(scope, receive, send) + + async def _guard_sync(self, scope, receive, send): + length = _content_length(scope) + limit = sync_max_bytes() + if length == 0: + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) + return + if length is not None and length > limit: + await _send_problem( + send, + 413, + _TITLE_TOO_LARGE, + "Synchronous ingest accepts at most 100 MiB. Use POST /ingest/jobs.", + ) + return + if length is not None: + await self.app(scope, receive, send) + return + try: + body = await _read_bounded(receive, limit) + except _TooLarge: + await _send_problem( + send, + 413, + _TITLE_TOO_LARGE, + "Synchronous ingest accepts at most 100 MiB. Use POST /ingest/jobs.", + ) + return + except _ClientGone: + return + if not body: + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) + return + await self.app( + _with_content_length(scope, len(body)), + _replay(body), + send, + ) + + async def _stream_job(self, scope, receive, send): + length = _content_length(scope) + limit = job_max_bytes() + if length == 0: + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) + return + if length is not None and length > limit: + await _send_problem( + send, + 413, + _TITLE_TOO_LARGE, + "Ingest jobs accept at most 2 GiB.", + ) + return + handle = _create_staged_file() + path = handle.name + try: + total = await _write_bounded(receive, handle, limit) + except _TooLarge: + handle.close() + _remove(path) + await _send_problem( + send, + 413, + _TITLE_TOO_LARGE, + "Ingest jobs accept at most 2 GiB.", + ) + return + except _ClientGone: + handle.close() + _remove(path) + return + except Exception: + handle.close() + _remove(path) + raise + handle.close() + if total == 0: + _remove(path) + await _send_problem(send, 400, _TITLE_BAD_REQUEST, _EMPTY_BODY) + return + token = staged_upload.set(path) + try: + await self.app( + _with_content_length(scope, len(_STAGED_BODY)), + _replay(_STAGED_BODY), + send, + ) + finally: + staged_upload.reset(token) + + +def _create_staged_file(): + return tempfile.NamedTemporaryFile( + dir=TEMP_DIR, + prefix="tagbase-upload-", + delete=False, + ) + + +def _content_length(scope) -> int | None: + for key, value in scope.get("headers", []): + if key.lower() == b"content-length": + try: + return int(value) + except ValueError: + return None + return None + + +def _with_content_length(scope, size: int): + headers = [ + (key, value) + for key, value in scope.get("headers", []) + if key.lower() != b"content-length" + ] + headers.append((b"content-length", str(size).encode("ascii"))) + updated = dict(scope) + updated["headers"] = headers + return updated + + +def _replay(body: bytes): + sent = False + + async def receive(): + nonlocal sent + if sent: + return {"type": _HTTP_REQUEST, "body": b"", "more_body": False} + sent = True + return {"type": _HTTP_REQUEST, "body": body, "more_body": False} + + return receive + + +async def _read_bounded(receive, limit: int) -> bytes: + chunks = [] + total = 0 + while True: + message = await receive() + if message["type"] == "http.disconnect": + raise _ClientGone() + if message["type"] != _HTTP_REQUEST: + continue + piece = message.get("body") or b"" + if piece: + total += len(piece) + if total > limit: + raise _TooLarge() + chunks.append(piece) + _beat() + if not message.get("more_body", False): + return b"".join(chunks) + + +async def _write_bounded(receive, handle, limit: int) -> int: + total = 0 + while True: + message = await receive() + if message["type"] == "http.disconnect": + raise _ClientGone() + if message["type"] != _HTTP_REQUEST: + continue + piece = message.get("body") or b"" + if piece: + total += len(piece) + if total > limit: + raise _TooLarge() + handle.write(piece) + _beat() + if not message.get("more_body", False): + return total + + +def _beat() -> None: + if _heartbeat is not None: + _heartbeat() + + +def _remove(path: str) -> None: + try: + os.remove(path) + except OSError: + pass + + +async def _send_problem(send, status: int, title: str, detail: str) -> None: + payload = json.dumps( + { + "type": TYPE_HTTP, + "title": title, + "status": status, + "detail": detail, + "trace_id": current_trace_id(), + } + ).encode("utf-8") + await send( + { + "type": "http.response.start", + "status": status, + "headers": [ + (b"content-type", b"application/problem+json"), + (b"content-length", str(len(payload)).encode("ascii")), + ], + } + ) + await send({"type": "http.response.body", "body": payload}) + + +class _TooLarge(Exception): + pass + + +class _ClientGone(Exception): + pass diff --git a/tagbase_server/tagbase_server/utils/io_utils.py b/tagbase_server/tagbase_server/utils/io_utils.py index fc752fc6..d62ce2d4 100644 --- a/tagbase_server/tagbase_server/utils/io_utils.py +++ b/tagbase_server/tagbase_server/utils/io_utils.py @@ -3,11 +3,8 @@ import os import re import tempfile -from time import time from urllib.request import urlopen -from pyunpack import Archive - logger = logging.getLogger(__name__) TEMP_DIR = tempfile.gettempdir() @@ -105,28 +102,3 @@ def process_post_input_data(filename, body): f.write(data) logger.debug(filepath) return filepath - - -def unpack_compressed_binary(file): - """ - Unpack a variety of compressed input files storing all .txt contents under the process temp directory. - No explicit cleanup is performed per-se however details of the implementation - can be found in https://docs.python.org/3/library/tempfile.html#tempfile.TemporaryFile - For a list of files which can be processed, see - https://pypi.org/project/patool/, and - https://github.com/ponty/pyunpack - - :param file: The .zip or .tar.gz file containing one or more eTUFF .txt files. - :type file: str - """ - etuff_files = [] - temp_dir = os.path.join(TEMP_DIR, str(time())) - if not os.path.exists(temp_dir): - os.mkdir(temp_dir) - logger.info("Unpacking %s archive to %s", file, temp_dir) - Archive(file).extractall(temp_dir) - for root, dirs, files in os.walk(temp_dir): - for file in files: - if file.endswith(".txt"): - etuff_files.append(os.path.join(root, file)) - return etuff_files diff --git a/tagbase_server/tagbase_server/utils/processing_utils.py b/tagbase_server/tagbase_server/utils/processing_utils.py index daf0ea73..7dbfe968 100644 --- a/tagbase_server/tagbase_server/utils/processing_utils.py +++ b/tagbase_server/tagbase_server/utils/processing_utils.py @@ -1,4 +1,5 @@ import logging +import os import time from datetime import datetime as dt from datetime import timezone @@ -529,9 +530,23 @@ def _compute_submission_hashes(submission_filename, file_content, metadata_conte ) -def process_etuff_file(file, version=None, notes=None): +def process_etuff_file( + file, version=None, notes=None, count_request=True, submission_name=None +): start = time.perf_counter() - submission_filename = file + source_path = file + submission_filename = submission_name or os.path.basename(source_path) + + def _record(outcome, duration): + if count_request: + record_ingest_request(outcome, duration) + + def _soft_error(outcome, duration, message): + _record(outcome, duration) + if not count_request: + raise RuntimeError(message) + return 1 + tracer = get_tracer() logger.info( "Processing etuff file: %s", @@ -541,8 +556,11 @@ def process_etuff_file(file, version=None, notes=None): conn = connect() if not hasattr(conn, "cursor"): record_db_error("connect") - record_ingest_request("error", round(time.perf_counter() - start, 2)) - return 1 + return _soft_error( + "error", + round(time.perf_counter() - start, 2), + "database unavailable", + ) conn.autocommit = True with tracer.start_as_current_span("ingest.parse_etuff"): @@ -556,17 +574,15 @@ def process_etuff_file(file, version=None, notes=None): file_content, metadata_content, number_global_attributes_lines, - ) = get_dataset_properties(submission_filename) + ) = get_dataset_properties(source_path) ( content_hash, metadata_hash, entire_file_hash, - ) = _compute_submission_hashes( - submission_filename, file_content, metadata_content - ) + ) = _compute_submission_hashes(source_path, file_content, metadata_content) except Exception: logger.exception("Error parsing etuff file: %s", submission_filename) - record_ingest_request("error", round(time.perf_counter() - start, 2)) + _record("error", round(time.perf_counter() - start, 2)) raise with conn: @@ -579,9 +595,7 @@ def process_etuff_file(file, version=None, notes=None): submission_filename, entire_file_hash, ) - record_ingest_request( - "duplicate", round(time.perf_counter() - start, 2) - ) + _record("duplicate", round(time.perf_counter() - start, 2)) return 1 dataset_id = get_dataset_id( @@ -628,9 +642,7 @@ def process_etuff_file(file, version=None, notes=None): dataset_id, metadata_hash, ) - record_ingest_request( - "metadata_only", round(time.perf_counter() - start, 2) - ) + _record("metadata_only", round(time.perf_counter() - start, 2)) return 1 proc_obs, len_proc_obs = _build_proc_observations( @@ -648,10 +660,11 @@ def process_etuff_file(file, version=None, notes=None): ) if sub_elapsed is False: record_db_error("migrate_proc_observations") - record_ingest_request( - "error", round(time.perf_counter() - start, 2) + return _soft_error( + "error", + round(time.perf_counter() - start, 2), + "failed to migrate proc observations", ) - return 1 record_rows_written(len_proc_obs) logger.info( "Successful migration of %s 'proc_observations'. Elapsed time: %s second(s).", @@ -660,9 +673,7 @@ def process_etuff_file(file, version=None, notes=None): ) except Exception: record_db_error("persist") - record_ingest_request( - "error", round(time.perf_counter() - start, 2) - ) + _record("error", round(time.perf_counter() - start, 2)) raise conn.commit() @@ -672,7 +683,7 @@ def process_etuff_file(file, version=None, notes=None): finish = time.perf_counter() elapsed = round(finish - start, 2) - record_ingest_request("success", elapsed) + _record("success", elapsed) logger.info( "Data file %s successfully ingested into Tagbase DB. Total time: %s second(s)", submission_filename, diff --git a/wiki/Ingestion-and-Access-Patterns.md b/wiki/Ingestion-and-Access-Patterns.md index 428ec873..5b0ac74e 100644 --- a/wiki/Ingestion-and-Access-Patterns.md +++ b/wiki/Ingestion-and-Access-Patterns.md @@ -24,13 +24,15 @@ The following emerging best practices can be used to drive throughput in the ing ## Use compressed binary containers when submitting etuff data to tagbase-server -Although tagbase-server is capable of ingesting plain text (.txt) utf-8 encoded etuff data via both POST and GET requests, we suggest first grouping and compressing multiple etuff files into a single `.zip` for example. tagbase-server will decompress and unpack the binary container and then ingest files in parallel. +Synchronous `POST /ingest` and `GET /ingest` still process the file inside the request and return HTTP 200 with a trace id. A synchronous POST body is limited to 100 MiB. `GET /ingest` is also limited by the 60-second request window, so a large remote file should be submitted with `POST /ingest/jobs` instead. -tagbase-server uses the powerful [patool](https://pypi.org/project/patool/) library to unpack a wide variety of files. See [supported formats](https://wummel.github.io/patool/#main_content) for more information. +`POST /ingest/jobs` accepts `.txt`, `.zip`, `.tar.gz`, and `.tgz` up to 2 GiB. The body is streamed to disk and the call returns HTTP 202 with a job id. Poll `GET /ingest/jobs/{job_id}` until `status` is `succeeded` or `failed`. The upload form and the fswatch service both use this job API. + +Archives are opened with the Python `zipfile` and `tarfile` libraries. Other archive types are rejected. Extraction refuses path traversal, symlink members, and encrypted ZIP members, and it stops when a configurable member count (default 10,000) or uncompressed size (default 10 GiB) is exceeded. ## Submit multiple files at once -As mentioned above, tagbase-server is capable of ingesting multiple files in parallel. It does this by using the powerful and lightweight [parmap](https://pypi.org/project/parmap/) library which utilizes as many processor cores as possible to perform parallel ingestion. +A synchronous request still walks the eTUFF members with [parmap](https://pypi.org/project/parmap/). The job worker runs one upload at a time. Parallelism beyond that single process is tracked as a follow-on issue. ## Use rsync