diff --git a/serverless-fleets/interactive-fleets/README.md b/serverless-fleets/interactive-fleets/README.md new file mode 100644 index 00000000..13af807e --- /dev/null +++ b/serverless-fleets/interactive-fleets/README.md @@ -0,0 +1,334 @@ +# interactive-fleets + +A general-purpose **submit-and-poll** pattern for IBM Code Engine Fleets. +A Code Engine **Function** acts as a synchronous HTTP endpoint. It accepts any JSON data, +dispatches it as a fleet task, polls COS for the result, and returns the enriched output — +or a retrieval URL if the task is still running when the poll window closes. + +This sample ships a simple **data enrichment** worker (adds timestamp, hostname, +word/character counts, and reversed key list) as the fleet task. Swap out the +enrichment logic in [`task-container/entrypoint.sh`](task-container/entrypoint.sh) +with any workload: ML inference, document processing, data transformation, etc. + +--- + +## Architecture + +```mermaid +flowchart TD + CallerPost["Caller — POST /\nJSON body with arbitrary data"] + CallerGet["Caller — GET /\n?key=value&... query params"] + CallerResult["Caller — GET /result/{task_id}\nRetrieve previously submitted result"] + + Fn["Code Engine Function · function/__main__.py\n─────────────────────────────────────\nGenerates task_id (UUID)\nSerialises input dict as JSON string\nObtains IAM token via CE_API_KEY → submits task\n add_tasks: {idx, args: ['']}\nPolls COS for {COS_PREFIX}/{task_id}.json\n via IAM Bearer token up to POLL_TIMEOUT s\nHTTP 200 — result returned synchronously\nHTTP 202 — timeout; returns task_id + retrieve_url\nGET /result/{task_id} → direct COS lookup (IAM)"] + + subgraph Fleet["Code Engine Fleet · cx3-4x8 (4 vCPU · 8 GiB)"] + Task["Task Container · task-container/entrypoint.sh\n─────────────────────────────────────\n1 vCPU / 2 GiB\nReceives input as JSON string in $1\nEnriches payload:\n + timestamp, processing_node (hostname)\n + word_count, char_count\n + reversed_keys, summary\nWrites result to /data/results/interactive-fleet-result/{task_id}.json"] + end + + COS["COS Persistent Data Store · fleet-output-store\ninteractive-fleet-result/{task_id}.json\n{task_id, status, elapsed_seconds,\n enriched_at, processing_node,\n input: {...}, output: {...}}"] + + CallerPost -->|"HTTP POST — submit + poll"| Fn + CallerGet -->|"HTTP GET — submit + poll"| Fn + CallerResult-->|"HTTP GET — retrieve only"| Fn + Fn -->|"CE v2 REST API POST .../add_tasks"| Fleet + Fn -.->|"S3 REST API GET /{bucket}/interactive-fleet-result/{task_id}.json\n(IAM Bearer — polls up to POLL_TIMEOUT seconds)"| COS + Task -->|"COS mount /data/results/interactive-fleet-result/{task_id}.json"| COS +``` + +--- + +## Files + +| File | Purpose | +|------|---------| +| [`deploy`](deploy) | Single deploy script — builds image, creates fleet, creates secret, deploys function | +| [`task-container/Dockerfile`](task-container/Dockerfile) | Fleet task image — ubi-minimal + bash + jq | +| [`task-container/entrypoint.sh`](task-container/entrypoint.sh) | Enrichment worker — receives argv key-value pairs, writes `/data/results/interactive-fleet-result/{task_id}.json` | +| [`function/__main__.py`](function/__main__.py) | CE Function — routes requests, submits tasks, polls COS, returns results | + +--- + +## Prerequisites + +- IBM Cloud CLI with Code Engine plugin + ```bash + ibmcloud plugin install code-engine + ``` +- A Code Engine project selected + ```bash + ibmcloud ce project select --name + ``` +- Persistent data stores provisioned (e.g. via `init-fleet-sandbox`): + ```bash + ibmcloud ce pds create --name fleet-task-store ... + ibmcloud ce pds create --name fleet-output-store ... + ``` +- Subnetpool `fleet-subnetpool` provisioned +- Container registry secret `ce-auto-icr-private-eu-de` (or set `FLEET_REGISTRY_SECRET`) +- Your IBM Container Registry namespace: + ```bash + export ICR_NAMESPACE= + ``` +- An IBM Cloud API key with **Code Engine Writer** and **COS Reader** access: + ```bash + export CE_API_KEY= + ``` + +--- + +## Deployment + +```bash +export ICR_NAMESPACE= +export CE_API_KEY= + +./deploy +``` + +The script prints the function endpoint at the end. Copy it for the examples below. + +```bash +export FN_URL=https://...codeengine.appdomain.cloud +``` + +### Deploy options + +| Variable | Default | Description | +|---|---|---| +| `SKIP_BUILD=1` | — | Skip image build (reuse existing image) | +| `SKIP_FLEET=1` | — | Skip fleet creation; `FLEET_ID` must be set | +| `SKIP_SECRET=1` | — | Skip secret creation/update | +| `FLEET_MAX_SCALE` | `2` | Maximum fleet worker nodes | +| `POLL_TIMEOUT` | `30` | Seconds the function polls COS before returning 202 | + +--- + +## Usage + +### 1. Submit a task and wait for the result (synchronous) + +The function submits the task to the fleet, polls COS for up to `POLL_TIMEOUT` seconds, +and returns the enriched result directly when it is ready. + +**POST with a JSON body:** + +```bash +curl -s -X POST "${FN_URL}" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "Alice", + "city": "Berlin", + "message": "Hello from interactive-fleets" + }' | jq . +``` + +**GET with query parameters:** + +```bash +curl -s "${FN_URL}?name=Alice&city=Berlin&message=Hello+from+interactive-fleets" | jq . +``` + +**HTTP 200 — result available within poll window:** + +```json +{ + "task_id": "a3f5b8c2-4d1e-4b5a-9c3f-1e2d3c4b5a6f", + "status": "completed", + "result": { + "task_id": "a3f5b8c2-4d1e-4b5a-9c3f-1e2d3c4b5a6f", + "status": "completed", + "elapsed_seconds": 0.0021, + "enriched_at": "2025-01-15T12:00:00.123456Z", + "processing_node": "worker-abc123", + "input": { + "name": "Alice", + "city": "Berlin", + "message": "Hello from interactive-fleets" + }, + "output": { + "word_count": 5, + "char_count": 29, + "reversed_keys": ["message", "city", "name"], + "summary": "Enriched 3 field(s) from input." + } + } +} +``` + +--- + +### 2. Submit a task asynchronously (fire and forget) + +Set `poll_timeout=0` to skip polling entirely and get the retrieval URL immediately. + +```bash +curl -s -X POST "${FN_URL}" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "Bob", + "city": "London", + "poll_timeout": 0 + }' | jq . +``` + +**HTTP 202 — task accepted, result not yet available:** + +```json +{ + "message": "Task submitted and is being processed", + "task_id": "b4e6c9d2-5e2f-4c6b-8d4e-2f3e4d5c6b7e", + "status": "pending", + "fleet_id": "1a2b3c4d-...", + "poll_timeout_seconds": 0, + "retrieve_url": "https:///result/b4e6c9d2-5e2f-4c6b-8d4e-2f3e4d5c6b7e" +} +``` + +--- + +### 3. Retrieve a result by task ID + +Use the `retrieve_url` from the 202 response, or construct it manually: + +```bash +curl -s "${FN_URL}/result/b4e6c9d2-5e2f-4c6b-8d4e-2f3e4d5c6b7e" | jq . +``` + +**HTTP 200** — result is ready: + +```json +{ + "task_id": "b4e6c9d2-...", + "status": "completed", + "elapsed_seconds": 0.0018, + "enriched_at": "2025-01-15T12:01:15.000000Z", + "processing_node": "worker-def456", + "input": { "name": "Bob", "city": "London" }, + "output": { + "word_count": 2, + "char_count": 10, + "reversed_keys": ["city", "name"], + "summary": "Enriched 2 field(s) from input." + } +} +``` + +**HTTP 404** — task is still running or ID is invalid: + +```json +{ + "task_id": "b4e6c9d2-...", + "status": "pending_or_not_found", + "retrieve_url": "https:///result/b4e6c9d2-..." +} +``` + +--- + +### 4. Provide a custom task ID + +```bash +curl -s -X POST "${FN_URL}" \ + -H "Content-Type: application/json" \ + -d '{ + "task_id": "my-custom-task-001", + "product": "widget", + "quantity": "42" + }' | jq . +``` + +The result will be stored at `interactive-fleet-result/my-custom-task-001.json` in the COS bucket. + +--- + +## Result JSON schema + +The fleet task writes the following structure to `interactive-fleet-result/{task_id}.json`: + +```json +{ + "task_id": "", + "status": "completed", + "elapsed_seconds": 0.0021, + "enriched_at": "ISO-8601 UTC timestamp", + "processing_node": "hostname of the fleet worker", + "input": { + "...": "all key/value pairs from the original request (control keys stripped)" + }, + "output": { + "word_count": 42, + "char_count": 210, + "reversed_keys": ["key3", "key2", "key1"], + "summary": "Enriched N field(s) from input." + } +} +``` + +--- + +## Configuration reference + +### Function environment variables + +| Variable | Default | Description | +|---|---|---| +| `FLEET_ID` | **required** | Target fleet UUID or name | +| `CE_API_KEY` | **required** (from `interactive-fleet-secrets`) | IBM Cloud API key — IAM token used for CE Fleet API **and** COS access | +| `COS_BUCKET` | `fleet-output-store` | COS bucket where results are stored | +| `COS_PREFIX` | `interactive-fleet-result` | Folder prefix inside the COS bucket | +| `COS_ENDPOINT` | `https://s3.direct.eu-de.cloud-object-storage.appdomain.cloud` | COS S3 endpoint | +| `POLL_TIMEOUT` | `30` | Max seconds to poll COS before returning HTTP 202 | +| `POLL_INTERVAL` | `2` | Seconds between COS poll attempts | + +### Task container environment variables + +| Variable | Default | Description | +|---|---|---| +| `RESULTS_DIR` | `/data/results` | Fleet COS mount root; script writes `interactive-fleet-result/.json` inside it | + +### Deploy script environment variables + +| Variable | Default | Description | +|---|---|---| +| `ICR_NAMESPACE` | **required** | Your IBM Container Registry namespace | +| `CE_API_KEY` | **required** | IBM Cloud API key (Code Engine Writer + COS Reader — fleet API and COS) | +| `NAME_PREFIX` | `interactive-fleets` | Prefix for all created resource names | +| `FLEET_IMAGE` | `private.de.icr.io//interactive-fleets-task:latest` | Task image | +| `FLEET_REGISTRY_SECRET` | `ce-auto-icr-private-eu-de` | Registry pull secret | +| `FLEET_MAX_SCALE` | `2` | Maximum worker nodes | +| `FLEET_SCALE_DOWN_DELAY` | `300` | Idle seconds before scaling down (5 min) | +| `FLEET_CPU` | `1` | vCPUs per task container | +| `FLEET_MEMORY` | `2G` | Memory per task container | +| `FN_NAME` | `interactive-fleet-fn` | CE Function name | +| `SECRET_NAME` | `interactive-fleet-secrets` | Secret holding `CE_API_KEY` | +| `COS_PREFIX` | `interactive-fleet-result` | Folder prefix in COS bucket for results | +| `POLL_TIMEOUT` | `30` | Max poll seconds (function env var) | + +--- + +## Customising the workload + +The enrichment logic lives entirely in [`task-container/entrypoint.sh`](task-container/entrypoint.sh) +in the enrichment section (steps 3–6). Replace it with any processing you need: + +- ML model inference +- External API calls +- File or document transformation +- Database writes + +The contract the function and worker share: +1. The function submits one task per request with: + ```json + { "idx": "", "args": [""] } + ``` +2. The worker receives the JSON string as `$1` (or `sys.argv[1]` for Python) and reads it directly — no key-value parsing needed. +3. The worker writes the result JSON to `{RESULTS_DIR}/interactive-fleet-result/{task_id}.json`. + +After changing the worker, rebuild and redeploy with: + +```bash +./deploy +# or, to skip fleet re-creation: +SKIP_FLEET=1 FLEET_ID= ./deploy +``` diff --git a/serverless-fleets/interactive-fleets/deploy b/serverless-fleets/interactive-fleets/deploy new file mode 100755 index 00000000..43d515e4 --- /dev/null +++ b/serverless-fleets/interactive-fleets/deploy @@ -0,0 +1,311 @@ +#!/bin/bash +# --------------------------------------------------------------------------- +# deploy — interactive-fleets full setup +# +# Steps: +# 1. Build the fleet task container image via CE buildrun +# 2. Create the CE Fleet (CPU profile — no prehook needed) +# 3. Create/update the API key secret +# 4. Deploy the CE Function +# +# Prerequisites: +# - ibmcloud CLI logged in with Code Engine plugin installed +# - A Code Engine project selected (ibmcloud ce project select --name ) +# - Persistent data stores provisioned: +# ibmcloud ce pds create --name fleet-task-store ... +# ibmcloud ce pds create --name fleet-output-store ... +# - Subnetpool 'fleet-subnetpool' provisioned +# - Registry secret 'ce-auto-icr-private-eu-de' (or set FLEET_REGISTRY_SECRET) +# - ICR_NAMESPACE exported (your IBM Container Registry namespace) +# - CE_API_KEY exported (IBM Cloud API key with CE Writer + COS access) +# +# Usage: +# export ICR_NAMESPACE= +# export CE_API_KEY= +# ./deploy +# +# # Override fleet ID for function deploy only: +# export FLEET_ID= +# SKIP_BUILD=1 SKIP_FLEET=1 ./deploy +# --------------------------------------------------------------------------- +set -euo pipefail + +SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd ) + +# ── Helpers ────────────────────────────────────────────────────────────────── +print_step() { echo -e "\n\033[94m═══ $1 ═══\033[0m"; } +print_msg() { echo -e "\033[94m$1\033[0m"; } +print_success() { echo -e "\033[32m$1\033[0m"; } +print_error() { echo -e "\033[31m$1\033[0m"; } + +# ── Configuration ───────────────────────────────────────────────────────────── + +ICR_REGISTRY="${ICR_REGISTRY:=private.de.icr.io}" + +if [[ -z "${ICR_NAMESPACE:-}" ]]; then + print_error "ICR_NAMESPACE is required. Set it to your IBM Container Registry namespace." + echo " export ICR_NAMESPACE=" + exit 1 +fi + +if [[ -z "${CE_API_KEY:-}" ]]; then + print_error "CE_API_KEY is required. Set it to your IBM Cloud API key." + echo " export CE_API_KEY=" + exit 1 +fi + +# Image +NAME_PREFIX="${NAME_PREFIX:=interactive-fleets}" +FLEET_IMAGE="${FLEET_IMAGE:=${ICR_REGISTRY}/${ICR_NAMESPACE}/interactive-fleets-task:latest}" +FLEET_REGISTRY_SECRET="${FLEET_REGISTRY_SECRET:=ce-auto-icr-private-eu-de}" +TASK_CONTAINER_DIR="${SCRIPT_DIR}/task-container" +FUNCTION_DIR="${SCRIPT_DIR}/function" + +# Fleet +FLEET_NAME="${FLEET_NAME:=}" # auto-generated if empty +FLEET_TASKS_STATE_STORE="${FLEET_TASKS_STATE_STORE:=fleet-task-store}" +FLEET_OUTPUT_STORE="${FLEET_OUTPUT_STORE:=fleet-output-store}" +FLEET_OUTPUT_COS_PATH="${FLEET_OUTPUT_COS_PATH:=}" # mount entire PDS root; task writes interactive-fleet-result/ subdir +FLEET_SUBNETPOOL="${FLEET_SUBNETPOOL:=fleet-subnetpool}" +FLEET_MAX_SCALE="${FLEET_MAX_SCALE:=2}" +FLEET_MIN_SCALE="${FLEET_MIN_SCALE:=0}" +FLEET_SCALE_DOWN_DELAY="${FLEET_SCALE_DOWN_DELAY:=300}" +FLEET_RETRYLIMIT="${FLEET_RETRYLIMIT:=2}" +FLEET_CPU="${FLEET_CPU:=1}" +FLEET_MEMORY="${FLEET_MEMORY:=2G}" + +# Function +FN_NAME="${FN_NAME:=interactive-fleet-fn}" +SECRET_NAME="${SECRET_NAME:=interactive-fleet-secrets}" +COS_PREFIX="${COS_PREFIX:=interactive-fleet-result}" +PDS_NAME="${PDS_NAME:=fleet-output-store}" +COS_BUCKET="${COS_BUCKET:=}" +COS_ENDPOINT="${COS_ENDPOINT:=}" +POLL_TIMEOUT="${POLL_TIMEOUT:=90}" +MAX_EXECUTION_TIME="${MAX_EXECUTION_TIME:=120}" +# COS access uses the IAM token derived from CE_API_KEY — no separate COS secret needed + +# Skip flags +SKIP_BUILD="${SKIP_BUILD:=0}" +SKIP_FLEET="${SKIP_FLEET:=0}" +SKIP_SECRET="${SKIP_SECRET:=0}" + +# ══════════════════════════════════════════════════════════════════════════════ +# STEP 1 — Build the task container image +# ══════════════════════════════════════════════════════════════════════════════ +if [[ "${SKIP_BUILD}" == "1" ]]; then + print_msg "\n[skip] SKIP_BUILD=1 — skipping task image build" +else + print_step "1 / 4 Building task container image" + BUILD_RUN_NAME="${NAME_PREFIX}-buildrun-$(uuidgen | tr '[:upper:]' '[:lower:]' | awk -F- '{print $1}')" + + echo "" + echo "ibmcloud ce buildrun submit" + echo " --name ${BUILD_RUN_NAME}" + echo " --source ${TASK_CONTAINER_DIR}" + echo " --image ${FLEET_IMAGE}" + echo " --registry-secret ${FLEET_REGISTRY_SECRET}" + echo " --strategy dockerfile" + echo " --wait" + echo "" + + ibmcloud ce buildrun submit \ + --name "${BUILD_RUN_NAME}" \ + --source "${TASK_CONTAINER_DIR}" \ + --image "${FLEET_IMAGE}" \ + --registry-secret "${FLEET_REGISTRY_SECRET}" \ + --strategy dockerfile \ + --wait + + print_success "✔ Task image built: ${FLEET_IMAGE}" +fi + +# ══════════════════════════════════════════════════════════════════════════════ +# STEP 2 — Create the fleet +# ══════════════════════════════════════════════════════════════════════════════ +if [[ "${SKIP_FLEET}" == "1" ]]; then + print_msg "\n[skip] SKIP_FLEET=1 — skipping fleet creation" + if [[ -z "${FLEET_ID:-}" ]]; then + print_error "FLEET_ID must be set when SKIP_FLEET=1" + exit 1 + fi +else + print_step "2 / 4 Creating CE Fleet" + + GEN_FLEET_NAME="${NAME_PREFIX}-$(uuidgen | tr '[:upper:]' '[:lower:]' | awk -F- '{print $1}')" + FLEET_NAME="${FLEET_NAME:-${GEN_FLEET_NAME}}" + + SCALE_FLAGS=(--scale-max "${FLEET_MAX_SCALE}") + [[ "${FLEET_MIN_SCALE}" != "0" ]] && SCALE_FLAGS+=(--scale-min "${FLEET_MIN_SCALE}") + [[ -n "${FLEET_SCALE_DOWN_DELAY}" ]] && SCALE_FLAGS+=(--scale-down-delay "${FLEET_SCALE_DOWN_DELAY}") + + echo "" + echo "ibmcloud ce fleet create" + echo " --name ${FLEET_NAME}" + echo " --tasks-state-store ${FLEET_TASKS_STATE_STORE}" + echo " --subnetpool-name ${FLEET_SUBNETPOOL}" + echo " --image ${FLEET_IMAGE}" + echo " --registry-secret ${FLEET_REGISTRY_SECRET}" + echo " --scale-max ${FLEET_MAX_SCALE}" + echo " --scale-down-delay ${FLEET_SCALE_DOWN_DELAY}" + echo " --retrylimit ${FLEET_RETRYLIMIT}" + echo " --cpu ${FLEET_CPU}" + echo " --memory ${FLEET_MEMORY}" + # Build mount-data-store value: omit subpath when FLEET_OUTPUT_COS_PATH is empty + if [[ -n "${FLEET_OUTPUT_COS_PATH}" ]]; then + MOUNT_DS="/data/results=${FLEET_OUTPUT_STORE}:${FLEET_OUTPUT_COS_PATH}" + else + MOUNT_DS="/data/results=${FLEET_OUTPUT_STORE}" + fi + + echo " --mount-data-store ${MOUNT_DS}" + echo "" + + ibmcloud ce fleet create \ + --name "${FLEET_NAME}" \ + --tasks-state-store "${FLEET_TASKS_STATE_STORE}" \ + --subnetpool-name "${FLEET_SUBNETPOOL}" \ + --image "${FLEET_IMAGE}" \ + --registry-secret "${FLEET_REGISTRY_SECRET}" \ + "${SCALE_FLAGS[@]}" \ + --retrylimit "${FLEET_RETRYLIMIT}" \ + --cpu "${FLEET_CPU}" \ + --memory "${FLEET_MEMORY}" \ + --mount-data-store "${MOUNT_DS}" + + _FLEET_LIST=$(ibmcloud ce fleet list --output json 2>/dev/null || true) + if [[ -n "${_FLEET_LIST}" ]]; then + FLEET_ID=$(echo "${_FLEET_LIST}" | jq -r --arg name "${FLEET_NAME}" \ + '.fleets[] | select(.name == $name) | .id // empty' 2>/dev/null || true) + fi + + print_success "✔ Fleet created: ${FLEET_NAME}" + if [[ -n "${FLEET_ID}" ]]; then + print_msg " Fleet ID: ${FLEET_ID}" + fi +fi + +# ══════════════════════════════════════════════════════════════════════════════ +# STEP 3 — Create/update the API key secret +# ══════════════════════════════════════════════════════════════════════════════ +if [[ "${SKIP_SECRET}" == "1" ]]; then + print_msg "\n[skip] SKIP_SECRET=1 — skipping secret creation" +else + print_step "3 / 4 Creating API key secret '${SECRET_NAME}'" + + if ibmcloud ce secret get --name "${SECRET_NAME}" >/dev/null 2>&1; then + print_msg " Secret '${SECRET_NAME}' already exists — updating..." + ibmcloud ce secret update --name "${SECRET_NAME}" \ + --from-literal CE_API_KEY="${CE_API_KEY}" + else + ibmcloud ce secret create --name "${SECRET_NAME}" \ + --from-literal CE_API_KEY="${CE_API_KEY}" + fi + + print_success "✔ Secret '${SECRET_NAME}' ready" +fi + +# ══════════════════════════════════════════════════════════════════════════════ +# STEP 4 — Deploy the CE Function +# ══════════════════════════════════════════════════════════════════════════════ +print_step "4 / 4 Deploying CE Function '${FN_NAME}'" + +# Auto-discover the real COS bucket name and endpoint from the PDS +if [[ -z "${COS_BUCKET}" ]]; then + PDS_JSON=$(ibmcloud ce pds get --name "${PDS_NAME}" --output json 2>/dev/null || echo "") + if [[ -n "${PDS_JSON}" ]]; then + COS_BUCKET=$(echo "${PDS_JSON}" | jq -r '.data.bucket_name // .cos_bucket_name // .cos_bucket // empty') + PDS_REGION=$(echo "${PDS_JSON}" | jq -r '.data.bucket_location // .bucket_region // .region // empty') + if [[ -n "${PDS_REGION}" && -z "${COS_ENDPOINT}" ]]; then + COS_ENDPOINT="https://s3.direct.${PDS_REGION}.cloud-object-storage.appdomain.cloud" + fi + fi +fi + +COS_BUCKET="${COS_BUCKET:=fleet-output-store}" +COS_ENDPOINT="${COS_ENDPOINT:=https://s3.direct.eu-de.cloud-object-storage.appdomain.cloud}" + +if [[ -z "${FLEET_ID:-}" ]]; then + if [[ -n "${FLEET_NAME:-}" ]]; then + print_msg " FLEET_ID not set — looking up fleet '${FLEET_NAME}'..." + _FLEET_LIST=$(ibmcloud ce fleet list --output json 2>/dev/null || true) + if [[ -n "${_FLEET_LIST}" ]]; then + FLEET_ID=$(echo "${_FLEET_LIST}" | jq -r --arg name "${FLEET_NAME}" \ + '.fleets[] | select(.name == $name) | .id // empty' 2>/dev/null || true) + fi + else + print_msg " FLEET_ID and FLEET_NAME not set — searching by name prefix '${NAME_PREFIX}'..." + # Fleet list output: {"fleets": [...], "limit": 0} + # Filter by NAME_PREFIX, sort by created_at ascending, take the last (newest). + _FLEET_LIST=$(ibmcloud ce fleet list --output json 2>/dev/null || true) + if [[ -n "${_FLEET_LIST}" ]]; then + FLEET_ID=$(echo "${_FLEET_LIST}" \ + | jq -r "([.fleets[] | select(.name | test(\"${NAME_PREFIX}\"))] | sort_by(.created_at) | last | .id // empty) // empty" \ + 2>/dev/null || true) + fi + fi +fi + +if [[ -z "${FLEET_ID:-}" ]]; then + print_error "FLEET_ID is required. Set export FLEET_ID= or run without SKIP_FLEET=1." + exit 1 +fi + +echo "" +echo "ibmcloud ce fn create/update" +echo " --name ${FN_NAME}" +echo " --build-source ${FUNCTION_DIR}" +echo " --runtime python-3.13" +echo " --maxexecutiontime ${MAX_EXECUTION_TIME}" +echo " --env FLEET_ID=${FLEET_ID}" +echo " --env COS_BUCKET=${COS_BUCKET}" +echo " --env COS_PREFIX=${COS_PREFIX}" +echo " --env COS_ENDPOINT=${COS_ENDPOINT}" +echo " --env POLL_TIMEOUT=${POLL_TIMEOUT}" +echo " --env-from-secret ${SECRET_NAME} # CE_API_KEY (fleet API + COS)" +echo "" + +FN_ARGS=( + --name "${FN_NAME}" + --build-source "${FUNCTION_DIR}" + --runtime python-3.13 + --maxexecutiontime "${MAX_EXECUTION_TIME}" + --env FLEET_ID="${FLEET_ID}" + --env COS_BUCKET="${COS_BUCKET}" + --env COS_PREFIX="${COS_PREFIX}" + --env COS_ENDPOINT="${COS_ENDPOINT}" + --env POLL_TIMEOUT="${POLL_TIMEOUT}" + --env-from-secret "${SECRET_NAME}" +) + +if ibmcloud ce fn get --name "${FN_NAME}" >/dev/null 2>&1; then + print_msg " Function '${FN_NAME}' already exists — updating..." + ibmcloud ce fn update "${FN_ARGS[@]}" +else + ibmcloud ce fn create "${FN_ARGS[@]}" +fi + +FN_ENDPOINT=$(ibmcloud ce fn get --name "${FN_NAME}" --output json \ + | jq -r '.endpoint // empty' 2>/dev/null || echo "") + +# ══════════════════════════════════════════════════════════════════════════════ +# Done +# ══════════════════════════════════════════════════════════════════════════════ +print_success "\n══════════════════════════════════════════" +print_success " interactive-fleets deployed successfully" +print_success "══════════════════════════════════════════" +echo "" +print_msg "Fleet: ${FLEET_NAME:-${FLEET_ID}}" +[[ -n "${FLEET_ID}" ]] && print_msg "Fleet ID: ${FLEET_ID}" +print_msg "Function: ${FN_NAME}" +if [[ -n "${FN_ENDPOINT}" ]]; then + print_msg "Endpoint: ${FN_ENDPOINT}" + echo "" + print_msg "Submit a task (synchronous):" + echo " curl -s -X POST ${FN_ENDPOINT} \\" + echo " -H 'Content-Type: application/json' \\" + echo " -d '{\"name\": \"World\", \"greeting\": \"Hello\"}' | jq ." + echo "" + print_msg "Retrieve a result:" + echo " curl -s ${FN_ENDPOINT}/result/ | jq ." +fi diff --git a/serverless-fleets/interactive-fleets/function/__main__.py b/serverless-fleets/interactive-fleets/function/__main__.py new file mode 100644 index 00000000..4d1b3d5a --- /dev/null +++ b/serverless-fleets/interactive-fleets/function/__main__.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +""" +Code Engine Function — interactive-fleets dispatcher + +A general-purpose submit-and-poll gateway for IBM Code Engine Fleets. +Accepts arbitrary JSON input, submits it as a fleet task, polls COS for the +result, and returns either the completed result (synchronous) or a retrieval +URL (asynchronous). + +────────────────────────────────────────────────────────────────────────────── +Endpoints +────────────────────────────────────────────────────────────────────────────── + + POST / Submit a new task with a JSON body. + GET / Submit a new task with URL query parameters. + GET /result/{task_id} Retrieve the result of a previously submitted task. + +────────────────────────────────────────────────────────────────────────────── +Submit (POST / or GET /) +────────────────────────────────────────────────────────────────────────────── + + Any key/value pairs in the request body or query string are forwarded to the + fleet task as its input. The function: + + 1. Generates a unique task_id (UUID) — or uses a caller-supplied one. + 2. Obtains an IAM token from CE_API_KEY. + 3. Serialises the input dict (plus task_id) as a single JSON string and + submits it to the CE Fleet via add_tasks with: + { "idx": "", "args": [""] } + The task container receives the JSON string as $1 and reads it directly. + 4. Polls COS for {COS_PREFIX}/{task_id}.json for up to POLL_TIMEOUT seconds. + 5a. If the result is available → HTTP 200 with the result JSON. + 5b. If polling times out → HTTP 202 with task_id and a retrieve_url. + + Reserved input keys (consumed by the function, not forwarded to the task): + task_id — custom task ID override + poll_timeout — override the server-side POLL_TIMEOUT for this request + +────────────────────────────────────────────────────────────────────────────── +Retrieve (GET /result/{task_id}) +────────────────────────────────────────────────────────────────────────────── + + Fetches {COS_PREFIX}/{task_id}.json from COS using the IAM token as Bearer. + HTTP 200 — result available; body contains the result JSON. + HTTP 404 — result not yet available (task still running or invalid ID). + +────────────────────────────────────────────────────────────────────────────── +Required environment variables +────────────────────────────────────────────────────────────────────────────── + + FLEET_ID — UUID or name of the target CE Fleet + CE_API_KEY — IBM Cloud API key (IAM token exchange — used for both CE Fleet API and COS) + COS_BUCKET — COS bucket name where task results are stored + +────────────────────────────────────────────────────────────────────────────── +Optional environment variables +────────────────────────────────────────────────────────────────────────────── + + CE_API_BASE_URL — CE API base URL (injected by CE runtime automatically) + CE_PROJECT_ID — CE project UUID (injected by CE runtime automatically) + COS_ENDPOINT — COS S3 endpoint (default: s3.direct.eu-de.cloud-object-storage.appdomain.cloud) + COS_PREFIX — Folder prefix inside the COS bucket (default: interactive-fleet-result) + POLL_TIMEOUT — Max seconds to poll COS before returning 202 (default: 30) + POLL_INTERVAL — Seconds between poll attempts (default: 2) +""" + +import http.client +import json +import os +import re +import time +import urllib.parse +import uuid + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +CE_API_BASE_URL = os.environ.get( + "CE_API_BASE_URL", + "https://api.private.eu-de.codeengine.cloud.ibm.com", +) +CE_PROJECT_ID = os.environ.get("CE_PROJECT_ID", "") +FLEET_ID = os.environ.get("FLEET_ID", "") +CE_API_KEY = os.environ.get("CE_API_KEY", "") + +# Self-URL — used to build retrieve_url in async (202) responses +_CE_FUNCTION = os.environ.get("CE_FUNCTION", "") +_CE_SUBDOMAIN = os.environ.get("CE_SUBDOMAIN", "") +_CE_DOMAIN = os.environ.get("CE_DOMAIN", "") +FUNCTION_URL = ( + f"https://{_CE_FUNCTION}.{_CE_SUBDOMAIN}.{_CE_DOMAIN}" + if _CE_FUNCTION and _CE_SUBDOMAIN and _CE_DOMAIN + else "" +) + +# COS — accessed via IAM Bearer token derived from CE_API_KEY +COS_ENDPOINT = os.environ.get( + "COS_ENDPOINT", + "https://s3.direct.eu-de.cloud-object-storage.appdomain.cloud", +) +COS_BUCKET = os.environ.get("COS_BUCKET", "fleet-output-store") +COS_PREFIX = os.environ.get("COS_PREFIX", "interactive-fleet-result").strip("/") + +POLL_TIMEOUT = float(os.environ.get("POLL_TIMEOUT", "30.0")) +POLL_INTERVAL = float(os.environ.get("POLL_INTERVAL", "2.0")) + +IAM_TOKEN_URL = "iam.cloud.ibm.com" + +# Keys that are consumed by the function and not forwarded to the fleet task +_RESERVED_KEYS = {"task_id", "poll_timeout", "__ce_query", "__ce_path", "__ce_method"} + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _get_iam_token(api_key: str) -> str: + """Exchange an IBM Cloud API key for an IAM bearer token.""" + body = urllib.parse.urlencode({ + "grant_type": "urn:ibm:params:oauth:grant-type:apikey", + "apikey": api_key, + }) + conn = http.client.HTTPSConnection(IAM_TOKEN_URL, timeout=15) + conn.request( + "POST", + "/identity/token", + body, + { + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, + ) + resp = conn.getresponse() + data = json.loads(resp.read()) + conn.close() + if resp.status != 200: + raise RuntimeError(f"IAM token request failed ({resp.status}): {data}") + return data["access_token"] + + +def _parse_url(url: str) -> tuple[str, int, bool]: + """Return (host, port, use_https) from a URL string.""" + https = url.startswith("https://") + url = url.removeprefix("https://").removeprefix("http://") + host, _, port_str = url.partition(":") + if "/" in host: + host = host.split("/")[0] + port = int(port_str) if port_str else (443 if https else 80) + return host, port, https + + +# --------------------------------------------------------------------------- +# CE Fleet API +# --------------------------------------------------------------------------- + +def _add_fleet_task( + api_base_url: str, + project_id: str, + fleet_id: str, + iam_token: str, + task_data: dict, + task_idx: str, +) -> tuple[int, dict]: + """ + POST /v2/projects/{project_id}/fleets/{fleet_id}/add_tasks?version=2025-07-10 + + Submits a single task to the fleet. + + The entire task_data dict is serialised as a single JSON string and passed + as args[0]. The task container reads it with: + + INPUT_JSON="$1" + + This avoids the fragile key-value argv convention and lets the container + receive a fully-typed JSON payload without any shell parsing. + + JSONL task specification (one task): + { "idx": "", "args": [""] } + + CE add_tasks body: + { + "tasks_specification": { + "json": [ + { "idx": "", "args": [""] } + ] + } + } + """ + path = f"/v2/projects/{project_id}/fleets/{fleet_id}/add_tasks?version=2025-07-10" + task_entry: dict = { + "idx": task_idx, + "args": [json.dumps(task_data)], # single arg: the full payload as JSON string + } + + payload = {"tasks_specification": {"json": [task_entry]}} + body = json.dumps(payload) + + host, port, https = _parse_url(api_base_url) + conn = ( + http.client.HTTPSConnection(host, port, timeout=20) + if https + else http.client.HTTPConnection(host, port, timeout=20) + ) + conn.request( + "POST", + path, + body, + { + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"Bearer {iam_token}", + }, + ) + resp = conn.getresponse() + data = resp.read() + conn.close() + return resp.status, json.loads(data) if data else {} + + +# --------------------------------------------------------------------------- +# COS access (IAM Bearer token) +# --------------------------------------------------------------------------- + +def _get_cos_object( + cos_endpoint: str, + bucket: str, + object_key: str, + iam_token: str, +) -> tuple[int, dict | None, str]: + """ + GET an object from COS via S3 REST API using an IAM Bearer token. + IBM COS accepts IAM tokens directly as Authorization: Bearer . + Returns (status_code, parsed_json_or_none, debug_string). + """ + host, port, https = _parse_url(cos_endpoint) + encoded_key = urllib.parse.quote(object_key.lstrip("/")) + path = f"/{bucket}/{encoded_key}" + target_url = f"{'https' if https else 'http'}://{host}:{port}{path}" + + try: + conn = ( + http.client.HTTPSConnection(host, port, timeout=10) + if https + else http.client.HTTPConnection(host, port, timeout=10) + ) + conn.request("GET", path, headers={"Authorization": f"Bearer {iam_token}"}) + resp = conn.getresponse() + data = resp.read() + conn.close() + + debug = f"COS GET {target_url} → HTTP {resp.status}" + + if resp.status == 200: + try: + return resp.status, json.loads(data), debug + except Exception: + return resp.status, {"raw": data.decode("utf-8", errors="replace")}, debug + + err = data.decode("utf-8", errors="replace")[:300] if data else "" + return resp.status, None, f"{debug}: {err}" + except Exception as exc: + return 500, None, f"COS GET {target_url} failed: {exc}" + + +def _poll_cos( + task_id: str, + iam_token: str, + max_duration: float = POLL_TIMEOUT, + interval: float = POLL_INTERVAL, +) -> dict | None: + """ + Poll COS for {COS_PREFIX}/{task_id}.json until found or timeout elapsed. + Returns the parsed JSON dict on success, None on timeout. + """ + object_key = f"{COS_PREFIX}/{task_id}.json" if COS_PREFIX else f"{task_id}.json" + deadline = time.time() + max_duration + + while time.time() < deadline: + status, result, _ = _get_cos_object(COS_ENDPOINT, COS_BUCKET, object_key, iam_token) + if status == 200 and result is not None: + return result + time.sleep(interval) + + # One final check after the loop + status, result, _ = _get_cos_object(COS_ENDPOINT, COS_BUCKET, object_key, iam_token) + return result if (status == 200 and result is not None) else None + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + +def _sanitize_id(val: str) -> str: + """Allow only alphanumeric, dashes, and underscores in a task ID.""" + return re.sub(r"[^a-zA-Z0-9_\-]", "", val) + + +def _json_response(status_code: int, body: dict) -> dict: + return { + "headers": {"Content-Type": "application/json"}, + "statusCode": status_code, + "body": body, + } + + +# --------------------------------------------------------------------------- +# Route handlers +# --------------------------------------------------------------------------- + +def _handle_retrieve(task_id: str, iam_token: str) -> dict: + """ + GET /result/{task_id} + Look up the result in COS via IAM Bearer token and return it, or 404 if not ready. + """ + object_key = f"{COS_PREFIX}/{task_id}.json" if COS_PREFIX else f"{task_id}.json" + status, result, debug = _get_cos_object(COS_ENDPOINT, COS_BUCKET, object_key, iam_token) + + if status == 200 and result is not None: + return _json_response(200, result) + + print(f"[fn] COS lookup for task {task_id}: {debug}") + body: dict = {"task_id": task_id, "status": "pending_or_not_found"} + if FUNCTION_URL: + body["retrieve_url"] = f"{FUNCTION_URL}/result/{task_id}" + return _json_response(404, body) + + +def _handle_submit(params: dict, iam_token: str) -> dict: + """ + POST / or GET / + Extract caller data from params, submit a fleet task, poll COS, and return. + """ + # ── 1. Resolve configuration ─────────────────────────────────────────── + fleet_id = str(params.get("fleet_id", FLEET_ID)).strip() or FLEET_ID + project_id = CE_PROJECT_ID + + missing_config = [k for k, v in { + "FLEET_ID": fleet_id, + "CE_PROJECT_ID": project_id, + "CE_API_KEY": CE_API_KEY, + }.items() if not v] + if missing_config: + return _json_response(500, {"error": f"Missing required configuration: {', '.join(missing_config)}"}) + + # ── 2. Resolve task_id and poll_timeout ──────────────────────────────── + raw_task_id = _sanitize_id(str(params.get("task_id", "")).strip()) + task_id = raw_task_id or str(uuid.uuid4()) + + try: + poll_timeout_sec = float(params.get("poll_timeout", POLL_TIMEOUT)) + except (TypeError, ValueError): + poll_timeout_sec = POLL_TIMEOUT + + # ── 3. Build the task payload (all non-reserved keys forwarded) ───────── + task_data = {k: v for k, v in params.items() if k not in _RESERVED_KEYS} + # Always include task_id so the worker knows its output filename + task_data["task_id"] = task_id + + # ── 4. Submit task to fleet ──────────────────────────────────────────── + # task_data is serialised as a single JSON string passed as args[0]. + # The container reads it with: INPUT_JSON="$1" + try: + status, response = _add_fleet_task( + CE_API_BASE_URL, project_id, fleet_id, iam_token, task_data, task_idx=task_id + ) + except Exception as exc: + return _json_response(502, {"error": f"Failed to call CE Fleet tasks API: {exc}"}) + + if status not in (200, 201, 202): + return _json_response(502, { + "error": "CE fleet tasks API returned an unexpected status", + "upstream_status": status, + "upstream_body": response, + }) + + # ── 5. Poll COS for result (IAM Bearer) ──────────────────────────────── + if poll_timeout_sec > 0: + result = _poll_cos( + task_id = task_id, + iam_token = iam_token, + max_duration = poll_timeout_sec, + interval = POLL_INTERVAL, + ) + if result is not None: + return _json_response(200, { + "task_id": task_id, + "status": "completed", + "result": result, + }) + + # ── 6. Polling timed out — return task_id for async retrieval ────────── + body_202: dict = { + "message": "Task submitted and is being processed", + "task_id": task_id, + "status": "pending", + "fleet_id": fleet_id, + "poll_timeout_seconds": poll_timeout_sec, + } + if FUNCTION_URL: + body_202["retrieve_url"] = f"{FUNCTION_URL}/result/{task_id}" + return _json_response(202, body_202) + + +# --------------------------------------------------------------------------- +# Function entry point +# --------------------------------------------------------------------------- + +def main(params: dict) -> dict: + """ + CE Function entry point. + + Routes: + GET /result/{task_id} → fetch result from COS (IAM Bearer) + POST / → submit task (IAM) + poll COS (IAM Bearer) + GET / → submit task (IAM) + poll COS (IAM Bearer) + + A single IAM token is obtained from CE_API_KEY at the start of each + invocation and reused for both the CE Fleet API and COS requests. + """ + if not CE_API_KEY: + return _json_response(500, {"error": "Missing required configuration: CE_API_KEY"}) + + # ── Obtain IAM token — used for both CE Fleet API and COS ───────────── + try: + iam_token = _get_iam_token(CE_API_KEY) + except Exception as exc: + return _json_response(502, {"error": f"Failed to obtain IAM token: {exc}"}) + + # ── Unpack query string ──────────────────────────────────────────────── + if "__ce_query" in params and isinstance(params["__ce_query"], str) and params["__ce_query"]: + try: + parsed_query = urllib.parse.parse_qs(params["__ce_query"]) + for qk, qv in parsed_query.items(): + if qk not in params and qv: + params[qk] = qv[0] + except Exception: + pass + + # ── Route: GET /result/{task_id} ─────────────────────────────────────── + ce_path = str(params.get("__ce_path", "")).strip("/") + if ce_path.startswith("result/"): + raw_task_id = ce_path[len("result/"):] + task_id = _sanitize_id(raw_task_id.strip()) + if not task_id: + return _json_response(400, {"error": "Invalid or missing task_id in path"}) + return _handle_retrieve(task_id, iam_token) + + # ── Route: POST / or GET / — submit a new task ───────────────────────── + return _handle_submit(params, iam_token) diff --git a/serverless-fleets/interactive-fleets/task-container/Dockerfile b/serverless-fleets/interactive-fleets/task-container/Dockerfile new file mode 100644 index 00000000..4e7f6d17 --- /dev/null +++ b/serverless-fleets/interactive-fleets/task-container/Dockerfile @@ -0,0 +1,29 @@ +# --------------------------------------------------------------------------- +# Fleet task image — interactive-fleets data enrichment worker +# +# Base: ubi-minimal +# Purpose: Receives a JSON payload as $1, enriches it, and writes the result +# to the COS mount at /data/results/interactive-fleet-result/{task_id}.json. +# --------------------------------------------------------------------------- +FROM registry.access.redhat.com/ubi10/ubi-minimal + +# Install runtime dependencies +# bash — shell interpreter +# jq — JSON processing +# util-linux — uuidgen +# (hostname read from /etc/hostname; elapsed time computed with bash arithmetic) +RUN microdnf install -y \ + bash \ + jq \ + util-linux \ + && microdnf clean all + +# Copy the task entrypoint +COPY entrypoint.sh /usr/local/bin/entrypoint.sh +RUN chmod +x /usr/local/bin/entrypoint.sh + +# Results are written to /data/results (the fleet COS mount root). +# The script appends interactive-fleet-result/.json inside this dir. +ENV RESULTS_DIR=/data/results + +ENTRYPOINT ["/usr/local/bin/entrypoint.sh"] diff --git a/serverless-fleets/interactive-fleets/task-container/entrypoint.sh b/serverless-fleets/interactive-fleets/task-container/entrypoint.sh new file mode 100644 index 00000000..9156c8bd --- /dev/null +++ b/serverless-fleets/interactive-fleets/task-container/entrypoint.sh @@ -0,0 +1,144 @@ +#!/bin/bash +# --------------------------------------------------------------------------- +# Fleet task entrypoint — data enrichment worker +# +# Receives input as a single JSON string in $1, serialised by the CE Function. +# The fleet add_tasks call looks like: +# +# { "idx": "", "args": ["{\"key\":\"value\",...}"] } +# +# The container receives this as: entrypoint.sh '{"key":"value",...}' +# so the entire payload is available as $1 with no parsing needed. +# +# Output JSON written to COS mount: +# ${RESULTS_DIR}/interactive-fleet-result/${task_id}.json +# +# Output schema: +# { +# "task_id": "", +# "status": "completed", +# "elapsed_seconds": 0.123, +# "enriched_at": "2025-01-15T12:00:00Z", +# "processing_node": "", +# "input": { ...original caller fields (task_id stripped)... }, +# "output": { +# "word_count": 42, +# "char_count": 210, +# "reversed_keys": ["key3","key2","key1"], +# "summary": "Enriched N field(s) from input." +# } +# } +# +# Exit codes: +# 0 — task completed; result file written +# 1 — task failed (error printed to stderr) +# --------------------------------------------------------------------------- +set -eo pipefail + +RESULTS_DIR="${RESULTS_DIR:-/data/results}" +COS_PREFIX="interactive-fleet-result" + +START_NS=$(date +%s%N) + +# --------------------------------------------------------------------------- +# 1. Read input JSON from $1 +# The function passes the entire payload as a single JSON string argument. +# --------------------------------------------------------------------------- +if [[ $# -lt 1 || -z "$1" ]]; then + echo "[task] ERROR: no input JSON provided as \$1" >&2 + exit 1 +fi + +INPUT_JSON="$1" +echo "[task] Received input: ${INPUT_JSON}" >&2 + +# Validate it is parseable JSON +if ! jq -e . > /dev/null 2>&1 <<< "${INPUT_JSON}"; then + echo "[task] ERROR: \$1 is not valid JSON: ${INPUT_JSON}" >&2 + exit 1 +fi + +# --------------------------------------------------------------------------- +# 2. Resolve task_id from the JSON payload or fall back to CE env vars / uuidgen +# --------------------------------------------------------------------------- +TASK_ID=$(jq -r '(.task_id // "") | select(. != "")' <<< "${INPUT_JSON}" || true) +TASK_ID="${TASK_ID:-${CE_TASK_INDEX:-}}" +TASK_ID="${TASK_ID:-${CE_TASK_ID:-}}" +TASK_ID="${TASK_ID:-$(uuidgen | tr '[:upper:]' '[:lower:]')}" + +echo "[task] Processing task_id=${TASK_ID}" >&2 + +# --------------------------------------------------------------------------- +# 3. Build the clean input object (strip internal routing keys) +# --------------------------------------------------------------------------- +CLEAN_INPUT=$(jq 'del(.task_id)' <<< "${INPUT_JSON}") + +# --------------------------------------------------------------------------- +# 4. Compute enrichment values +# --------------------------------------------------------------------------- + +# Concatenate all string values for word/char counting +ALL_TEXT=$(jq -r '[to_entries[] | .value | tostring] | join(" ")' <<< "${CLEAN_INPUT}") +WORD_COUNT=$(echo "${ALL_TEXT}" | wc -w | tr -d ' ') +CHAR_COUNT=${#ALL_TEXT} +FIELD_COUNT=$(jq 'length' <<< "${CLEAN_INPUT}") + +# Reversed key list as a JSON array +REVERSED_KEYS=$(jq '[keys_unsorted | reverse[]]' <<< "${CLEAN_INPUT}") + +SUMMARY="Enriched ${FIELD_COUNT} field(s) from input." + +# --------------------------------------------------------------------------- +# 5. Compute elapsed time (seconds with 4 decimal places) +# --------------------------------------------------------------------------- +END_NS=$(date +%s%N) +# Pure bash integer arithmetic — no awk needed +_ELAPSED_NS=$(( END_NS - START_NS )) +_ELAPSED_S=$(( _ELAPSED_NS / 1000000000 )) +_ELAPSED_MS=$(( (_ELAPSED_NS % 1000000000) / 100000 )) +ELAPSED=$(printf "%d.%04d" "${_ELAPSED_S}" "${_ELAPSED_MS}") + +# --------------------------------------------------------------------------- +# 6. Assemble the result JSON +# --------------------------------------------------------------------------- +ENRICHED_AT=$(date -u +"%Y-%m-%dT%H:%M:%SZ") +# Read hostname from /etc/hostname — no external binary needed on ubi-minimal +PROCESSING_NODE=$(cat /etc/hostname 2>/dev/null || echo "unknown") + +# Use --arg (string) for numeric values then convert with tonumber inside jq, +# avoiding --argjson failures when shell produces non-JSON-safe strings. +RESULT=$(jq -n \ + --arg task_id "${TASK_ID}" \ + --arg enriched_at "${ENRICHED_AT}" \ + --arg processing_node "${PROCESSING_NODE}" \ + --arg elapsed "${ELAPSED}" \ + --argjson input "${CLEAN_INPUT}" \ + --arg word_count "${WORD_COUNT}" \ + --arg char_count "${CHAR_COUNT}" \ + --argjson reversed_keys "${REVERSED_KEYS}" \ + --arg summary "${SUMMARY}" \ + '{ + task_id: $task_id, + status: "completed", + elapsed_seconds: ($elapsed | tonumber), + enriched_at: $enriched_at, + processing_node: $processing_node, + input: $input, + output: { + word_count: ($word_count | tonumber), + char_count: ($char_count | tonumber), + reversed_keys: $reversed_keys, + summary: $summary + } + }') + +# --------------------------------------------------------------------------- +# 7. Write result to COS mount +# Path: /data/results/interactive-fleet-result/.json +# --------------------------------------------------------------------------- +mkdir -p "${RESULTS_DIR}/${COS_PREFIX}" +OUT_PATH="${RESULTS_DIR}/${COS_PREFIX}/${TASK_ID}.json" +echo "${RESULT}" > "${OUT_PATH}" + +echo "[task] Done. Result written to ${OUT_PATH}" >&2 +echo "${RESULT}"