# Amazon CloudWatch Logs Source Connector for Confluent Platform The AWS CloudWatch Logs Source connector is used to import data from AWS CloudWatch Logs, and write them into a Kafka topic. Moreover, the connector sources from a single log group and writes to one topic per log stream. There is a topic format configuration available to customize the topic names of each log stream. If specific customizations for topics such as multiple log streams writing to the same topic are desired, SMTs can be used for such actions. This connector can start at one task supporting all importation of data and can scale up to one task per log stream which will raise performance to the highest that Amazon supports (100,000 logs per second or 10 MB per second). ## Features The AWS CloudWatch Logs Source connector includes the following features: - [At least once delivery](#aws-cloudwatch-logs-at-least-once-delivery) - [Multiple tasks](#aws-cloudwatch-logs-multiple-tasks) - [Topic format customizability](#aws-cloudwatch-logs-topic-customizability) - [Log stream selection](#aws-cloudwatch-logs-stream-selection) - [Query implementation options](#aws-cloudwatch-logs-query-implementation-options) ### At least once delivery Records imported from AWS CloudWatch Logs are delivered with at least once semantics. Duplicates will generally be limited, however, as there will only be repeats in the chance of unexpected termination of the connector. ### Multiple tasks The AWS CloudWatch Logs Source connector supports running one or more tasks. You can specify the number of tasks in the `tasks.max` configuration parameter. Multiple tasks may improve performance when moving a large amount of data. ### Topic format customizability Because this connector is designed to write to a topic per log stream, custom topic formats can be created or all records can be written to exactly one topic. ### Log stream selection The log streams from which logs are imported from can be specified, or as a default, all will be used. ### Query implementation options Users can switch between different implementation approaches to query events based on their requirements: - **GetLogEvents**: Queries events one stream at a time, making it suitable for focused log analysis. This implementation is capped at 50 log streams. - **FilterLogEvents**: Queries events across multiple streams simultaneously, which enables broader log analysis. This implementation can handle more than 50 log streams, making it ideal for large-scale deployments with extensive logging infrastructure. The choice between implementations allows users to optimize performance and resource utilization based on their specific use case and infrastructure scale. ## Install the AWS CloudWatch Logs Connector You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - Kafka Broker: Confluent Platform 3.3.0 or later. - Connect: Confluent Platform 4.1.0 or later. - Java 1.8. - AWS account. - At least one AWS CloudWatch log group and log stream in AWS CloudWatch Logs. - At minimum, AWS permissions `logs:GetLogEvents`, `logs:FilterLogEvents`, `logs:DescribeLogGroups`, and `logs:DescribeLogStreams` are required for this connector. Confluent recommends you review the [CloudWatch Logs Permissions Reference](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/permissions-reference-cwl.html). - An installation of the latest (`latest`) connector version. To install the `latest` connector version, navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install confluentinc/kafka-connect-aws-cloudwatch-logs:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent connect plugin install confluentinc/kafka-connect-aws-cloudwatch-logs:2.0.1-SNAPSHOT ``` ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-aws-cloudwatch-logs) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins). ## License You can use this connector for a 30-day trial period without a license key. After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information. See [Confluent Platform license](aws_cloudwatch_logs_source_connector_config.md#aws-cloudwatch-logs-source-connector-license-config) for license properties and [License topic configuration](aws_cloudwatch_logs_source_connector_config.md#aws-cloudwatch-logs-source-license-topic-configuration) for information about the license topic. ## Linux on IBM Z (s390x) support Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x). The connector supports the same capability available on x86_64 unless otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support). ## Configuration Properties For a complete list of configuration properties for this connector, see [Configuration Reference for Amazon CloudWatch Logs Source Connector for Confluent Platform](aws_cloudwatch_logs_source_connector_config.md#aws-cloudwatch-logs-source-connector-config). ## Upgrading to Version 2.x Starting with version 2.x, the AWS CloudWatch Logs Source connector uses the AWS SDK v2, upgrading from v1. This upgrade does not allow backward compatibility with versions 1.x. If you use any custom or built-in AWS credentials providers, you must update your implementation to ensure compatibility with the new AWS SDK v2. Follow the steps below based on the type of credentials provider your implementation uses. ### Custom credentials provider - Update [credentials provider](#aws-cloudwatch-logs-credentials-providers) interface: If you have implemented a custom credentials provider, you must update your code to implement the new AWS SDK v2 interface. Old interface: [com.amazon.auth.AWSCredentialsProvider](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html?session_ref=direct) New interface: [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) - Set a new instantiation method: The connector no longer uses a default no-args constructor to create an instance of your custom provider. Instead, your credentials provider class must now implement a static, no-argument `create()` method that returns a new instance of your provider. ### AWS built-in credentials provider - Use v2 equivalent: If you use a credentials provider built into the AWS SDK itself, you must ensure you use the v2 equivalent of that provider. - Use `create()` method: This provider must implement a static, no-argument `create()` method that returns a new provider instance. ### Default credentials provider - Review changes to the [default chain](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html): The default credentials provider chain in the AWS SDK v2 has changed the order in which it searches for credentials. The SDK v2 version now checks for system properties before checking for environment variables. This change may affect your connector’s authentication if you previously relied on environment variables and also have system properties set. For more information about credentials provider migration, see [Credentials provider changes for AWS SDK v2](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html). ## Quick Start For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ### Preliminary setup To add a new connector plugin you must restart Connect. Use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) command to restart Connect. ```bash confluent local services connect stop && confluent local services connect start ``` Your output should resemble: ```text Using CONFLUENT_CURRENT: /Users/username/Sandbox/confluent-snapshots/var/confluent.NuZHxXfq Starting Zookeeper Zookeeper is [UP] Starting Kafka Kafka is [UP] Starting Schema Registry Schema Registry is [UP] Starting Kafka REST Kafka REST is [UP] Starting Connect Connect is [UP] ``` Check if the AWS CloudWatch Logs plugin has been installed correctly and picked up by the plugin loader: ```text curl -sS localhost:8083/connector-plugins | jq '.[].class' | grep "cloudwatch logs" ``` Your output should resemble: ```text "io.confluent.connect.aws.cloudwatch.AwsCloudWatchSourceConnector" ``` ### AWS CloudWatch Logs setup You can use the AWS Management Console to set up your AWS CloudWatch log group and log stream as shown in the [Amazon CloudWatch Logs User Guide](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html), or you can complete the following steps: 1. Sign up for an [AWS account](https://docs.aws.amazon.com/streams/latest/dev/before-you-begin.html#setting-up-sign-up-for-aws). 2. Set up [AWS Credentials](#aws-cloudwatch-logs-awscredentials). 3. [Create a log group](https://docs.aws.amazon.com/cli/latest/reference/logs/create-log-group.html) in AWS CloudWatch Logs. ```text aws logs create-log-group --log-group my-log-group ``` 4. [Create a log stream](https://docs.aws.amazon.com/cli/latest/reference/logs/create-log-stream.html) in AWS CloudWatch Logs. ```text aws logs create-log-stream --log-group my-log-group --log-stream my-log-stream ``` 5. [Insert records](https://docs.aws.amazon.com/cli/latest/reference/logs/put-log-events.html) into your log stream. If this is the first time inserting logs into a new log stream, a sequence token is not needed. However, after the first `put-log-event`, a sequence token is returned. You will need the sequence token as a parameter for the next `put-log-event`. ```text aws logs put-log-events --log-group my-log-group --log-stream my-log-stream --log-events timestamp=,message=some-string ``` The example shows a log event at a specified timestamp with a specified message put into the specified log stream and log group. Enter the following command to get a sequence token: ```text aws logs describe-log-streams --log-group my-log-group ``` Output providing the sequence token is displayed: ```json { "logStreams": [ { "logStreamName": "my-log-stream", "creationTime": 1569709821347, "lastIngestionTime": 1569709984113, "uploadSequenceToken": "49595785783592846449895609848346364951147972276040781330", "storedBytes": 0 } ] } ``` The example below shows how you can use the sequence token to generate logs for your stream. ```text aws logs put-log-events --log-group my-log-group --log-stream my-log-stream --log-events timestamp=,message=bananas --sequence-token 49595785783592846449895609848346364951147972276040781330 ``` ### Source Connector Configuration Start the services using the Confluent CLI: ```bash confluent local start ``` Create a configuration file named aws-cloudwatch-logs-source-config.json with the following contents. ```text { "name": "aws-cloudwatch-logs-source", "config": { "connector.class": "io.confluent.connect.aws.cloudwatch.logs.AwsCloudWatchSourceConnector", "tasks.max": "1", "aws.cloudwatch.logs.url": "https://logs.us-east-2.amazonaws.com", "aws.cloudwatch.log.group": "my-log-group", "aws.cloudwatch.log.streams": "my-log-stream", "name": "aws-cloudwatch-logs-source", "confluent.topic.bootstrap.servers": "localhost:9092", "confluent.topic.replication.factor": "1" } } ``` The important configuration parameters used here are: - **aws.cloudwatch.logs.url**: The endpoint URL that the source connector connects to pull the specified logs. - **aws.cloudwatch.log.group**: The AWS CloudWatch log group under which the log streams are contained. - **aws.cloudwatch.log.streams**: A list of AWS CloudWatch log streams from which the logs are pulled from. The default value is to use all log streams from the configured log group. - **tasks.max**: The maximum number of tasks that should be created for this connector. - You may pass your [AWS Credentials](https://docs.confluent.io/kafka-connect-kinesis/current/index.html#aws-credentials) to the AWS CloudWatch Logs Connector through your source connector configuration. To pass AWS credentials in the source configuration set the **aws.access.key.id** and the **aws.secret.key.id**: parameters. ```text "aws.access.key.id": "aws.secret.access.key": ``` Run this command to start the AWS CloudWatch Logs Source connector. ```bash confluent local load aws-cloudwatch-logs-source --config aws-cloudwatch-logs-source-config.json ``` To check that the connector started successfully view the Connect worker’s log by running: ```bash confluent local services connect log ``` Start a Kafka Consumer in a separate terminal session to view the data exported by the connector into the kafka topic ```text path/to/confluent/bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic my-log-group.my-log-stream --from-beginning ``` Finally, stop the Confluent services using the command: ```bash confluent local stop ``` ### Remove unused resources [Delete your log group](https://docs.aws.amazon.com/cli/latest/reference/logs/delete-log-group.html) and clean up resources to avoid incurring any unintended charges. ```text aws logs delete-log-group --log-group my-log-group ``` ### AWS Credentials By default, the AWS CloudWatch Logs connector looks for AWS credentials in the following locations and in the following order: 1. The `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` environment variables accessible to the Connect worker processes where the connector will be deployed. These variables are recognized by the AWS CLI and all AWS SDKs (except for the AWS SDK for .NET). You use export to set these variables. ```bash export AWS_ACCESS_KEY_ID= export AWS_SECRET_ACCESS_KEY= ``` The `AWS_ACCESS_KEY` and `AWS_SECRET_KEY` can be used instead, but are not recognized by the AWS CLI. 2. The `aws.accessKeyId` and `aws.secretKey` Java system properties on the Connect worker processes where the connector will be deployed. However, these variables are only recognized by the AWS SDK for Java and are not recommended. 3. The `~/.aws/credentials` file located in the home directory of the operating system user that runs the Connect worker processes. These credentials are recognized by most AWS SDKs and the AWS CLI. Use the following AWS CLI command to create the credentials file: ```bash aws configure ``` You can also manually create the credentials file using a text editor. The file should contain lines in the following format: ```bash [default] aws_access_key_id = aws_secret_access_key = ``` #### NOTE When creating the credentials file, make sure that the user creating the credentials file is the same user that runs the Connect worker processes and that the credentials file is in this user’s home directory. Otherwise, the AWS CloudWatch Logs connector will not be able to find the credentials. See [AWS Credentials File Format](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format) for additional details. Choose one of the above to define the AWS credentials that the AWS CloudWatch Logs connectors use, verify the credentials implementation is set correctly, and then restart all of the Connect worker processes. #### NOTE Confluent recommends using either **Environment variables** or a **Credentials file** because these are the most straightforward, and they can be checked using the AWS CLI tool before running the connector. ### Credentials Providers A *credentials provider* is a Java class that implements the [software.amazon.awssdk.auth.credentials .AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) interface in the AWS Java library and returns AWS credentials from the environment. By default the AWS CloudWatch Logs connector configuration property `aws.credentials.provider.class` uses the [software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) class. This class and interface implementation chains together five other credential provider classes. The [software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html) implementation looks for credentials in the following order: 1. **Environment variables** using the [software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/EnvironmentVariableCredentialsProvider.html) class implementation. This implementation uses environment variables `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`. Environment variables `AWS_ACCESS_KEY` and `AWS_SECRET_KEY` are also supported by this implementation; however, these two variables are only recognized by the AWS SDK for Java and are not recommended. 2. **Java system properties** using the [software.amazon.awssdk.auth.credentials.SystemPropertyCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/SystemPropertyCredentialsProvider.html) class implementation. This implementation uses Java system properties `aws.accessKeyId` and `aws.secretKey`. 3. **Credentials file** using the [software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/ProfileCredentialsProvider.html) class implementation. This implementation uses a credentials file located in the path `~/.aws/credentials`. This credentials provider can be used by most AWS SDKs and the AWS CLI. Use the following AWS CLI command to create the credentials file: ```bash aws configure ``` You can also manually create the credentials file using a text editor. The file should contain lines in the following format: ```bash [default] aws_access_key_id = aws_secret_access_key = ``` #### NOTE When creating the credentials file, make sure that the user creating the credentials file is the same user that runs the Connect worker processes and that the credentials file is in this user’s home directory. Otherwise, the AWS CloudWatch Logs connector will not be able to find the credentials. See [AWS Credentials File Format](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format) for additional details. ### Using Trusted Account Credentials This connector can assume a role and use credentials from a separate trusted account. This is a default feature provided with recent versions of this connector that include an updated version of the AWS SDK. #### IMPORTANT You cannot use assumed role credentials to access AWS through a proxy server without first passing [environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-proxy.html) or [system properties](https://confluence.atlassian.com/kb/how-to-configure-outbound-http-and-https-proxy-for-your-atlassian-application-834000120.html). This is due to an AWS SDK [limitation](https://github.com/aws/aws-sdk-java/issues/2558). After you create the trust relationship, an IAM user or an application from the trusted account can use the [AWS Security Token Service (AWS STS)](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html) `AssumeRole` API operation. This operation provides temporary security credentials that enable access to AWS resources for the connector. For details, see [Creating a Role to Delegate Permissions to an IAM User](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user.html). Example: : ```bash Profile in ~/.aws/credentials:
[default] role_arn=arn:aws:iam::037803949979:role/kinesis_cross_account_role source_profile=staging role_session_name = OPTIONAL_SESSION_NAME
[staging] aws_access_key_id = aws_secret_access_key = ``` To allow the connector to assume a role with the right permissions, set the [Amazon Resource Name (ARN)](https://docs.aws.amazon.com/credref/latest/refdocs/setting-global-role_arn.html) for this role. Additionally, you must choose between `source_profile` or `credential_source` as the way to get credentials that have permission to assume the role, in the environment where the connector is running. #### NOTE When setting up trusted account credentials, be aware that the approach of loading profiles from both `~/.aws/credentials` and `~/.aws/config` does not work when configuring this connector. Assumed role settings and credentials must be placed in the `~/.aws/credentials` file. ### Using Other Implementations You can use a different credentials provider. To do this, set the `aws.credentials.provider.class` property to the name of any class that implements the [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) interface. #### IMPORTANT If you are using a different credentials provider, do not include the `aws.access.key.id` and `aws.secret.key.id` in the connector configuration file. If these parameters are included, they will override the custom credentials provider class. Complete the following steps to use a different credentials provider: 1. Find or create a Java credentials provider class that implements the [software.amazon.awssdk.auth. credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) interface. 2. Put the class file in a JAR file. 3. Place the JAR file in the `share/java/kafka-connect-aws-cloudwatch-logs` directory on **all Connect workers**. 4. Restart the Connect workers. 5. Change the AWS CloudWatch Logs connector property file to use your custom credentials. Add the provider class entry `aws.credentials.provider.class=` in the AWS CloudWatch Logs connector properties file. #### IMPORTANT You must use the fully qualified class name in the `` entry. ## Examples ### Property based example ```properties name=aws-cloudwatch-logs-source-connector connector.class=io.confluent.connect.aws.cloudwatch.logs.AwsCloudWatchSourceConnector tasks.max=1 aws.access.key.id=< Optional Configuration > aws.secret.access.key=< Optional Configuration > aws.cloudwatch.log.group=< Required Configuration > aws.cloudwatch.log.streams=< Optional Configuration > confluent.topic.bootstrap.servers=localhost:9092 confluent.topic.replication.factor=1 ``` ### REST based example This configuration is used typically along with [distributed workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to `connector.json`, configure all of the required values, and use the command below to post the configuration to one the distributed connect worker(s). Check here for more information about the Kafka Connect [REST API](/platform/current/connect/references/restapi.html). ```bash { "name" : "aws-cloudwatch-logs-source-connector", "config" : { "name" : "aws-cloudwatch-logs-source-connector", "connector.class" : "io.confluent.connect.aws.cloudwatch.logs.AwsCloudWatchSourceConnector", "tasks.max" : "1", "aws.access.key.id" : "< Optional Configuration >", "aws.secret.access.key" : "< Optional Configuration >", "aws.cloudwatch.log.group" : "< Required Configuration >", "aws.cloudwatch.log.streams : "< Optional Configuration - defaults to all log streams in the log group >" } } ``` Use curl to post the configuration to one of the Kafka Connect Workers. Change `http://localhost:8083/` the endpoint of one of your Kafka Connect workers. ```bash curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors ``` ```bash curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json \ http://localhost:8083/connectors/aws-cloudwatch-logs-source-connector/config ```