Tasks API

Overview

The Tasks API manages the work task queue for asynchronous operations like pipeline execution, ML training, and digital twin updates. These endpoints require authentication when WORKER_AUTH_TOKEN is configured.

List Work Tasks

Returns the current queue length.

curl -X GET http://localhost:8080/api/worktasks \
  -H "Authorization: Bearer your-token"

Response

queue_length

number

Current number of tasks in the queue

Submit Work Task

Submits a new work task to the queue.

curl -X POST http://localhost:8080/api/worktasks \
  -H "Authorization: Bearer your-token" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "pipeline_execution",
    "priority": 1,
    "project_id": "proj-123",
    "task_spec": {
      "pipeline_id": "pipe-456",
      "parameters": {}
    },
    "resource_requirements": {
      "cpu": "1",
      "memory": "2Gi",
      "gpu": false
    },
    "data_access": {
      "input_datasets": ["dataset-1"],
      "output_location": "s3://output/",
      "storage_credentials": "creds-secret"
    }
  }'

Request Body

type

string

required

Task type: pipeline_execution, ml_training, ml_inference, or digital_twin_update

priority

number

Task priority (higher = higher priority, defaults to 1)

project_id

string

required

Project ID

task_spec

object

required

Task-specific parameters

Show TaskSpec

pipeline_id

string

Pipeline ID (for pipeline_execution)

model_id

string

Model ID (for ml_training/ml_inference)

parameters

object

Additional parameters

preferred_cluster

string

Preferred cluster (empty = auto)

resource_requirements

object

required

Resource requirements

Show Resources

cpu

string

CPU request (e.g., “1”, “2”)

memory

string

Memory request (e.g., “2Gi”, “4Gi”)

gpu

boolean

Whether GPU is required

data_access

object

required

Data access configuration

Show DataAccess

input_datasets

array

Input dataset IDs

output_location

string

Output storage location

storage_credentials

string

Storage credentials reference

Response

worktask_id

string

Work task ID

type

string

Task type

status

string

Current status: queued, scheduled, spawned, executing, completed, failed, timeout, or cancelled

priority

number

Task priority

submitted_at

string

ISO 8601 timestamp

started_at

string

ISO 8601 timestamp (if started)

completed_at

string

ISO 8601 timestamp (if completed)

project_id

string

Project ID

task_spec

object

Task-specific parameters

resource_requirements

object

Resource requirements

data_access

object

Data access configuration

error_message

string

Error message (if failed)

kubernetes_job_name

string

Kubernetes job name (if spawned)

cluster_name

string

Cluster task was dispatched to

retry_count

number

Number of retry attempts

max_retries

number

Maximum retry attempts

retry_reason

string

Reason for retry

Get Work Task

Returns a single work task by ID.

curl -X GET http://localhost:8080/api/worktasks/task-123 \
  -H "Authorization: Bearer your-token"

Path Parameters

string

required

Work task ID

Response

Returns the work task object (see Submit Work Task for schema).

Update Work Task Status

Updates the status of a work task. Called by worker jobs to report progress.

curl -X POST http://localhost:8080/api/worktasks/task-123 \
  -H "Authorization: Bearer your-token" \
  -H "Content-Type: application/json" \
  -d '{
    "worktask_id": "task-123",
    "status": "completed",
    "output_location": "s3://output/result.json",
    "metadata": {
      "records_processed": 1000
    }
  }'

Path Parameters

string

required

Work task ID

Request Body

worktask_id

string

required

Work task ID (must match path parameter)

status

string

required

New status: executing, completed, failed, timeout, or cancelled

output_location

string

Output location (for completed tasks)

metadata

object

Additional metadata

error_message

string

Error message (for failed tasks)

Response

{
  "status": "updated"
}

Or for retryable errors:

{
  "status": "requeued_for_retry"
}

Automatic Retry

The system automatically retries tasks with transient errors (OOMKilled, Evicted, DeadlineExceeded). Tasks are requeued with incremented retry count up to the maximum retry limit.

Get Metrics

Returns queue and task metrics.

curl -X GET http://localhost:8080/api/metrics

Response

queue

object

Queue metrics

Show Queue

length

number

Current queue length

tasks_by_status

object

Task counts grouped by status

Show Example

queued

number

Number of queued tasks

executing

number

Number of executing tasks

completed

number

Number of completed tasks

failed

number

Number of failed tasks

tasks_by_type

object

Task counts grouped by type

Show Example

pipeline_execution

number

Number of pipeline execution tasks

ml_training

number

Number of ML training tasks

ml_inference

number

Number of ML inference tasks

digital_twin_update

number

Number of digital twin update tasks

timestamp

string

ISO 8601 timestamp of metrics snapshot

WebSocket Updates

Subscribe to real-time task updates via WebSocket:

const ws = new WebSocket('ws://localhost:8080/ws/tasks');

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  console.log('Task update:', message);
};

Message Format

{
  "event": "task_update",
  "task": {
    "worktask_id": "task-123",
    "status": "completed",
    "completed_at": "2024-01-15T10:35:00Z"
  }
}

Overview

Endpoints

Overview

List Work Tasks

Response

Submit Work Task

Request Body

Response

Get Work Task

Path Parameters

Response

Update Work Task Status

Path Parameters

Request Body

Response

Automatic Retry

Get Metrics

Response

WebSocket Updates

Message Format

Build docs developers (and LLMs) love

Overview

Endpoints

​Overview

​List Work Tasks

​Response

​Submit Work Task

​Request Body

​Response

​Get Work Task

​Path Parameters

​Response

​Update Work Task Status

​Path Parameters

​Request Body

​Response

​Automatic Retry

​Get Metrics

​Response

​WebSocket Updates

​Message Format

Build docs developers (and LLMs) love

Overview

List Work Tasks

Response

Submit Work Task

Request Body

Response

Get Work Task

Path Parameters

Response

Update Work Task Status

Path Parameters

Request Body

Response

Automatic Retry

Get Metrics

Response

WebSocket Updates

Message Format