HTTP API

e6 Ingestion Engine exposes an HTTP API for managing pipelines, connections, and jobs programmatically.

Base URL

http://<controller-host>:8000/api/v1

All endpoints below are relative to this base URL.

Authentication

Include the API token in the Authorization header:

curl -H "Authorization: Bearer <token>" http://localhost:8000/api/v1/pipelines

Response Format

All responses are JSON. Errors return:

{
  "error": "descriptive error message"
}

List endpoints use cursor-based pagination:

{
  "data": [...],
  "has_more": true
}

Pass ?starting_after=<last_id>&limit=10 to paginate.

---

Pipelines

Create Pipeline

POST /pipelines

Creates a pipeline and starts it immediately.

Request:

{
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000
}

Field	Type	Required	Description
name	string	yes	Pipeline name
query	string	yes	SQL query defining the pipeline
parallelism	integer	yes	Number of parallel subtasks
checkpoint_interval_micros	integer	no	Checkpoint interval in microseconds (default: 30s)

Response: Pipeline object

---

Create Preview Pipeline

POST /pipelines/preview

Creates a short-lived pipeline (60s TTL) for testing queries interactively.

Request:

{
  "query": "SELECT * FROM source WHERE amount > 100",
  "enable_sinks": false
}

Field	Type	Required	Description
query	string	yes	SQL query to preview
enable_sinks	boolean	no	Whether to write to real sinks (default: false)

Response: Pipeline object

---

List Pipelines

GET /pipelines

Query Parameters:

Parameter	Type	Description
starting_after	string	Cursor for pagination
limit	integer	Items per page (default: 10)

Response:

{
  "data": [Pipeline, ...],
  "has_more": false
}

---

Get Pipeline

GET /pipelines/:id

Response: Pipeline object

{
  "id": "abc123",
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000,
  "stop": "none",
  "created_at": 1700000000000000,
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "action": "stop",
  "action_text": "Stop",
  "action_in_progress": false,
  "preview": false
}

---

Update Pipeline

PATCH /pipelines/:id

Updates pipeline configuration. Only provided fields are changed.

Request:

{
  "parallelism": 8,
  "checkpoint_interval_micros": 60000000,
  "stop": "checkpoint"
}

Field	Type	Description
parallelism	integer	New parallelism (triggers rescaling)
checkpoint_interval_micros	integer	New checkpoint interval
stop	string	Stop mode: "none", "graceful", "immediate", "checkpoint", "force"

Response: Pipeline object

---

Restart Pipeline

POST /pipelines/:id/restart

Restarts a stopped or failed pipeline.

Request:

{
  "force": false
}

Field	Type	Description
force	boolean	Skip taking a final checkpoint before restart (default: false)

Response: Pipeline object

---

Delete Pipeline

DELETE /pipelines/:id

Deletes a pipeline. The pipeline must be in a terminal state (Stopped, Finished, or Failed).

Response: 200 OK

---

Validate Query

POST /pipelines/validate_query

Validates a SQL query without creating a pipeline. Returns the execution graph or validation errors.

Request:

{
  "query": "INSERT INTO sink SELECT * FROM source"
}

Response:

{
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "errors": []
}

---

Jobs

A job represents a single execution of a pipeline. Each time a pipeline starts or restarts, a new job is created.

List Jobs

GET /jobs

Returns all jobs across all pipelines.

GET /pipelines/:pipeline_id/jobs

Returns jobs for a specific pipeline.

Response:

{
  "data": [
    {
      "id": 1,
      "running_desired": true,
      "state": "Running",
      "run_id": 3,
      "start_time": 1700000000000000,
      "finish_time": null,
      "tasks": 8,
      "failure_message": null,
      "created_at": 1700000000000000
    }
  ]
}

---

Get Job Errors

GET /pipelines/:pipeline_id/jobs/:job_id/errors

Returns error and warning messages for a job.

Query Parameters:

Parameter	Type	Description
starting_after	string	Cursor for pagination
limit	integer	Items per page

Response:

{
  "data": [
    {
      "id": "msg-123",
      "created_at": 1700000000000000,
      "operator_id": "operator_2",
      "task_index": 0,
      "level": "Error",
      "message": "Connection refused",
      "details": "..."
    }
  ],
  "has_more": false
}

---

Get Job Checkpoints

GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints

Returns all checkpoints for a job.

Response:

{
  "data": [
    {
      "epoch": 5,
      "backend": "s3",
      "start_time": 1700000150000000,
      "finish_time": 1700000151000000
    }
  ]
}

