polling an api

The http_poll input sends a GET request on a timer. It sends the whole reply into the pipeline on each read. It is the snapshot mode of the database inputs, for a system that has only an API.

// config.json — the machine list from an ERP, every hour
{ "type": "http_poll",
  "url": "https://erp.example.com/api/machines",
  "interval_secs": 3600,
  "items": "/data/machines",
  "auth": { "type": "bearer", "token": "${ERP_TOKEN}" },
  "max_batch": 500 }

The input has no connection. The url and the auth are on the component.

what a reply becomes

• A JSON array becomes one message for each element.
• Any other JSON value becomes one message.
• items is a JSON pointer into a reply that wraps its records. For example, /data/machines reads the array at data.machines. The input splits the value at the pointer with the same rules.
• If the pointer finds nothing, the read fails. Thus a change in the shape of the API shows as an error, and the pipeline does not stop without a sign.

performance

Set max_batch. The default is 1. With the default, a reply of 500 machines makes 500 passes through the run loop. The input does not wait for a batch to fill.

snapshots only

Use this input for reference data: a list of machines, recipes, sites or thresholds that changes rarely. The input reads the whole list each time and sends all of it. It does not know what changed since the last read.

This works well when the list is small and the destination keeps the latest value for each key. Examples are a Tidepool table with a primary key, or a remember transform keyed by id. A row that arrives again changes nothing. A row that was lost arrives on the next read.

There is no incremental mode. APIs use many different methods to ask for new rows, for example a cursor parameter, a Link header, a page number or a since value. If the data grows too fast to read whole on each interval, send it through a broker or to the http input.

The input does not see deletes. A machine that is not in the list is not sent. If deletes are important, make the API report them, for example with "active": false.

failures, the interval and the reply size

• interval_secs: the wait after the end of one read and before the next read. The first read occurs when the pipeline starts.
• timeout_seconds: the longest time for one request. The default is 30.
• Reply size: the input holds a reply in memory, so the limit is 64 MiB. A larger reply fails the read.

These problems fail a read:

• a host that is not reachable;
• a status that is not 2xx (the error quotes the reply of the API);
• a body that is not JSON;
• an items pointer that finds nothing.

The input reports a failure one time and retries with backoff. After a read succeeds, the interval starts again.

credentials and metadata

auth is the same block as on the http output: a bearer token or a header that you name. Write the value as a ${NAME} reference.

With an envelope, each message carries:

• the url that it came from, without a username or password;
• a polled_at time, which is the same for all messages of one read.

http_poll

Gets a url at an interval and sends the full reply each time.

Each read is a GET. A reply that is an array gives one message for each element. Any other reply gives one message. Use items for a reply that has the records inside it, for example {"data": {"machines": [...]}}.

Use this input for reference data, for example a list of machines or recipes that changes rarely. The input has no watermark and no pages. Each read sends all records again. Send them to an output that writes the latest value for each key.

A read fails when the url cannot be reached, when the status is not 2xx, when the body is not JSON, or when items finds nothing. The input reports the failure one time and tries again with backoff. The interval starts again after the next successful read.

field	type		description
url	string	required	The url to get, for example https://erp.example.com/api/machines.
interval_secs	integer	required	The time between two reads, in s. The time starts at the end of one read. The first read occurs when the pipeline starts.
auth	bearer | header	optional	The credential that the input sends. If you do not set it, the input sends no credential.
items	string	optional	The position of the records in the reply, as a JSON pointer. For example, /data/machines reads the array at data.machines. If you do not set it, the input uses the full reply. An array gives one message for each element. Any other value gives one message.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only messages that are already read into a batch. It does not wait for a batch to fill.
timeout_seconds	integer	optional	The maximum time for one request, in s. The default is 30.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

auth — type: "bearer"

field	type		description
token	string	required	The token. Use a ${NAME} reference, so that the config file keeps only the name and the secret store keeps the value.

auth — type: "header"

field	type		description
name	string	required	The name of the header. The http input compares the name without case. On an http input, the name must not be a header that an envelope copies into the messages.
value	string	required	The exact value of the header. Use a ${NAME} reference.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
url	The url that the input read the message from, with no username and no password.
polled_at	The start time of the read that returned the message, as RFC 3339. All messages of one read have the same value. Use it to tell one snapshot from the next.