> ## Documentation Index > Fetch the complete documentation index at: https://docs.e6data.com/llms.txt > Use this file to discover all available pages before exploring further. # HTTP source **Source** The HTTP Source connector turns an Ingest 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](https://github.com/tokio-rs/axum) HTTP server inside the Ingest 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 Ingest 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} port: 8080 path: /events buffer_size: 1000 max_body_size: 5242880 ``` * **`port`** (required, `u16`) - Port number the embedded HTTP server binds to. The Ingest 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](#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](#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 `laminar_worker_http_source_bytes_rejected_total` metric (label `reason`). **Bearer Token** - the simplest option. The client sends `Authorization: Bearer ` and Ingest Engine does a constant-time string comparison: ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} auth: type: bearer token: "my-secret-token" ``` **Basic Auth** - the client sends `Authorization: Basic `. Ingest Engine base64-decodes the header and compares the `username:password` string: ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} auth: type: basic username: "user" password: "pass" ``` ### 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 - Ingest Engine checks `ctx.task_info.parallelism` and aborts before attempting to bind. If you need to scale HTTP ingestion throughput beyond what a single job can handle, use a [Pipeline Replica Set](/ingest/concepts/pipeline-replica-sets), which runs multiple independent replicas of the pipeline behind one load-balanced endpoint. Alternatively, 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. *** ## Schema The schema block defines the expected structure of incoming request bodies. It is required - the HTTP source connector will not start without it. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} schema: format: json: {} fields: - field_name: event_type field_type: type: primitive: Utf8 nullable: false - field_name: payload field_type: type: primitive: Utf8 nullable: true ``` See the [Schema Reference](/ingest/reference/schema) 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: ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} schema: format: json: {} framing: newline_delimited: {} fields: # ... ``` With `newline_delimited` framing, the body is split on newline characters. Each line is deserialized independently. This works well with newline-delimited JSON (NDJSON): ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST http://localhost:8080/events \ -d '{"event_type": "a", "user": "alice"} {"event_type": "b", "user": "bob"} {"event_type": "c", "user": "carol"}' ``` 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. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} schema: format: json: {} bad_data: drop: {} fields: # ... ``` *** ## 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:** ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST http://localhost:8080/events \ -H "Content-Type: application/json" \ -d '{"event_type": "order.placed", "payload": "{\"order_id\": 12345}"}' ``` **With bearer auth:** ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST http://localhost:8080/events \ -H "Authorization: Bearer my-secret-token" \ -d '{"event_type": "order.placed", "payload": "{\"order_id\": 12345}"}' ``` **Multiple records (newline-delimited), gzip compressed:** ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} echo '{"event_type": "a"} {"event_type": "b"} {"event_type": "c"}' | gzip | \ curl -X POST http://localhost:8080/events \ -H "Content-Encoding: gzip" \ --data-binary @- ``` 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 `laminar_worker_http_source_events_dropped_total` and `laminar_worker_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:** ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} port: 8080 path: /webhooks/events bind_address: 0.0.0.0 buffer_size: 5000 max_body_size: 10485760 auth: type: bearer token: "${WEBHOOK_SECRET}" service_name: webhook-ingest ``` **Schema:** ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} schema: format: json: {} framing: newline_delimited: {} bad_data: drop: {} fields: - field_name: event_type field_type: type: primitive: Utf8 nullable: false - field_name: payload field_type: type: primitive: Utf8 nullable: true - field_name: timestamp field_type: type: primitive: Utf8 nullable: false - field_name: source_id field_type: type: primitive: Int64 nullable: true ``` **Sending data:** ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST http://localhost:8080/webhooks/events \ -H "Authorization: Bearer ${WEBHOOK_SECRET}" \ -H "Content-Type: application/json" \ -d '{"event_type": "user.created", "payload": "{\"user_id\": 42}", "timestamp": "2024-01-15T10:30:00Z", "source_id": 1}' ``` *** ## JSON Schema Reference
Connection Table Schema ```json theme={"theme":{"light":"github-light","dark":"github-dark"}} { "type": "object", "properties": { "port": { "type": "integer", "description": "Port number to listen on", "minimum": 1, "maximum": 65535 }, "path": { "type": "string", "description": "HTTP endpoint path for POST requests", "default": "/" }, "bind_address": { "type": "string", "description": "Network interface to bind to", "default": "0.0.0.0" }, "buffer_size": { "type": "integer", "description": "Maximum requests to buffer in memory before dropping", "default": 1000, "minimum": 1 }, "max_body_size": { "type": "integer", "description": "Maximum request body size in bytes", "default": 5242880, "minimum": 1 }, "auth": { "oneOf": [ { "type": "object", "properties": { "type": { "const": "bearer" }, "token": { "type": "string" } }, "required": ["type", "token"] }, { "type": "object", "properties": { "type": { "const": "basic" }, "username": { "type": "string" }, "password": { "type": "string" } }, "required": ["type", "username", "password"] } ] }, "service_name": { "type": "string", "description": "Kubernetes Service name suffix (resource becomes http-src-{service_name})", "default": "" } }, "required": ["port"] } ```
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.