.. _aws-cloudwatch-logs-source-connector:

|kconnect-long| |aws| CloudWatch Logs Source Connector
======================================================

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 10MB per second).


Prerequisites
-------------

The following are required to run the |kconnect-long| |aws| CloudWatch Logs Connector:

* |ak| Broker: |cp| 3.3.0 or above
* |kconnect|: |cp| 4.1.0 or above
* Java 1.8
* |aws| account
* At least one |aws| CloudWatch log group and log stream in |aws| CloudWatch Logs

Features
--------

The |aws| CloudWatch Logs connector offers a variety of features:

* **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.

* **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.

Install the |aws| CloudWatch Logs Connector
-------------------------------------------

.. include:: ../includes/connector-install.rst

.. include:: ../includes/connector-install-hub.rst

.. codewithvars:: bash

   confluent-hub install confluentinc/kafka-connect-aws-cloudwatch:latest

.. include:: ../includes/connector-install-version.rst

.. codewithvars:: bash

   confluent-hub install confluentinc/kafka-connect-aws-cloudwatch:1.0.0-preview

--------------------------
Install Connector Manually
--------------------------
`Download and extract the ZIP file <https://www.confluent
.io/connector/kafka-connect-aws-cloudwatch/#download>`__ for your
connector and then follow the manual connector installation :ref:`instructions <connect_install_connectors>`.

License
-------

.. include:: ../includes/enterprise-license.rst

See :ref:`aws-cloudwatch-logs-source-connector-license-config` for license properties and :ref:`aws-cloudwatch-logs-source-license-topic-configuration` for information about the license topic.


.. _aws_cloudwatch_logs_quickstart:

|aws| CloudWatch Logs Source Connector Quick Start
--------------------------------------------------

-----------------
Preliminary Setup
-----------------

To add a new connector plugin you must restart |kconnect|. Use the
:ref:`Confluent CLI <cli>` command to restart |kconnect|:

.. codewithvars:: bash

   |confluent_stop| connect && |confluent_start| connect

Your output should resemble:

::

   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:

::

   curl -sS localhost:8083/connector-plugins | jq '.[].class' | grep cloudwatch

Your output should resemble:

::

   "io.confluent.connect.aws.cloudwatch.AwsCloudWatchSourceConnector"

.. _cloudwatch-awscredentials:

-----------------
|aws| Credentials
-----------------

By default, the |aws| CloudWatch Logs connector looks for |aws| credentials in the following
locations and in the following order:

#. 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.

   .. sourcecode:: bash

      export AWS_ACCESS_KEY_ID=<your_access_key_id>
      export AWS_SECRET_ACCESS_KEY=<your_secret_access_key>

   The ``AWS_ACCESS_KEY`` and ``AWS_SECRET_KEY`` can be used instead, but are not recognized by the |aws| CLI.

#. 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.

#. 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:

   .. sourcecode:: bash

      aws configure

   You can also manually create the credentials file using a text editor. The file should contain lines in the following format:

   .. sourcecode:: bash

      [default]
      aws_access_key_id = <your_access_key_id>
      aws_secret_access_key = <your_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 kinesis 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 `com.amazon.auth
.AWSCredentialsProvider <https://docs.aws.amazon
.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/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 `com.amazon.auth.DefaultAWSCredentialsProviderChain <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html>`__ class. This class and
interface implementation chains together five other credential provider classes.

The `com.amazonaws.auth.DefaultAWSCredentialsProviderChain <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html>`__ implementation looks for credentials in the following order:

#. **Environment variables** using the `com.amazonaws.auth.EnvironmentVariableCredentialsProvider <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/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.

#. **Java system properties** using the `com.amazonaws.auth.SystemPropertiesCredentialsProvider <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/SystemPropertiesCredentialsProvider.html>`__ class implementation. This implementation uses Java system properties ``aws.accessKeyId`` and ``aws.secretKey``.

#. **Credentials file** using the `com.amazonaws.auth.profile.ProfileCredentialsProvider <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/profile/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:

   .. sourcecode:: bash

      aws configure

   You can also manually create the credentials file using a text editor. The file should contain lines in the following format:

   .. sourcecode:: bash

      [default]
      aws_access_key_id = <your_access_key_id>
      aws_secret_access_key = <your_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 kinesis 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.

