Confluent Cloud Metrics¶

Metrics in Confluent Cloud are available either through first-class integrations with third-party monitoring providers or by directly querying the Confluent Cloud Metrics API. Users who would like to monitor Confluent Cloud are encouraged to use an integration to reduce the operational burden of monitoring. The Confluent Cloud Metrics API supports a diverse set of querying patterns to support usage and performance analysis over time.

This page is meant to be instructional and to help you get started with using the metrics that Confluent Cloud provides. For more information on the Confluent Cloud Metrics API, see the API Reference.

Metrics Quick Start¶

Prerequisites

Access to Confluent Cloud
Internet connectivity

Note

The Confluent Cloud RBAC MetricsViewer role provides service account access to the Metrics API for all clusters in an organization. This role also enables service accounts to import metrics into third-party metrics platforms. For details, refer to Add the MetricsViewer role to a new service account in the Confluent Cloud Console below.

Create a Cloud API key to authenticate to the Metrics API. For example:

confluent login
confluent api-key create --resource cloud

Note

You must use a Cloud API Key to communicate with the Metrics API. Using the Cluster API Key that is used to communicate with Kafka will result in an authentication error.

Add the MetricsViewer role to a new service account in the Confluent Cloud Console¶

The MetricsViewer role provides service account access to the Metrics API for all clusters in an organization. This role also enables service accounts to import metrics into third-party metrics platforms.

To assign the MetricsViewer role to a new service account:

In the top-right administration menu (☰) in the upper-right corner of the Confluent Cloud user interface, click ADMINISTRATION > Cloud API keys.
Click Add key.
Click the Granular access tile to set the scope for the API key. Click Next.
Click Create a new one and specify the service account name, and optionally, a description. Click Next.
The API key and secret are generated for the service account. You will need this API key and secret to connect to the cluster, so be sure to safely store this information. Click Save. The new service account with the API key and associated ACLs is created. When you return to the API access tab, you can view the newly-created API key to confirm.
Return to Accounts & access in the administration menu, and in the Accounts tab, click Service accounts to view your service accounts.
Select the service account that you want to assign the MetricsViewer role to.
In service account’s details page, click Access.
In the tree view, open the resource where you want the service account to have the MetricsViewer role.
Click Add role assignment and select the MetricsViewer tile. Click Save.

When you return to Accounts & access, you can view the resources for the organization, and also see that the service account you created has the MetricsViewer role binding.

Add the MetricsViewer role to a service account using the CLI¶

Run the following commands to add a role binding for MetricsViewer to a new service account. Remember to log in with the confluent login command first.

Create the service account:

confluent iam service-account create MetricsImporter --description "A test service account to import Confluent Cloud metrics into our monitoring system"

Your output should resemble:

+-------------+--------------------------------+
| ID          | sa-123abc                      |
| Name        | MetricsImporter                |
| Description | A test service account to      |
|             | import Confluent Cloud metrics |
|             | into our monitoring system     |
+-------------+--------------------------------+

Make note of the ID field.

Add the MetricsViewer role binding to the service account:

confluent iam rbac role-binding create --role MetricsViewer --principal User:sa-123abc

Your output should resemble:

+-----------+----------------+
| Principal | User:sa-123abc |
| Role      | MetricsViewer  |
+-----------+----------------+

List the role bindings to confirm that the MetricViewer role was created:

confluent iam rbac role-binding list --principal User:sa-123abc

Your output should resemble:

    Principal    | Email |     Role      | Environment | ...
-----------------+-------+---------------+-------------+----
  User:sa-123abc |       | MetricsViewer |             |

List the existing service accounts:

confluent iam service-account list

Your output should resemble:

     ID     |              Name              |           Description
------------+--------------------------------+-----------------------------------
  sa-1a2b3c | test-account                   | for testing
  sa-112233 | ProactiveSupport.1614189731753 | SA for Proactive Support
  sa-aabbcc | KSQL.lksqlc-ab123              | SA for KSQL w/ ID lksqlc-ab123
            |                                | and Name ksqlDB_app_0
  ...

Create an API key and add it to the new service account:

confluent api-key create --resource cloud --service-account sa-123abc

