Streams API¶
When Benthos is run in streams mode, it will open up an HTTP REST API for creating and managing independent streams of data instead of creating a single stream.
Each stream has its own input, buffer, pipeline, and output sections which contains an isolated stream of data with its own lifetime. A stream config cannot include resources, and instead, these should be created and modified using the /resources/{type}/{id} endpoint.
A walkthrough on using this API can be found here.
API¶
GET /ready¶
Returns a 200 OK response if all active streams are connected to their respective inputs and outputs at the time of the request. Otherwise, a 503 response is returned along with a message naming the faulty stream.
If zero streams are active, this endpoint still returns a 200 OK response.
GET /streams¶
Returns a map of existing streams by their unique identifiers to an object showing their status and uptime.
Response 200
{
"<string, stream id>": {
"active": "<bool, whether the stream is running>",
"uptime": "<float, uptime in seconds>",
"uptime_str": "<string, human readable string of uptime>"
}
}
POST /streams¶
Sets the entire collection of streams to the body of the request. Streams that exist but aren't within the request body are removed, streams that exist already and are in the request body are updated, other streams within the request body are created.
Response 200
The streams were updated successfully.
Response 400
A configuration was invalid or has linting errors. If linting errors were detected, then a JSON response is provided of the form:
If you wish for the streams API to proceed with configurations that contain linting errors, then you can override this check by setting the URL param chilled to true, e.g. /streams?chilled=true.
POST /streams/{id}¶
Create a new stream identified by id by posting a body containing the stream configuration in either JSON or YAML format. The configuration should be a standard Benthos configuration containing the sections input, buffer, pipeline, and output.
Request Body Example
URL: /streams/foo
input:
file:
paths: [ /tmp/input.ndjson ]
pipeline:
processors:
- mapping: root = content().uppercase()
output:
file:
path: /tmp/output.ndjson
Response 200
The stream was created successfully.
Response 400
The configuration was invalid, or has linting errors. If linting errors were detected, then a JSON response is provided of the form:
If you wish for the streams API to proceed with configurations that contain linting errors, then you can override this check by setting the URL param chilled to true, e.g. /streams/foo?chilled=true.
GET /streams/{id}¶
Read the details of an existing stream identified by id.
Response 200
{
"active": "<bool, whether the stream is running>",
"uptime": "<float, uptime in seconds>",
"uptime_str": "<string, human readable string of uptime>",
"config": "<object, the configuration of the stream>"
}
PUT /streams/{id}¶
Update an existing stream identified by id by posting a body containing the new stream configuration in either JSON or YAML format. The configuration should be a standard Benthos configuration containing the sections input, buffer, pipeline, and output.
The previous stream will be shut down before and a new stream will take its place.
Response 200
The stream was updated successfully.
Response 400
The configuration was invalid, or has linting errors. If linting errors were detected then a JSON response is provided of the form:
If you wish for the streams API to proceed with configurations that contain linting errors then you can override this check by setting the URL param chilled to true, e.g. /streams/foo?chilled=true.
PATCH /streams/{id}¶
Update an existing stream identified by id by posting a body containing only changes to be made to the existing configuration. The existing configuration will be patched with the new fields and the stream restarted with the result.
Response 200
The stream was patched successfully.
DELETE /streams/{id}¶
Attempt to shut down and remove a stream identified by id.
Response 200
The stream was found, shut down, and removed successfully.
GET /streams/{id}/stats¶
Read the metrics of an existing stream as a hierarchical JSON object.
Response 200
The stream was found.
POST /resources/{type}/{id}¶
Add or modify a resource component configuration of a given type identified by a unique id. The configuration must be in JSON or YAML format and must only contain configuration fields for the component.
Valid component types are cache, input, output, processor, and rate_limit.
Request Body Example
URL: /resources/cache/foo
Response 200
The resource was created successfully.
Response 400
The configuration was invalid or has linting errors. If linting errors were detected, then a JSON response is provided of the form:
If you wish for the streams API to proceed with configurations that contain linting errors, then you can override this check by setting the URL param chilled to true, e.g. /resources/cache/foo?chilled=true.