.. _aws-cloud-other-credentials-implementations:

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 `com.amazon.auth.AWSCredentialsProvider <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/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:

#. Find or create a Java credentials provider class that implements the `com.amazon.auth.AWSCredentialsProvider <https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/AWSCredentialsProvider.html>`__ interface.

#. Put the class file in a JAR file.

#. Place the JAR file in the ``share/java/kafka-connect-aws-cloudwatch`` directory on **all Connect workers**.

#. Restart the Connect workers.

#. Change the |aws| CloudWatch Logs connector property file to use your custom credentials. Add the provider class
   entry ``aws.credentials.provider.class=<className>`` in the |aws| CloudWatch Logs connector properties file.

   .. important::

      You must use the fully qualified class name in the ``<className>`` entry.


---------------------------
|aws| CloudWatch Logs Setup
---------------------------

You can use the |aws| Management Console to set up your |aws| CloudWatch log group and log
stream as shown `here`_ or you can complete the following steps:

#. Make sure you have an `AWS account`_.
#. Set up :ref:`cloudwatch-awscredentials`.
#. `Create a log group`_ in |aws| CloudWatch Logs.

   ::

      aws logs create-log-group --log-group my-log-group

#. `Create a log stream`_ in |aws| CloudWatch Logs.

   ::

      aws logs create-log-stream --log-group my-log-group --log-stream my-log-stream

#. `Insert Records`_ into your log stream. If this is the first time inserting logs into a new
   log stream, then no sequence token is needed. However, after the first put, a
   sequence token is returned. You will need this token as a parameter for the next put.

   ::

      aws logs put-log-events --log-group my-log-group --log-stream my-log-stream --log-events timestamp=<time>,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:

   ::
    
     aws logs describe-log-streams --log-group my-log-group
      
   Output providing the sequence token is displayed:
 
   .. code-block:: 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.

   ::
    
     aws logs put-log-events --log-group my-log-group --log-stream my-log-stream --log-events timestamp=<time>,message=bananas --sequence-token 49595785783592846449895609848346364951147972276040781330
    

------------------------------
Source Connector Configuration
------------------------------

Start the services using the |confluent-cli|:

.. codewithvars:: bash

   |confluent_start|

Create a configuration file named aws-cloudwatch-logs-source-config.json with the following
contents.


::

   {
     "name": "aws-cloudwatch-logs-source",
     "config": {
       "connector.class": "io.confluent.connect.aws.cloudwatch.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 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 :ref:`awscredentials` 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.

   ::

      "aws.access.key.id":<your-access-key-id>
      "aws.secret.access.key":<your-secret-access-key>

Run this command to start the |aws| CloudWatch Logs source connector.

.. include:: ../../includes/confluent-local-consume-limit.rst

.. codewithvars:: bash

   |confluent_load| aws-cloudwatch-logs-source|dash| -d aws-cloudwatch-logs-source-config.json


To check that the connector started successfully view the Connect
worker's log by running:

.. codewithvars:: bash

   |confluent_log| connect

Start a Kafka Consumer in a separate terminal session to view the data exported by
the connector into the kafka topic

::

    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:

.. codewithvars:: bash

   |confluent_stop|

-----------------------
Remove unused resources
-----------------------

`Delete your log group`_ and clean up resources to avoid incurring any
unintended charges.

::

   aws logs delete-log-group --log-group my-log-group


.. _aws-cloudwatch-source-connector-examples:

Examples
--------

----------------------
Property based example
----------------------

.. codewithvars:: properties
    :name: connector.properties
    :emphasize-lines: 6

    name=aws-cloudwatch-logs-source-connector
    connector.class=io.confluent.connect.aws.cloudwatch.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 :ref:`distributed workers <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
|kconnect-long| :ref:`REST API <connect_userguide_rest>`.

.. codewithvars:: bash
    :emphasize-lines: 8,9

    {
      "name" : "aws-cloudwatch-logs-source-connector",
      "config" : {
        "name" : "aws-cloudwatch-logs-source-connector",
        "connector.class" : "io.confluent.connect.aws.cloudwatch.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 |kconnect-long| Workers. Change
``http://localhost:8083/`` the endpoint of one of your |kconnect-long| workers.

.. codewithvars:: bash

    curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors


.. codewithvars:: bash

    curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json \
    http://localhost:8083/connectors/aws-cloudwatch-logs-source-connector/config

------------------------
Additional Documentation
------------------------

.. toctree::
   :maxdepth: 1

   aws_cloudwatch_logs_source_connector_config

.. _this: #AWS-Credentials
.. _AWS credentials: https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html
.. _Default AWS Credentials Provider Chain: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html
.. _here: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html
.. _AWS account: https://docs.aws.amazon.com/streams/latest/dev/before-you-begin.html#setting-up-sign-up-for-aws
.. _Create a log group: https://docs.aws.amazon.com/cli/latest/reference/logs/create-log-group.html
.. _Create a log stream: https://docs.aws.amazon.com/cli/latest/reference/logs/create-log-stream.html
.. _Insert Records: https://docs.aws.amazon.com/cli/latest/reference/logs/put-log-events.html
.. _Profile Credentials Provider: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/profile/ProfileCredentialsProvider.html
.. _AWS Credentials File Format: https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format
.. _Environment Variable Credentials Provider: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/EnvironmentVariableCredentialsProvider.html
.. _System Properties Credentials Provider: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/SystemPropertiesCredentialsProvider.html
.. _SystemPropertiesCredentialsProvider: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/SystemPropertiesCredentialsProvider.html
.. _AWSCredentialsProvider: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html
.. _Configurable interface: https://kafka.apache.org/21/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html
.. _Delete your log group: https://docs.aws.amazon.com/cli/latest/reference/logs/delete-log-group.html