Your output should resemble:

It may take a couple of minutes for the API key to be ready.
Save the API key and secret. The secret is not retrievable later.
+---------+------------------------------------------------------------------+
| API Key | 1234567ABCDEFGHI                                                 |
| Secret  | ABCDEF123456.................................................... |
+---------+------------------------------------------------------------------+

Save the API key and secret in a secure location.

Integrate with third-party monitoring¶

Integrating directly with a third-party monitoring tool allows you to monitor Confluent Cloud alongside the rest of your applications.

Datadog¶

Datadog provides an integration where users can input a Cloud API key into the Datadog UI, select resources to monitor, and see metrics in minutes using an out-of-the-box dashboard. If you use Datadog, create your Cloud API key and follow the instructions from Datadog to get started.

Dynatrace¶

Dynatrace provides an extension where users can input a Cloud API key into the Dynatrace Monitoring Configuration, select resources to monitor, and see metrics in minutes in a prebuilt dashboard. If you use Dynatrace, create your Cloud API key and follow the instructions to get started.

Grafana Cloud¶

Grafana Labs provides an integration where users can input a Cloud API key into the Grafana Cloud UI, select resources to monitor, and see metrics in minutes using an out-of-the-box-dashboard. If you use Grafana Cloud, create your Cloud API key and follow the instructions to get started.

Prometheus¶

Prometheus servers can scrape the Confluent Cloud Metrics API directly by making use of the export endpoint. This endpoint returns the single most recent data point for each metric, for each distinct combination of labels in the Prometheus exposition or Open Metrics format. For more information, see Export metric values.

New Relic OpenTelemetry¶

You can collect metrics about your Confluent Cloud-managed Kafka deployment with the New Relic OpenTelemetry collector. The collector is a component of OpenTelemetry that collects, processes, and exports telemetry data to New Relic, or any observability back-end. For more information, see Collector for Confluent Cloud and Kafka monitoring.

Discovery using the Metrics API¶

The following examples use HTTPie rather than cURL. This software package can be installed using most common software package managers by following the documentation .

The Confluent Cloud Metrics API provides endpoints for programmatic discovery of available resources and their metrics. This resource and metric metadata is represented by descriptor objects.

The discovery endpoints can be used to avoid hardcoding metric and resource names into client scripts.

Discover available resources¶

A resource represents the entity against which metrics are collected, for example, a Kafka cluster, a Kafka Connector, a ksqlDB application, etc.

Get a description of the available resources by sending a GET request to the descriptors/resources endpoint of the API:

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/descriptors/resources' --auth '<API_KEY>:<SECRET>'

This returns a JSON document describing the available resources to query and their labels.

Discover available metrics¶

Get a description of the available metrics by sending a GET request to the descriptors/metrics endpoint of the API:

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/descriptors/metrics?resource_type=kafka' --auth '<API_KEY>:<SECRET>'

Note

The resource_type query parameter is required to specify the type of resource for which to list metrics. The valid resource types can be determined using the /descriptors/resources endpoint.

This returns a JSON document describing the available metrics to query and their labels.

A human-readable list of the current metrics is available in the API Reference.

Example Queries¶

The Confluent Cloud Metrics API has an expressive query language that allows users to flexibly filter and group timeseries data. Example queries are provided as a template. Additional examples can be found within the Cloud Console which also uses the Confluent Cloud Metrics API.

Query for bytes sent to consumers per minute grouped by topic¶

Create a file named sent_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.server/sent_bytes"
    }
  ],
  "filter": {
    "field": "resource.kafka.id",
    "op": "EQ",
    "value": "lkc-XXXXX"
  },
  "granularity": "PT1M",
  "group_by": [
    "metric.topic"
  ],
  "intervals": [
    "2019-12-19T11:00:00-05:00/2019-12-19T11:05:00-05:00"
  ],
  "limit": 25
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < sent_bytes_query.json

Your output should resemble:

{
  "data": [
    {
      "timestamp": "2019-12-19T16:01:00Z",
      "metric.topic": "test-topic",
      "value": 0.0
    },
    {
      "timestamp": "2019-12-19T16:02:00Z",
      "metric.topic": "test-topic",
      "value": 157.0
    },
    {
      "timestamp": "2019-12-19T16:03:00Z",
      "metric.topic": "test-topic",
      "value": 371.0
    },
    {
      "timestamp": "2019-12-19T16:04:00Z",
      "metric.topic": "test-topic",
      "value": 0.0
    }
  ]
}

Note

If you haven’t produced data during the time window, the dataset is empty for a given topic.

Query for bytes received from producers per minute grouped by topic¶

Create a file named received_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.server/received_bytes"
    }
  ],
  "filter": {
    "field": "resource.kafka.id",
    "op": "EQ",
    "value": "lkc-XXXXX"
  },
  "granularity": "PT1M",
  "group_by": [
    "metric.topic"
  ],
  "intervals": [
    "2019-12-19T11:00:00-05:00/2019-12-19T11:05:00-05:00"
  ],
  "limit": 25
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < received_bytes_query.json

Your output should resemble:

{
  "data": [
    {
      "timestamp": "2019-12-19T16:00:00Z",
      "metric.topic": "test-topic",
      "value": 72.0
    },
    {
      "timestamp": "2019-12-19T16:01:00Z",
      "metric.topic": "test-topic",
      "value": 139.0
    },
    {
      "timestamp": "2019-12-19T16:02:00Z",
      "metric.topic": "test-topic",
      "value": 232.0
    },
    {
      "timestamp": "2019-12-19T16:03:00Z",
      "metric.topic": "test-topic",
      "value": 0.0
    },
    {
      "timestamp": "2019-12-19T16:04:00Z",
      "metric.topic": "test-topic",
      "value": 0.0
    }
  ]
}

Note

If you haven’t produced data during the time window, the dataset is empty for a given topic.

Query for max retained bytes per hour over 2 hours for a cluster `lkc-XXXXX`¶

Create a file named cluster_retained_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs:

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.server/retained_bytes"
    }
  ],
  "filter": {
    "field": "resource.kafka.id",
    "op": "EQ",
    "value": "lkc-XXXXX"
  },
  "granularity": "PT1H",
  "intervals": [
    "2019-12-19T11:00:00-05:00/P0Y0M0DT2H0M0S"
  ],
  "limit": 5
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < cluster_retained_bytes_query.json

Your output should resemble:

{
  "data": [
    {
      "timestamp": "2019-12-19T16:00:00Z",
      "value": 507350.0
    },
    {
      "timestamp": "2019-12-19T17:00:00Z",
      "value": 507350.0
    }
  ]
}

Query for average consumer lag over the last hour grouped by topic and consumer group¶

Create a file named consumer_lag_max_hour.json using the following template. Be sure to change lkc-XXXXX and note the interval is for the last hour with a 1-minute interval.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.server/consumer_lag_offsets"
    }
  ],
  "filter": {
    "field": "resource.kafka.id",
    "op": "EQ",
    "value": "lkc-XXXXX"
  },
  "granularity": "PT1H",
  "group_by": [
    "metric.consumer_group_id",
    "metric.topic"
  ],
  "intervals": [
    "PT1M/now"
  ],
  "limit": 25
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < consumer_lag_max_hour.json

Your output should resemble:

{
  "data": [
    {
      "metric.consumer_group_id": "group_1",
      "metric.topic": "test_topic_1",
      "timestamp": "2022-03-23T21:00:00Z",
      "value": 0.0
    },
    {
      "metric.consumer_group_id": "group_2",
      "metric.topic": "test_topic_2",
      "timestamp": "2022-03-23T21:00:00Z",
      "value": 6.0
    }
  ]
}

Query for the number of streaming units used per hour for ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_streaming_unit_count.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/streaming_unit_count"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-XXXXX"
  },
  "granularity": "PT1H",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ],
  "group_by": [
    "resource.ksql.id"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_streaming_unit_count.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 4.0
    }
  ]
}

Query for the max % of storage used over all CSUs for a ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_storage_utilization.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/storage_utilization"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_storage_utilization.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 0.85
    }
  ]
}

