Ctrlk
For the complete documentation index, see llms.txt. This page is also available as Markdown.

HTTP source

Source

The HTTP Source connector turns an e6 Ingestion Engine pipeline into an HTTP server that accepts POST requests. Each request body becomes one or more records in the pipeline. It is useful for webhooks, application event ingestion, and any integration where you want to push data directly into a streaming pipeline without an intermediate message broker.

The connector runs an embedded Axum HTTP server inside the e6 Ingestion Engine worker process. There is no connection profile - the server config (port, path, auth) lives entirely in the table config. Incoming request bodies are buffered in a bounded in-memory channel, deserialized according to the schema you define, and emitted as Arrow record batches for downstream processing. The server also transparently decompresses gzip-encoded request bodies (Content-Encoding: gzip).

Source Config

The table config tells e6 Ingestion Engine what port to listen on, which path to accept requests at, how large requests can be, and how much to buffer before dropping events.

port: 8080
path: /events
buffer_size: 1000
max_body_size: 5242880

  • port (required, u16) - Port number the embedded HTTP server binds to. The e6 Ingestion Engine worker listens on this port locally. To expose it externally, use a Kubernetes Service/Ingress or Docker port mapping. There is no default - you must specify it.

  • path (String, default "/") - HTTP endpoint path where POST requests are accepted. Requests to any other path receive a 404. Use this to namespace endpoints when running multiple HTTP source pipelines on different paths behind a shared ingress, e.g. /webhooks/stripe or /events/clickstream.

  • bind_address (String, default "0.0.0.0") - Network interface to bind to. 0.0.0.0 listens on all interfaces (the right choice for containers and production). Use 127.0.0.1 to restrict to localhost during local development.

  • buffer_size (usize, default 1000) - Capacity of the in-memory channel between the HTTP handler and the pipeline processing loop. Each accepted request occupies one slot. When the buffer is full, new requests are rejected with HTTP 429 Too Many Requests so clients can back off (see Response Codes). The default of 1000 assumes roughly 100 KB payloads, targeting around 100 MB of memory usage. Increase for smaller payloads, decrease for larger ones. This value also controls the concurrency semaphore - at most buffer_size requests are processed concurrently.

  • max_body_size (usize, default 5242880 / 5 MB) - Maximum allowed request body size in bytes. Requests exceeding this limit are rejected with 413 Payload Too Large by the Axum body limit layer before the handler ever sees the data. Set this based on your expected payload sizes to protect the worker from memory pressure caused by unexpectedly large requests.

  • auth (optional) - Authentication configuration. When set, every request must include a valid Authorization header. Unauthenticated requests receive 401 Unauthorized. See Authentication below.

  • service_name (String, default "") - An identifier used to generate the Kubernetes Service resource name for this HTTP source. When set, the controller creates a Kubernetes Service named http-src-{service_name} so that other services in the cluster can route traffic to this pipeline. If left empty, the default naming applies. Only relevant for Kubernetes deployments.

Authentication

The HTTP source supports optional authentication to protect the endpoint. When configured, every incoming request must carry a valid Authorization header - requests without one, or with incorrect credentials, receive a 401 Unauthorized response and the body bytes are tracked in the bytes_rejected_auth metric.

Bearer Token - the simplest option. The client sends Authorization: Bearer <token> and e6 Ingestion Engine does a constant-time string comparison:

Basic Auth - the client sends Authorization: Basic <base64(username:password)>. e6 Ingestion Engine base64-decodes the header and compares the username:password string:

Parallelism Constraint

Parallelism must be 1. The HTTP source binds to a specific TCP port, and only one OS process can bind to a given port at a time. If the pipeline is configured with parallelism greater than 1, the operator panics at startup with a clear error message. This is enforced in the on_start hook - e6 Ingestion Engine checks ctx.task_info.parallelism and aborts before attempting to bind.

If you need to scale HTTP ingestion throughput beyond what a single worker can handle, place a load balancer or ingress controller in front and run multiple independent pipelines on different ports, or use a message broker (Kafka, Redpanda) as a fan-out layer.

Relay and Replay

