The /kinesis endpoint receives data from Amazon Data Firehose (formerly Kinesis Data Firehose). It implements the Firehose HTTP endpoint delivery request and response specification, so you can add Observe as an HTTP Endpoint destination on any Firehose stream.

When to use this endpoint

Use the Kinesis endpoint when your data already flows through Firehose, or when the source can only deliver to Firehose. Common cases:

• CloudWatch Logs subscription filters that deliver to a Firehose stream
• Applications or AWS services that write records to a Firehose stream or Kinesis data stream
• Third-party services that export to a customer-owned Firehose stream

To collect AWS logs and metrics across many accounts, start with the AWS integration instead. It configures the AWS side for you.

Prerequisites

• A Datastream and a datastream token for it. Copy the token when you create it; Observe only shows the secret once.
• Your Observe customer ID. See Where do I find my customer ID?.
• Permission to create or edit Amazon Data Firehose streams in your AWS account.

Configure the Firehose stream

1. In the AWS console, open Amazon Data Firehose and create a stream, or edit an existing one.
2. Set Destination to HTTP Endpoint.
3. Set HTTP endpoint URL to https://${OBSERVE_CUSTOMER}.collect.observeinc.com/v1/kinesis. To change how Observe decodes records, append query parameters.
4. Set Access key to your datastream token. Firehose sends it in the X-Amz-Firehose-Access-Key header, and Observe uses it as the bearer token for the request.
5. Set Content encoding to GZIP. This compresses each request in transit and keeps large buffers under the request size limit.
6. Optionally, add Parameters (key/value pairs). Observe adds them to every observation's EXTRA column, which is a convenient way to tag a stream, for example environment=prod.
7. Under Buffer hints, set Buffer size to 1 MiB. See Request size for why.
8. Configure an S3 backup bucket for failed records, so data is not lost if delivery keeps failing past the retry duration.

Request size

Observe rejects requests larger than 4 MB. Firehose buffer sizes are measured before Firehose base64-encodes each record, which adds about a third to the payload, and Firehose allows buffer sizes far larger than 4 MB. To stay under the limit:

• Keep Buffer size at 1 MiB, and no more than 3 MiB.
• Enable GZIP content encoding.

If requests exceed the limit, Firehose reports delivery errors and eventually writes the records to your S3 backup bucket.

How records are decoded

Each Firehose request contains one or more records. Observe creates at least one observation per record:

1. Base64 decoding. Firehose base64-encodes every record. Observe decodes it.

2. Gzip decompression. Observe tries to gunzip each record. Records that are not gzip-compressed are kept as they are. CloudWatch Logs subscriptions gzip each record, so their data arrives as readable JSON without extra configuration. To keep compressed bytes as they are, use option=passthrough.

3. Splitting large records. A record larger than maxlen (3 MiB by default) is split into several observations:

   • CloudWatch Logs payloads are split by logEvents. Each piece keeps the original owner, logGroup, logStream, subscriptionFilters, and messageType, so every observation is valid CloudWatch JSON.
   • Newline-delimited JSON is split on line boundaries, so no JSON object is cut in half.
   • Other text is split at a UTF-8 character boundary. Every piece after the first has continue set to true, so you can reassemble the text in OPAL.

   Records sent with option=binary are never split.

Query parameters

Append these to the endpoint URL to change decoding:

option can be repeated, for example ?option=binary&option=passthrough. When two values conflict, the last one wins.

Any other query parameter is added to EXTRA. For example, https://${OBSERVE_CUSTOMER}.collect.observeinc.com/v1/kinesis?source=checkout tags every observation from that stream with "source": "checkout".

Data format

Each observation has these fields:

For example, a record from a CloudWatch Logs subscription looks like this after decoding:

{
  "data": "{\"messageType\":\"DATA_MESSAGE\",\"owner\":\"123456789012\",\"logGroup\":\"/aws/lambda/checkout\",\"logStream\":\"2026/11/03/[$LATEST]abc123\",\"subscriptionFilters\":[\"to-observe\"],\"logEvents\":[{\"id\":\"3812...\",\"timestamp\":1793750400000,\"message\":\"START RequestId: ...\"}]}",
  "requestId": "ed4acda5-034f-9f42-bba1-f29aea6d7d8f",
  "timestamp": 1793750401234
}

To work with the payload, parse data in OPAL:

make_col payload:parse_json(string(FIELDS.data))
make_col logGroup:string(payload.logGroup), logStream:string(payload.logStream)

CloudWatch Logs also sends CONTROL_MESSAGE records to check that the destination is reachable. Filter them out with filter string(payload.messageType) = "DATA_MESSAGE".

Responses and retries

Observe answers each request in the Firehose response format. A successful delivery returns 200 OK:

{"requestId": "ed4acda5-034f-9f42-bba1-f29aea6d7d8f", "timestamp": 1793750401500}

A failed delivery returns an error status with an errorMessage. Firehose shows the message in the stream's delivery error logs, retries for the configured retry duration, and then writes the records to your S3 backup bucket. Common errors:

Test the endpoint

You can send a request in the Firehose format with curl. Record data must be base64-encoded:

DATA=$(printf '{"message":"hello from firehose"}' | base64)
curl -sS "https://${OBSERVE_CUSTOMER}.collect.observeinc.com/v1/kinesis" \
  -H "Content-Type: application/json" \
  -H "X-Amz-Firehose-Access-Key: ${DATASTREAM_TOKEN}" \
  -H "X-Amz-Firehose-Request-Id: test-1" \
  -d "{\"requestId\":\"test-1\",\"timestamp\":$(date +%s000),\"records\":[{\"data\":\"${DATA}\"}]}"

A successful response echoes the request ID: {"requestId":"test-1", "timestamp":...}. The observation appears in your datastream's dataset within a few minutes.