Query for the bytes of ksqlDB storage used by a query on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_query_storage.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/task_stored_bytes"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
    "metric.query_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to changed API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_storage.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 7688174488
    }
  ]
}

Query for the bytes of storage used by a task on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_task_storage.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/task_stored_bytes"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
    "metric.query_id",
    "metric.task_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_task_storage.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "metric.task_id": "1_1",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 1079295760
    }
  ]
}

Query for the query saturation on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_query_saturation.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/query_saturation"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_saturation.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 0.85
    }
  ]
}

Query for the total bytes consumed by ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_bytes_consumed.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/consumed_total_bytes"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_bytes_consumed.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 1024
    }
  ]
}

Query for the total bytes produced by ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_bytes_produced.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/produced_total_bytes"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_bytes_produced.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 1024
    }
  ]
}

Query for the total topic offsets processed by task on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_offsets_processed.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/offsets_processed_total"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id",
      "metric.tasK_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offsets_processed.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.task_id": "1_1",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 123
    }
  ]
}

Query for the total topic offsets processed by all tasks of query on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_offsets_processed.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/offsets_processed_total"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offsets_processed.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 123
    }
  ]
}

Query for the current committed offset lag by task on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_offset_lag.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/committed_offset_lag"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id",
      "metric.tasK_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offset_lag.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.task_id": "1_1",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 456
    }
  ]
}

Query for the current total committed offset lag for all tasks in query on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_offset_lag.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/committed_offset_lag"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offset_lag.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 456
    }
  ]
}

Query for the total number of processing errors by query on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_processing_errors.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/processing_errors_total"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_processing_errors.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 16
    }
  ]
}

Query for the total number of restarts due to failure by query on ksqlDB cluster `lksqlc-XXXXX`¶

Create a file named ksql_query_restarts.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.ksql/query_restarts"
    }
  ],
  "filter": {
    "field": "resource.ksql.id",
    "op": "EQ",
    "value": "lksqlc-xxxxx"
  },
  "granularity": "PT1M",
  "group_by": [
      "metric.query_id"
  ],
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_restarts.json

Your output should resemble:

{
  "data": [
    {
      "resource.ksql.id": "lksqlc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "metric.query_id": "CTAS_PAGEVIEWS_2",
      "value": 3
    }
  ]
}

Query for the number of schemas in the Schema Registry cluster `lsrc-XXXXX`¶

Create a file named schema_count.json using the following template. Be sure to change lsrc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.schema_registry/schema_count"
    }
  ],
  "filter": {
    "field": "resource.schema_registry.id",
    "op": "EQ",
    "value": "lsrc-XXXXX"
  },
  "granularity": "PT1M",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T10:01:00Z"
  ],
  "group_by": [
    "resource.schema_registry.id"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < schema_count.json

Your output should resemble:

{
  "data": [
    {
      "resource.schema_registry.id": "lsrc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 1.0
    }
  ]
}

Query for the hourly number of records received by a sink Connector `lcc-XXXXX`¶

Create a file named sink_connector_records.json using the following template. Be sure to change lcc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.connect/received_records"
    }
  ],
  "filter": {
    "field": "resource.connector.id",
    "op": "EQ",
    "value": "lcc-XXXXX"
  },
  "granularity": "PT1H",
  "intervals": [
    "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
  ],
  "group_by": [
    "resource.connector.id"
  ]
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < sink_connector_records.json

Your output should resemble:

{
  "data": [
    {
      "resource.connector.id": "lcc-XXXXX",
      "timestamp": "2021-02-24T10:00:00Z",
      "value": 26455991.0
    }
  ]
}

Query for metrics for a specific Principal ID¶

The metric.principal_id label can be used to filter metrics by specific users or service accounts. Metrics such as io.confluent.kafka.server/active_connection_count and io.confluent.kafka.server/request_count support filtering by the metric.principal_id label. To see all metrics that currently support this label see the API Reference.

Create a file named principal_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

{
  "aggregations": [
    {
      "metric": "io.confluent.kafka.server/active_connection_count"
    }
  ],
  "filter": {
    "field": "resource.kafka.id",
    "op": "EQ",
    "value": "lkc-XXXXX"
  },
  "granularity": "PT1H",
  "group_by": [
    "metric.principal_id"
  ],
  "intervals": [
    "2022-01-01T00:00:00Z/PT1H"
  ],
  "limit": 5
}

Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < principal_query.json

