Streams API
When Redpanda Connect 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.
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.
{
"<string, stream id>": "<object, a standard {page-component-title} stream configuration>"
}
Response 400
A configuration was invalid, or has linting errors. If linting errors were detected then a JSON response is provided of the form:
{
"linting_errors": [
"<a description of the error"
]
}
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 Redpanda Connect 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 400
The configuration was invalid, or has linting errors. If linting errors were detected then a JSON response is provided of the form:
{
"linting_errors": [
"<a description of the error"
]
}
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
.
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 Redpanda Connect 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 400
The configuration was invalid, or has linting errors. If linting errors were detected then a JSON response is provided of the form:
{
"linting_errors": [
"<a description of the error"
]
}
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.
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
.
Response 400
The configuration was invalid, or has linting errors. If linting errors were detected then a JSON response is provided of the form:
{
"linting_errors": [
"<a description of the error"
]
}
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
.