---

Get Checkpoint Details

GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints/:epoch/operator_checkpoint_groups

Returns per-operator checkpoint details for a specific epoch.

Response:

{
  "data": [
    {
      "operator_id": "operator_0",
      "bytes": 4096,
      "finish_time": 1700000151000000,
      "subtasks": [...]
    }
  ]
}

---

Get Job Metrics

GET /pipelines/:pipeline_id/jobs/:job_id/operator_metric_groups

Returns per-operator metrics for a running job.

---

Get Job Output (SSE)

GET /pipelines/:pipeline_id/jobs/:job_id/output

Streams pipeline output in real-time via Server-Sent Events. Only available for pipelines with a preview sink.

Response: text/event-stream

Each event contains:

{
  "column_names": ["id", "name", "amount"],
  "data": [["1", "Alice", "100"], ["2", "Bob", "200"]]
}

---

Connection Profiles

A connection profile stores reusable connection credentials (e.g., Kafka bootstrap servers, authentication).

Create Connection Profile

POST /connection_profiles

Request:

{
  "name": "production-kafka",
  "connector": "kafka",
  "config": {
    "bootstrap_servers": "kafka-1:9092,kafka-2:9092",
    "authentication": {
      "type": "sasl_plain",
      "username": "user",
      "password": "pass"
    }
  }
}

Response: ConnectionProfile object

Typed endpoints are also available for specific connectors:

POST /connectors/kafka/profiles
POST /connectors/confluent/profiles

---

List Connection Profiles

GET /connection_profiles

Response:

{
  "data": [
    {
      "id": "prof-123",
      "name": "production-kafka",
      "connector": "kafka",
      "config": {...},
      "description": "Kafka cluster"
    }
  ]
}

---

Get Connection Profile

GET /connection_profiles/:id

---

Delete Connection Profile

DELETE /connection_profiles/:id

Fails if the profile is referenced by any connection tables.

---

Test Connection Profile

POST /connection_profiles/test

Tests connectivity with the provided configuration.

Request: Same as create.

Response:

{
  "message": "Successfully connected",
  "done": true
}

---

Autocomplete

GET /connection_profiles/:id/autocomplete

Returns available values for the connection (e.g., Kafka topic names).

Response:

{
  "values": {
    "topics": ["orders", "users", "events"]
  }
}

---

Connection Tables

A connection table defines a specific source or sink - binding a connector, connection profile, schema, and table-specific configuration together.

Create Connection Table

POST /connection_tables

Request:

{
  "name": "orders",
  "connector": "kafka",
  "connection_profile_id": "prof-123",
  "config": {
    "topic": "orders",
    "read_mode": "read_uncommitted"
  },
  "schema": {
    "format": { "json": {} },
    "fields": [
      { "field_name": "id", "field_type": { "primitive": "Int64" } },
      { "field_name": "amount", "field_type": { "primitive": "Float64" } }
    ]
  }
}

Typed endpoints are also available:

POST /connectors/kafka/tables
POST /connectors/confluent/tables
POST /connectors/iceberg/tables
POST /connectors/delta/tables
POST /connectors/filesystem/tables
POST /connectors/kinesis/tables
POST /connectors/http_source/tables
POST /connectors/stdout/tables

Response: ConnectionTable object

{
  "id": 1,
  "pub_id": "tbl-456",
  "name": "orders",
  "created_at": 1700000000000000,
  "connector": "kafka",
  "table_type": "source",
  "config": {...},
  "schema": {...},
  "consumers": 1
}

---

List Connection Tables

GET /connection_tables

Query Parameters:

Parameter	Type	Description
starting_after	string	Cursor for pagination
limit	integer	Items per page

---

Get Connection Table

GET /connection_tables/:id

---

Delete Connection Table

DELETE /connection_tables/:id

---

Test Connection Table

POST /connection_tables/test

Tests reading from a source with the provided configuration. Returns results as a Server-Sent Events stream.

---

Validate Schema

POST /connection_tables/schemas/test

Validates a schema definition.

---

---

Connectors

List Available Connectors

GET /connectors

Returns metadata about all available connectors.

Response:

{
  "data": [
    {
      "name": "kafka",
      "description": "Apache Kafka source and sink",
      "connection_config": {...},
      "table_config": {...}
    }
  ]
}

---

Health

Ping

GET /ping

Returns "Pong" - used for health checks. No authentication required.