> For the complete documentation index, see [llms.txt](https://docs.onum.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.onum.com/listeners/listener-integrations/collect-data-from-amazon-kinesis.md).

# Collect data from Amazon Kinesis

{% hint style="info" %}
See the changelog of the **Amazon Kinesis** Listener [here](/listeners-changelog/amazon-kinesis-data-stream-listener.md).
{% endhint %}

{% hint style="warning" %}
The **Amazon Kinesis** Listener is a **Pull** Listener and therefore should not be used in environments with more than one cluster.
{% endhint %}

## Overview

Onum supports integration with [Amazon Kinesis Data Stream](https://aws.amazon.com//kinesis/data-streams/).

Amazon Kinesis Data Streams is a fully managed, serverless streaming data service that allows you to ingest, store, and process real-time data streams. It's designed for high-throughput, low-latency data ingestion from various sources, enabling real-time analytics and applications.

## Prerequisites

{% hint style="warning" %}
In order to use the **Amazon Kinesis** Listener, you must activate the following environment variable in your distributor using docker compose: `SINGLETON_LISTENER_EXECUTOR=true`
{% endhint %}

## Amazon Kinesis Data Stream Setup

{% stepper %}
{% step %}
Go to **IAM (Identity and Access Management)** to manage users, groups, roles and permissions. 

Under **Permissions Policies**, make sure you have assigned the policy `AmazonKinesisFullAccess` to give full access to Kinesis resources. Alternatively, if you have custom permissions, go to **Policies - Create Policy** and in the **JSON** tab, paste your custom JSON e.g.

```json
{
  "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow"
        "Action": [
        "kinesis:CreateStream",
        "kinesis:DescribeStream",
        "kinesis:PutRecord"
        ],
        "Resource": "*"
 		  }        
	 ]
}
```

{% endstep %}

{% step %}
Run the following command to test the configuration:

```
aws kinesis list-streams
```

If your IAM permission are correct, you'll see a list of streams.
{% endstep %}
{% endstepper %}

## Onum Setup

{% stepper %}
{% step %}
Log in to your Onum tenant and click **Listeners > New listener**.
{% endstep %}

{% step %}
Double-click the **Amazon Kinesis Data Stream** Listener.
{% endstep %}

{% step %}
Enter a **Name** for the new Listener. Optionally, add a **Description** and some **Tags** to identify the Listener.
{% endstep %}

{% step %}
In the **Connection settings** section, click **Select region** and choose the region of your AWS data center. Your region is displayed in the top right-hand corner of your AWS console.
{% endstep %}

{% step %}
Click **Select access key ID** and choose it from your [Secrets](https://docs.onum.com/administration/global-settings/organization-settings/secrets-management), or click **New secret** to generate a new one. 

The **Access Key ID** is found in the **IAM Dashboard** of the **AWS Management Console**.

1. In the left panel, click on **Users**.
2. Select your **IAM user**.
3. Under the **Security Credentials** tab, scroll to **Access Keys**, and you will find existing **Access Key IDs** (but not the secret access key).
   {% endstep %}

{% step %}
Click **Select secret key** and choose it from your [Secrets](https://docs.onum.com/administration/global-settings/organization-settings/secrets-management), or click **New secret** to generate a new one.

Under **Access keys**, you can see your **Access Key IDs**, but AWS **will not show the Secret Access Key**. You must have it saved somewhere. If you don't have the secret key saved, you need to create a new one.

{% hint style="info" %}
Learn more about secrets in Onum in [this article](/settings/organization-settings/secrets-management.md).
{% endhint %}
{% endstep %}

{% step %}
You can enable the **Assume Role** option, which provides a more secure and scalable alternative to traditional credential-based authentication.

Instead of using static credentials, this method leverages an Amazon Resource Name (ARN) to define and assume an IAM role with the necessary permissions to access your SQS queue. By introducing the ARN into the Listener configuration, Falcon Onum can securely assume the designated role and interact with your AWS resources in a controlled and auditable way. To enable access, we need cross-account access from our AWS infrastructure to yours. This is a one-time setup. 

First, you must create an IAM role in your AWS account (or use an existing one) with the following trust policy:

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::721751517261:role/<CustomersRole>"
      },
      "Action": ["sts:AssumeRole", "sts:TagSession"]
    }
  ]
}
```

{% hint style="warning" %}
This trust policy allows our AWS account to assume the role. You can attach whatever permissions you want to the role — the trust policy only controls who can assume it, not what it can do.
{% endhint %}

Once the IAM role is ready, please [contact us](broken://spaces/cSjT21I4EUhzghjc1rER/pages/nW8oZycpfVthtI5KYdQs) to share your full role ARN (e.g. `arn:aws:iam::012345678901:role/YourRoleName`) so we can complete the configuration on our end. Then, configure the **Assume Role** options as follows:

* **Role ARN** -  **Amazon Resource Name** used to access S3 and SQS resources. This is the unique identifier for the specific IAM Role that you want to assume and use.
* **External ID** - Shared secret used to authenticate the usage of this role.
* **Role Session** - Name of the session, used to audit usage of this role ( `s3-listener` by default)
* **STS Region** - If not set, it will use the bucket or the queue region.
* **STS Session Duration** - How much the Assume Role session will last before reauthentication. Uses Golang duration strings, like `1s`, `1m`, `1h`. If not set, it uses the maximum session duration configured for that role. The maximum and default duration is 1h.
  {% endstep %}

{% step %}
Enter the following information in the **Data Stream configuration** section:

* **Stream name** - Enter the unique identifier of your Kinesis Data Stream. To get it:
  1. Go to: <https://console.aws.amazon.com/kinesis>
  2. Select **Data Streams** under **Amazon Kinesis** in the sidebar.
  3. The **Stream name** will be in the first column e.g. `my-kinesis-stream-prod`
* **Shard ID** - The shard is the basic unit of capacity in a Kinesis Data Stream, acting like a partition for your data stream and determining how your data is ingested, stored, and consumed. Click your Data Stream name to find your **Shard ID** in the **Shards** tab, e.g.: `shardId-000000000001`
  {% endstep %}

{% step %}
Activate the **Enable KMS Decryption** option if you want to automatically decrypt KMS-encrypted messages. If you activate it, you'll need to configure the following settings:

* **Select KMS Key ID** - Choose your KMS Key ID from your [Secrets](/settings/organization-settings/secrets-management.md) or click **New secret** to define a new one. To find your KMS Key ID, go to the [KMS console](https://www.google.com/search?q=KMS+console\&sca_esv=fb6ec2c420ab31bb\&rlz=1C5GCCM_en\&ei=wmJvaeqqEMG0i-gP3efY4Qg\&ved=2ahUKEwjz4qXq_ZmSAxVt3QIHHRR6NToQgK4QegQIBBAB\&uact=5\&oq=FIND+MY+kms+kEY+id\&gs_lp=Egxnd3Mtd2l6LXNlcnAiEkZJTkQgTVkga21zIGtFWSBpZDIFECEYnwVIwx9QAFi4HXACeAGQAQCYAa0BoAGiE6oBBDAuMjC4AQPIAQD4AQGYAhagApgUqAIKwgIOEAAYgAQYsQMYgwEYigXCAggQABiABBixA8ICCxAAGIAEGLEDGIMBwgIOEC4YgAQYsQMY0QMYxwHCAgsQLhiABBjRAxjHAcICChAAGIAEGEMYigXCAhAQABiABBixAxhDGIMBGIoFwgIKEC4YgAQYQxiKBcICBRAAGIAEwgIQEAAYAxi0AhjqAhiPAdgBAcICDhAuGIAEGMcBGI4FGK8BwgILEC4YgAQYsQMYgwHCAggQLhiABBixA8ICERAuGIAEGLEDGNEDGIMBGMcBwgIdEC4YgAQYxwEYjgUYrwEYlwUY3AQY3gQY4ATYAQLCAg0QABiABBixAxhDGIoFwgIREC4YgAQYsQMYxwEYjgUYrwHCAgcQABiABBgTwgIIEAAYExgWGB7CAgoQABgTGBYYChgewgIFEAAY7wXCAgYQABgWGB7CAggQABiABBiiBJgDBPEFghVN0Fvc7rq6BgQIARgKugYGCAIQARgUkgcEMi4yMKAHkY4BsgcEMC4yMLgHkRTCBwYwLjcuMTXIB1KACAA\&sclient=gws-wiz-serp) and select **Customer managed keys**. You will find it in the **Key ID** column.
* **Set decryption timeout** - Activate this option if you need to set a timeout for the decryption. Indicate it in the **Seconds** field.
* **Enable KMS encryption context** - You can set an optional set of non-secret key–value pairs that can contain additional contextual information about the data. Use the **Field** and **Value** settings and click **Add field** to add the required pairs.
  {% endstep %}

{% step %}
In the **Advanced Configuration** section, activate the **Use compression** option if you need to compress your data. Choose the required type (**Gzip**, **Bzip2** or **Zlib**).
{% endstep %}

{% step %}
Optionally, enter the **Custom endpoint** if you have a non-default URL that directs API requests to a specific Kinesis service endpoint.
{% endstep %}

{% step %}
Finally, click **Create labels**. Optionally, you can set labels to be used for internal Onum routing of data. By default, data will be set as **Unlabeled**.

{% hint style="info" %}
Learn more about labels in [this article](/listeners/labels.md).
{% endhint %}
{% endstep %}

{% step %}
Click **Create listener** when you're done.
{% endstep %}
{% endstepper %}

## Output Ports <a href="#ports" id="ports"></a>

The **Amazon Kinesis** Listener has two output ports:

* **Default port** - Events are sent through this port if no error occurs while processing them.
* **Error port** - Events are sent through this port if an error occurs while processing them.