> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/http-source.md). # 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](https://github.com/tokio-rs/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. ```yaml 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](#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 `bytes_rejected_auth` metric. **Bearer Token** - the simplest option. The client sends `Authorization: Bearer ` and e6 Ingestion Engine does a constant-time string comparison: ```yaml auth: type: bearer token: "my-secret-token" ``` **Basic Auth** - the client sends `Authorization: Basic `. e6 Ingestion Engine base64-decodes the header and compares the `username:password` string: ```yaml 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 - 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](/ingestion-engine/architecture/overview.md#system-model) 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. ```yaml 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](/ingestion-engine/get-started/schema.md) 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 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 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 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 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 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 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 `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:** ```yaml 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 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 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 { "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"] } ```
--- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/connectors/http-source.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.