[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for Apache Kafka®](../index.md) > [Step-by-step guides](index.md) > Logs and monitoring > Viewing cluster logs

# Viewing Apache Kafka® cluster logs

Managed Service for Apache Kafka® allows you to [get a cluster log snippet](#get-log) for the selected time period and [view logs in real time](#get-log-stream).

{% note info %}

In this context, the log refers to the system logs of a cluster and its hosts. This log has nothing to do with the Apache Kafka® topic partition log where the broker writes messages from producers.

{% endnote %}

{% note info %}

Cluster logs are stored for 45 days.

{% endnote %}

## Getting a cluster log {#get-log}

{% list tabs group=instructions %}

- Management console {#console}

    1. In the [management console](https://console.yandex.cloud), navigate to the relevant folder.
    1. Navigate to **Managed Service for Kafka**.
    1. Click the name of your cluster and select the ![image](../../_assets/console-icons/receipt.svg) **Logs** tab.
    1. Select **Origin**, **Hosts**, and ** Severity**.
    1. Specify a time period for the log entries you want to view.

- CLI {#cli}

    If you do not have the Yandex Cloud CLI yet, [install and initialize it](../../cli/quickstart.md#install).

    The folder used by default is the one specified when [creating](../../cli/operations/profile/profile-create.md) the CLI profile. To change the default folder, use the `yc config set folder-id <folder_ID>` command. You can also specify a different folder for any command using `--folder-name` or `--folder-id`. If you access a resource by its name, the search will be limited to the default folder. If you access a resource by its ID, the search will be global, i.e., through all folders based on access permissions.

    1. See the description of the CLI command for viewing cluster logs:

        ```bash
        yc managed-kafka cluster list-logs --help
        ```

    1. Run the following command to get cluster logs (our example only shows some of the available parameters):

        ```bash
        yc managed-kafka cluster list-logs <cluster_name_or_ID> \
           --limit <entry_number_limit> \
           --columns <list_of_log_columns> \
           --filter <entry_filtration_settings> \
           --since <time_range_left_boundary> \
           --until <time_range_right_boundary>
        ```

        Where:

        * `--limit`: limits on the number of entries to output.
        * `--columns`: List of log columns to draw data from.
            * `hostname`: [Host name](cluster-hosts.md).
            * `message`: Message output by the component.
            * `severity`: Logging level, e.g., `INFO`.
            * `origin`: Message origin, e.g., `kafka_server` or `kafka_controller`.

        * `--filter`: Record filter settings, e.g., `message.hostname='node1.mdb.yandexcloud.net'`.
        * `--since`: Left boundary of a time range in [RFC-3339](https://www.ietf.org/rfc/rfc3339.html), `HH:MM:SS` format or a time interval relative to the current time. Examples: `2006-01-02T15:04:05Z`, `15:04:05`, `2h`, `3h30m ago`.
        * `--until`: right boundary of a time range, the format is similar to that of `--since`.

    You can get the cluster name and ID with the [list of clusters in the folder](cluster-list.md#list-clusters).

- REST API {#api}

    1. [Get an IAM token for API authentication](../api-ref/authentication.md) and put it into an environment variable:

        ```bash
        export IAM_TOKEN="<IAM_token>"
        ```

    1. Call the [Cluster.listLogs](../api-ref/Cluster/listLogs.md) method, e.g., via the following [cURL](https://curl.se/) request:

        ```bash
        curl \
            --request GET \
            --header "Authorization: Bearer $IAM_TOKEN" \
            --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<cluster_ID>:logs' \
            --url-query columnFilter=<list_of_output_data_columns> \
            --url-query fromTime=<time_range_left_boundary> \
            --url-query toTime=<time_range_right_boundary>
        ```

        Where:

        * `columnFilter`: List of output columns:

            * `hostname`: Host name.
            * `component`: Type of component to log, Example: `HTTP-Session`.
            * `message`: Message output by the component.
            * `query_id`: Request ID.
            * `severity`: Logging level, e.g., `Debug`.
            * `thread`: ID of the thread involved in query handling.

            You can specify only one column in the `columnFilter` parameter. If you want to filter logs by more than one column, provide a list of the columns in several parameters.

        * `fromTime`: Left boundary of a time range in [RFC-3339](https://www.ietf.org/rfc/rfc3339.html) format, Example: `2006-01-02T15:04:05Z`.

        * `toTime`: End of the time range in the same format as `fromTime`.

        
        You can request the cluster ID with the [list of clusters in the folder](cluster-list.md#list-clusters).


    1. View the [server response](../api-ref/Cluster/listLogs.md#yandex.cloud.mdb.kafka.v1.ListClusterLogsResponse) to make sure your request was successful.

- gRPC API {#grpc-api}

    1. [Get an IAM token for API authentication](../api-ref/authentication.md) and put it into an environment variable:

        ```bash
        export IAM_TOKEN="<IAM_token>"
        ```

    1. Clone the [cloudapi](https://github.com/yandex-cloud/cloudapi) repository:
       
       ```bash
       cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
       ```
       
       Below, we assume that the repository contents reside in the `~/cloudapi/` directory.

    1. Call the [ClusterService/ListLogs](../api-ref/grpc/Cluster/listLogs.md) method, e.g., via the following [gRPCurl](https://github.com/fullstorydev/grpcurl) request:

        ```bash
        grpcurl \
            -format json \
            -import-path ~/cloudapi/ \
            -import-path ~/cloudapi/third_party/googleapis/ \
            -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/cluster_service.proto \
            -rpc-header "Authorization: Bearer $IAM_TOKEN" \
            -d '{
                    "cluster_id": "<cluster_ID>",
                    "column_filter": [<list_of_output_data_columns>],
                    "from_time": "<time_range_left_boundary>" \
                    "to_time": "<time_range_right_boundary>"
                }' \
            mdb.api.cloud.yandex.net:443 \
            yandex.cloud.mdb.kafka.v1.ClusterService.ListLogs
        ```

        Where:

        * `service_type`: Type of service to request logs for. The only valid value is `CLICKHOUSE`.
        * `column_filter`: List of output data columns:

            * `hostname`: Host name.
            * `component`: Type of component to log, Example: `HTTP-Session`.
            * `message`: Message output by the component.
            * `query_id`: Request ID.
            * `severity`: Logging level, e.g., `Debug`.
            * `thread`: ID of the thread involved in query handling.

            You can specify more than one column in the `column_filter` parameter if you want to filter logs by multiple columns.

        * `from_time`: Left boundary of a time range in [RFC-3339](https://www.ietf.org/rfc/rfc3339.html) format, Example: `2006-01-02T15:04:05Z`.

        * `to_time`: End of the time range in the same format as `from_time`.

        
        You can request the cluster ID with the [list of clusters in the folder](cluster-list.md#list-clusters).


    1. Check the [server response](../api-ref/grpc/Cluster/listLogs.md#yandex.cloud.mdb.kafka.v1.ListClusterLogsResponse) to make sure your request was successful.

{% endlist %}

## Getting a log stream {#get-log-stream}

This method allows you to get cluster logs in real time.

{% list tabs group=instructions %}

- CLI {#cli}

    If you do not have the Yandex Cloud CLI yet, [install and initialize it](../../cli/quickstart.md#install).

    The folder used by default is the one specified when [creating](../../cli/operations/profile/profile-create.md) the CLI profile. To change the default folder, use the `yc config set folder-id <folder_ID>` command. You can also specify a different folder for any command using `--folder-name` or `--folder-id`. If you access a resource by its name, the search will be limited to the default folder. If you access a resource by its ID, the search will be global, i.e., through all folders based on access permissions.

    To view cluster logs in real time, run this command:

    ```bash
    yc managed-kafka cluster list-logs <cluster_name_or_ID> --follow
    ```

    You can get the cluster name and ID with the [list of clusters in the folder](cluster-list.md#list-clusters).

- REST API {#api}

    1. [Get an IAM token for API authentication](../api-ref/authentication.md) and put it into an environment variable:

        ```bash
        export IAM_TOKEN="<IAM_token>"
        ```

    1. Call the [Cluster.streamLogs](../api-ref/Cluster/streamLogs.md) method, e.g., via the following [cURL](https://curl.se/) request:

        ```bash
        curl \
            --request GET \
            --header "Authorization: Bearer $IAM_TOKEN" \
            --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<cluster_ID>:stream_logs' \
            --url-query columnFilter=<list_of_output_data_columns> \
            --url-query fromTime=<time_range_left_boundary> \
            --url-query toTime=<time_range_right_boundary> \
            --url-query filter=<log_filter>
        ```

        Where:

        * `columnFilter`: List of output columns:

            * `hostname`: Host name.
            * `component`: Type of component to log, Example: `HTTP-Session`.
            * `message`: Message output by the component.
            * `query_id`: Request ID.
            * `severity`: Logging level, e.g., `Debug`.
            * `thread`: ID of the thread involved in query handling.

            You can specify only one column in the `columnFilter` parameter. If you want to filter logs by more than one column, provide a list of the columns in several parameters.

        * `fromTime`: Left boundary of a time range in [RFC-3339](https://www.ietf.org/rfc/rfc3339.html) format, Example: `2006-01-02T15:04:05Z`.

        * `toTime`: End of the time range in the same format as `fromTime`.

            If you omit this parameter, new logs will be sent to the log stream as they arrive. Semantically, this behavior is similar to `tail -f`.

        * `filter`: Log filter. You can use a filter to stream only the logs you need.

            To learn more about filters and their syntax, see the [API reference](../api-ref/Cluster/streamLogs.md#query_params).

            {% note tip %}
            
            A filter can contain quotation marks and other characters. Escape them if you need to.
            
            {% endnote %}
            
            Supported filters:
            
            * `message.hostname`: Filtering by host name.
            * `message.severity`: Filtering by logging level.

        
        You can get the cluster ID with the [list of clusters in the folder](cluster-list.md#list-clusters).


    1. View the [server response](../api-ref/Cluster/streamLogs.md#yandex.cloud.mdb.kafka.v1.StreamLogRecord) to make sure your request was successful.

- gRPC API {#grpc-api}

    1. [Get an IAM token for API authentication](../api-ref/authentication.md) and put it into an environment variable:

        ```bash
        export IAM_TOKEN="<IAM_token>"
        ```

    1. Clone the [cloudapi](https://github.com/yandex-cloud/cloudapi) repository:
       
       ```bash
       cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
       ```
       
       Below, we assume that the repository contents reside in the `~/cloudapi/` directory.

    1. Call the [ClusterService/StreamLogs](../api-ref/grpc/Cluster/streamLogs.md) method, e.g., via the following [gRPCurl](https://github.com/fullstorydev/grpcurl) request:

        ```bash
        grpcurl \
            -format json \
            -import-path ~/cloudapi/ \
            -import-path ~/cloudapi/third_party/googleapis/ \
            -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/cluster_service.proto \
            -rpc-header "Authorization: Bearer $IAM_TOKEN" \
            -d '{
                    "cluster_id": "<cluster_ID>",
                    "column_filter": [<list_of_output_data_columns>],
                    "from_time": "<time_range_left_boundary>",
                    "to_time": "<time_range_right_boundary>",
                    "filter": "<log_filter>"
                }' \
            mdb.api.cloud.yandex.net:443 \
            yandex.cloud.mdb.kafka.v1.ClusterService.StreamLogs
        ```

        Where:

        * `column_filter`: List of output columns:

            * `hostname`: Host name.
            * `component`: Type of component to log, Example: `HTTP-Session`.
            * `message`: Message output by the component.
            * `query_id`: Request ID.
            * `severity`: Logging level, e.g., `Debug`.
            * `thread`: ID of the thread involved in query handling.

            You can specify more than one column in the `column_filter` parameter if you want to filter logs by multiple columns.

        * `from_time`: Left boundary of a time range in [RFC-3339](https://www.ietf.org/rfc/rfc3339.html) format, Example: `2006-01-02T15:04:05Z`.

        * `to_time`: End of the time range in the same format as `from_time`.

            If you omit this parameter, new logs will be sent to the log stream as they arrive. Semantically, this behavior is similar to `tail -f`.

        * `filter`: Log filter. You can use a filter to stream only the logs you need.

            {% note tip %}
            
            A filter can contain quotation marks and other characters. Escape them if you need to.
            
            {% endnote %}
            
            Supported filters:
            
            * `message.hostname`: Filtering by host name.
            * `message.severity`: Filtering by logging level.

            To learn more about filters and their syntax, see the [API reference](../api-ref/grpc/Cluster/streamLogs.md).

        
        You can get the cluster ID with the [list of clusters in the folder](cluster-list.md#list-clusters).


    1. Check the [server response](../api-ref/grpc/Cluster/streamLogs.md#yandex.cloud.mdb.kafka.v1.StreamLogRecord) to make sure your request was successful.

{% endlist %}