Unlike Kafka or CDC, an HTTP producer has no durable offset to rewind to. If the source is restarting, recovering, or shedding load (returning 429/503), any data pushed during that window is simply lost - the producer would have to retry it itself. The relay and replay services close this gap, giving HTTP ingestion broker-like durability without introducing a message broker. Both are optional: a deployment that can tolerate dropped requests, or that already fronts the source with its own queue, can skip them.

Relay is an HTTP reverse proxy you place in front of the source pods. It forwards each request to the source and, whenever the source returns a 5xx/429 or is unreachable, transparently buffers the request body to object storage (S3/GCS/Azure) and still returns 202 Accepted to the producer. Producers never observe the outage. The relay discovers which pipelines to buffer for automatically - the controller labels each HTTP source's Kubernetes Service so the relay can find it.

Replay is the catch-up half. It runs as a periodic job (on a cron schedule rather than as a long-lived daemon): each pass reads the buffered objects back and re-POSTs them to the source once it has recovered, then exits. To avoid racing the relay as it writes the current buffer, replay stays a couple of minutes behind real time, and it checkpoints its progress per topic so an interrupted pass resumes where it left off. Delivery is at-least-once - if your pipeline needs to deduplicate replayed records, do so using a content-derived key.

Together they form a durable buffer in front of the source: the relay absorbs and persists traffic during outages, and replay drains that buffer back into the pipeline after recovery. See the architecture overview for where these sit in the system.

Schema

The schema block defines the expected structure of incoming request bodies. It is required - the HTTP source connector will not start without it.

See the Schema Reference for field type details.

Format

The format field determines how request bodies are deserialized into typed Arrow columns:

  • json: {} - JSON (the most common choice for HTTP payloads)

  • avro: {} - Apache Avro

  • protobuf: {} - Protocol Buffers

  • parquet: {} - Apache Parquet

  • raw_string: {} - Raw text, no parsing

  • raw_bytes: {} - Raw binary, no parsing

Framing

By default, each HTTP request body is treated as a single record. If you want to send multiple records in one request, configure framing to split the body into individual records before deserialization:

With newline_delimited framing, the body is split on newline characters. Each line is deserialized independently. This works well with newline-delimited JSON (NDJSON):

Without framing, that entire body would be treated as a single record (and likely fail JSON deserialization since it is not valid JSON).

Bad Data Handling

The bad_data field controls what happens when a record fails deserialization:

  • fail: {} (default) - The pipeline stops with an error. Use this when data quality is critical and you want to catch schema mismatches immediately.

  • drop: {} - The bad record is silently dropped and a warning is logged (rate-limited to avoid log flooding). The pipeline continues processing subsequent records. Use this when you expect occasional malformed data and prefer availability over completeness.

Sending Data

Send data to the running pipeline with any HTTP client. The only requirement is a POST request to the configured path with a body matching the expected format.

Single record:

With bearer auth:

Multiple records (newline-delimited), gzip compressed:

The server automatically decompresses gzip-encoded bodies before processing. No special config is needed - the Content-Encoding: gzip header is sufficient.

Response Codes

The HTTP source returns truthful status codes: 200 on success, 429 when overloaded, and 503 when the pipeline is shutting down. The handler stays non-blocking - the response is sent as soon as the request is enqueued or rejected, without waiting for downstream processing - but clients and load balancers can now back off correctly when the engine sheds load.

Code
Meaning
When it happens

200

OK

Request accepted and buffered for processing.

400

Bad Request

The request body is empty (zero bytes).

401

Unauthorized

Authentication is configured but the Authorization header is missing, malformed, or contains incorrect credentials.

413

Payload Too Large

The request body exceeds the max_body_size limit. Enforced by the Axum body limit layer before the handler runs.

429

Too Many Requests

The in-flight concurrency limit or internal buffer is saturated. Clients should back off and retry.

503

Service Unavailable

The pipeline is shutting down or its internal channel is closed. Clients should retry against another endpoint.

Clients should treat 429 and 503 as retryable load-shedding signals. For monitoring rejection rates and buffer pressure in production, use the Prometheus metrics http_source_events_dropped_total and http_source_buffer_messages.

Complete Example

A webhook ingestion pipeline that accepts JSON events with bearer token authentication, drops malformed records, and handles newline-delimited batches.

Table config:

Schema:

Sending data:

JSON Schema Reference

Connection Table Schema
PreviousFilesystem
NextOverview

Last updated

Was this helpful?