Your output should resemble:

{
  "data": [
    {
      "metric.principal_id": "sa-abc123",
      "timestamp": "2022-01-01T00:00:00Z",
      "value": 430.99999999997
    },
    {
      "metric.principal_id": "u-def456",
      "timestamp": "2022-01-01T00:00:00Z",
      "value": 427.93333333332
    },
    {
      "metric.principal_id": "u-abc123",
      "timestamp": "2022-01-01T00:00:00Z",
      "value": 333.19999999997
    }
  ],
  "meta": {
    "pagination": {
      "next_page_token": "eyJ2ZXJzaW9uIjoiMSIsInJlcXVlc3RI",
      "page_size": 5
    }
  }

Note

Topics without reported metric values during the specified interval aren’t returned.

FAQ¶

Can the Metrics API be used to reconcile my bill?¶

No, the Metrics API is intended to provide information for the purposes of monitoring, troubleshooting, and capacity planning. It is not intended as an audit system for reconciling bills as the metrics do not include request overhead for the Kafka protocol at this time. For more details, see the billing documentation.

Why am I seeing empty data sets for topics that exist on queries other than for `retained_bytes`?¶

If there are only values of 0.0 in the time range queried, than the API will return an empty set. When there is non-zero data within the time range, time slices with values of 0.0 are returned.

Why didn’t `retained_bytes` decrease after I changed the retention policy for my topic?¶

The value of retained_bytes is calculated as the maximum over the interval for each data point returned. If data has been deleted during the current interval, you will not see the effect until the next time range window begins. For example, if you produced 4GB of data per day over the last 30 days and queried for retained_bytes over the last 3 days with a 1 day interval, the query would return values of 112GB, 116GB, 120GB as a time series. If you then deleted all data in the topic and stopped producing data, the query would return the same values until the next day. When queried at the start of the next day, the same query would return 116GB, 120GB, 0GB.

What are the supported granularity levels?¶

Data is stored at a granularity of one minute. However, the allowed granularity for a query is restricted by the size of the query’s interval. Please see the API Reference for the currently supported granularity levels and query restrictions.

How do I monitor consumer lag?¶

Query for the average consumer lag by using the Metrics API.
Also, there are multiple other ways to monitor Consumer lag including the client metrics, UI, CLI, and Admin API. These methods are all available when using Confluent Cloud.

What is the retention time of metrics in the Metrics API?¶

Metrics are retained for seven days.

How do I know if a given metric is in preview or generally available (GA)?¶

We are always looking to add new metrics, but when we add a new metric, we need to take some time to stabilize how we expose it, to ensure that it’s suitable for most use cases. Each metric’s lifecycle stage (preview, generally available, etc.) is included in the response from the /descriptors/metrics endpoint. While a metric is in preview we may make breaking changes to its labels without an API version change, as we iterate to provide the best possible experience.

What should I do if a query to Metrics API returns a timeout response (HTTP error code 504)?¶

If queries are exceeding the timeout (maximum query time is 60s) you may consider one or more of the following approaches:

Reduce the time interval.
Reduce the granularity of data returned.
Break up the query on the client side to return fewer data points. For example, you can query for specific topics instead of all topics at once.

These approaches are especially important to when querying for partition-level data over days-long intervals.

Why are my Confluent Cloud metrics displaying only 1hr/6hrs/24hrs worth of data?¶

This is a known limitation that occurs in some clusters with a partition count of more than 2,000. We are working on the issue, but there is no fix at this time.

What should I do if a query returns a 5xx response code?¶

We recommended retrying these type of responses. Usually, this is an indication of a transient server-side issue. You should design your client implementations for querying the Metrics API to be resilient to this type of response for minutes-long periods.

Suggested Resources¶

Podcast: Adopting OpenTelemetry in Confluent and Beyond ft. Xavier Léauté
Podcast: Multi-Cloud Monitoring and Observability with the Metrics API ft. Dustin Cote
To learn how to architect, monitor, and optimize your Kafka applications on Confluent Cloud, refer to Develop Client Applications on Confluent Cloud.

Confluent Cloud Metrics¶

Metrics Quick Start¶

Add the MetricsViewer role to a new service account in the Confluent Cloud Console¶

Add the MetricsViewer role to a service account using the CLI¶

Integrate with third-party monitoring¶

Datadog¶

Dynatrace¶

Grafana Cloud¶

Prometheus¶

New Relic OpenTelemetry¶

Discovery using the Metrics API¶

Discover available resources¶

Discover available metrics¶

Example Queries¶

Query for bytes sent to consumers per minute grouped by topic¶

Query for bytes received from producers per minute grouped by topic¶

Query for max retained bytes per hour over 2 hours for a cluster lkc-XXXXX¶

Query for average consumer lag over the last hour grouped by topic and consumer group¶

Query for the number of streaming units used per hour for ksqlDB cluster lksqlc-XXXXX¶

Query for the max % of storage used over all CSUs for a ksqlDB cluster lksqlc-XXXXX¶

Query for the bytes of ksqlDB storage used by a query on ksqlDB cluster lksqlc-XXXXX¶

Query for the bytes of storage used by a task on ksqlDB cluster lksqlc-XXXXX¶

Query for the query saturation on ksqlDB cluster lksqlc-XXXXX¶

Query for the total bytes consumed by ksqlDB cluster lksqlc-XXXXX¶

Query for the total bytes produced by ksqlDB cluster lksqlc-XXXXX¶

Query for the total topic offsets processed by task on ksqlDB cluster lksqlc-XXXXX¶

Query for the total topic offsets processed by all tasks of query on ksqlDB cluster lksqlc-XXXXX¶

Query for the current committed offset lag by task on ksqlDB cluster lksqlc-XXXXX¶

Query for the current total committed offset lag for all tasks in query on ksqlDB cluster lksqlc-XXXXX¶

Query for the total number of processing errors by query on ksqlDB cluster lksqlc-XXXXX¶

Query for the total number of restarts due to failure by query on ksqlDB cluster lksqlc-XXXXX¶

Query for the number of schemas in the Schema Registry cluster lsrc-XXXXX¶

Query for the hourly number of records received by a sink Connector lcc-XXXXX¶

Query for metrics for a specific Principal ID¶

FAQ¶

Can the Metrics API be used to reconcile my bill?¶

Why am I seeing empty data sets for topics that exist on queries other than for retained_bytes?¶

Why didn’t retained_bytes decrease after I changed the retention policy for my topic?¶

What are the supported granularity levels?¶

How do I monitor consumer lag?¶

What is the retention time of metrics in the Metrics API?¶

How do I know if a given metric is in preview or generally available (GA)?¶

What should I do if a query to Metrics API returns a timeout response (HTTP error code 504)?¶

Why are my Confluent Cloud metrics displaying only 1hr/6hrs/24hrs worth of data?¶

What should I do if a query returns a 5xx response code?¶

Suggested Resources¶

Query for max retained bytes per hour over 2 hours for a cluster `lkc-XXXXX`¶

Query for the number of streaming units used per hour for ksqlDB cluster `lksqlc-XXXXX`¶

Query for the max % of storage used over all CSUs for a ksqlDB cluster `lksqlc-XXXXX`¶

Query for the bytes of ksqlDB storage used by a query on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the bytes of storage used by a task on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the query saturation on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total bytes consumed by ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total bytes produced by ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total topic offsets processed by task on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total topic offsets processed by all tasks of query on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the current committed offset lag by task on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the current total committed offset lag for all tasks in query on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total number of processing errors by query on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the total number of restarts due to failure by query on ksqlDB cluster `lksqlc-XXXXX`¶

Query for the number of schemas in the Schema Registry cluster `lsrc-XXXXX`¶

Query for the hourly number of records received by a sink Connector `lcc-XXXXX`¶

Why am I seeing empty data sets for topics that exist on queries other than for `retained_bytes`?¶

Why didn’t `retained_bytes` decrease after I changed the retention policy for my topic?¶