# Run Example Queries for the Confluent Cloud Metrics API The Confluent Cloud Metrics API has an expressive query language that lets you flexibly filter and group time-series data. The following example queries serve as a template. Find more examples in the Cloud Console, which also uses the Confluent Cloud [Metrics API](https://api.telemetry.confluent.cloud/docs). To learn how to identify clients sending deprecated requests to a cluster, see [Client Deprecation in Confluent Cloud](../client-apps/deprecate-how-to.md#cloud-id-deprecated-clients). Timestamps in metrics queries use Coordinated Universal Time (UTC). Use either UTC or an offset appropriate for your location. In cases where a `principal_id` is returned, the response also contains a `principal_name` field with the human-readable name for the `principal_id`. You should use the `principal_name` field for display purposes, and the `principal_id` field for any programmatic use cases such as alerting or monitoring. For an example, see [Query for metrics for a specific principal ID](#query-metrics-by-principal). > ##### On this page > > * [Monitor Kafka clusters and clients](#monitor-ak-clusters-and-clients) > * [Monitor Schema Registry](#monitor-sr) > * [Monitor connectors](#monitor-connectors) > * [Monitor Flink](#monitor-af) > * [Monitor ksqlDB](#monitor-ksqldb) ## Monitor Kafka clusters and clients Monitoring Apache Kafka® clusters and clients shows you how much data is flowing through a cluster, whether consumer groups are keeping up, and which principals are consuming cluster resources. The following examples query metrics for Kafka clusters, topics, and clients. ### Query for bytes produced to the cluster per minute grouped by topic This query measures bytes produced (ingress). If you want to query bytes consumed (egress), see [Query for bytes consumed from the cluster per minute grouped by topic](#metrics-api-bytes-sent). If you are using Cluster Linking, the `received_bytes` does not include the mirror-in bytes to the cluster. You can use the `cluster_link_destination_response_bytes` metrics to query the mirror-in bytes instead. 1. 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. ```json { "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 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < received_bytes_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @received_bytes_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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 } ] } ``` ### Query for bytes consumed from the cluster per minute grouped by topic This query measures bytes consumed (egress). If you want to query bytes produced (ingress), see [Query for bytes produced to the cluster per minute grouped by topic](#metrics-api-bytes-received). If you are using Cluster Linking, the `sent_bytes` metric also includes the mirror-out bytes from the cluster. For details about `sent_bytes` and `received_bytes` with Cluster Linking, see [Cluster Linking Performance Limits](https://docs.confluent.io/cloud/current/multi-cloud/cluster-linking/index.html#performance-limits). 1. 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. ```json { "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 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < sent_bytes_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @sent_bytes_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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 } ] } ``` At the topic scope, these metrics emit data points only when there is active production or consumption during the requested time window. If there is no activity for a topic during the window, the dataset is empty for that topic. ### Query for max retained bytes for a cluster `lkc-XXXXX` This query measures the maximum number of bytes retained by a cluster, sampled hourly over a two-hour window. 1. 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: ```json { "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 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < cluster_retained_bytes_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @cluster_retained_bytes_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2019-12-19T16:00:00Z", "value": 507350.0 }, { "timestamp": "2019-12-19T17:00:00Z", "value": 507350.0 } ] } ``` ### Query for average consumer lag by topic and consumer group This query measures average consumer lag over the last hour, grouped by topic and consumer group, so you can identify which consumers are falling behind. 1. 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 one-minute granularity. ```json { "aggregations": [ { "metric": "io.confluent.kafka.server/consumer_lag_offsets" } ], "filter": { "field": "resource.kafka.id", "op": "EQ", "value": "lkc-XXXXX" }, "granularity": "PT1M", "group_by": [ "metric.consumer_group_id", "metric.topic" ], "intervals": [ "PT1H/now" ], "limit": 25 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < consumer_lag_max_hour.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @consumer_lag_max_hour.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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 } ] } ``` ## Monitor Schema Registry The Metrics API exposes Schema Registry metrics through the `schema_registry` resource type, identified by `resource.schema_registry.id` (for example, `lsrc-XXXXX`). The following metrics are available, each prefixed with `io.confluent.kafka.schema_registry/`: | Metric | Description | |---------------------------------|----------------------------------------------------------------------| | `schema_count` | Number of registered schemas. | | `schema_operations_count` | Delta count of schema create, delete, and read operations. | | `request_count` | Delta count of requests received by the Schema Registry server. | | `exporter_tasks` | Number of schema exporters by status (RUNNING, STARTING, or PAUSED). | | `exporter_starting_progress` | Progress of a schema exporter in the starting phase, from 0 to 100. | | `schema_transfer_success_total` | Count of schemas successfully registered by each exporter. | | `num_deks` | Number of data encryption keys. | | `num_keks` | Number of key encryption keys. | | `num_keks_shared` | Number of key encryption keys shared with Confluent. | For a worked example, see [Query for the number of schemas in the Schema Registry cluster lsrc-XXXXX](#sr-query-total-schemas-in-cluster). ### Query for the number of schemas in the Schema Registry cluster `lsrc-XXXXX` This query measures the number of registered schemas in a Schema Registry cluster at a given time. 1. 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. ```json { "aggregations": [ { "time_agg": "MAX", "agg": "SUM", "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < schema_count.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @schema_count.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.schema_registry.id": "lsrc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "value": 1.0 } ] } ``` ## Monitor connectors Monitoring connectors shows you how much data your connectors are moving and whether they have the system resources they need. The following examples query metrics for connectors. ### Query for hourly records received by a sink connector `lcc-XXXXX` This query measures the number of records a sink connector receives per hour. 1. Create a file named `sink_connector_record_number.json` using the following template. Be sure to change `lcc-XXXXX` and the timestamp values to match your needs. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your Confluent Cloud cluster credentials (`--resource cloud` credentials). ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < sink_connector_record_number.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @sink_connector_record_number.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.connector.id": "lcc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "value": 26455991.0 } ] } ``` ### Query for free memory on a custom connector `clcc-XXXXX` This query measures the total free memory available to a custom connector, which can help you determine whether the connector needs more resources. 1. Create a file named `custom_connector_free_memory.json` using the following template. Be sure to change `clcc-XXXXX` and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.system/memory_free_bytes" } ], "filter": { "field": "resource.custom_connector.id", "op": "EQ", "value": "clcc-XXXXX" }, "granularity": "PT1H", "intervals": [ "2023-05-09T10:00:00Z/2023-05-09T15:00:00Z" ], "group_by": [ "resource.custom_connector.id" ] } ``` 2. Submit the query as a `POST` using the following command. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' --auth ':' < custom_connector_free_memory.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' -u ':' -d @custom_connector_free_memory.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.custom_connector.id": "clcc-XXXXXX", "timestamp": "2023-05-09T10:00:00Z", "value": 125229329.06666666 }, { "resource.custom_connector.id": "clcc-XXXXXX", "timestamp": "2023-05-09T11:00:00Z", "value": 125193966.93333334 }, { "resource.custom_connector.id": "clcc-XXXXXX", "timestamp": "2023-05-09T12:00:00Z", "value": 125140241.06666666 }, { "resource.custom_connector.id": "clcc-XXXXXX", "timestamp": "2023-05-09T13:00:00Z", "value": 125099622.4 }, { "resource.custom_connector.id": "clcc-XXXXXX", "timestamp": "2023-05-09T14:00:00Z", "value": 124849493.33333333 } ] } ``` For Cloud Console metrics for custom connectors, see [View metrics](../connectors/bring-your-connector/custom-connector-manage.md#cc-byoc-view-metrics). ### Query for percent CPU used by a custom connector `clcc-XXXXX` This query measures the percentage of CPU used by a custom connector, which can help you determine whether the connector needs more resources. 1. Create a file named `custom_connector_percent_cpu.json` using the following template. Be sure to change `clcc-XXXXX` and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.system/cpu_load_percent" } ], "filter": { "field": "resource.custom_connector.id", "op": "EQ", "value": "clcc-XXXXX" }, "granularity": "PT1H", "intervals": [ "2023-05-09T10:00:00Z/2023-05-09T15:00:00Z" ], "group_by": [ "resource.custom_connector.id" ] } ``` 2. Submit the query as a `POST` using the following command. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' --auth ':' < custom_connector_percent_cpu.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' -u ':' -d @custom_connector_percent_cpu.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.custom_connector.id": "clcc-XXXXX", "timestamp": "2023-05-09T10:00:00Z", "value": 0.021009808092643977 }, { "resource.custom_connector.id": "clcc-XXXXX", "timestamp": "2023-05-09T11:00:00Z", "value": 0.01990721858932965 }, { "resource.custom_connector.id": "clcc-XXXXX", "timestamp": "2023-05-09T12:00:00Z", "value": 0.020799848444189233 }, { "resource.custom_connector.id": "clcc-XXXXX", "timestamp": "2023-05-09T13:00:00Z", "value": 0.019948515028905416 }, { "resource.custom_connector.id": "clcc-XXXXX", "timestamp": "2023-05-09T14:00:00Z", "value": 0.020734587261390117 } ] } ``` For Cloud Console metrics for custom connectors, see [View metrics](../connectors/bring-your-connector/custom-connector-manage.md#cc-byoc-view-metrics). ### Query for metrics for a specific principal ID You can use the `metric.principal_id` label 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. This is particularly useful for monitoring and alerting on the activity of specific principals in your cluster. Each active `metric.principal_id` also provides a `metric.principal_name` label that contains the human-readable name of the principal, such as the username or service account name, if a name has been defined. To see all metrics that currently support the `metric.principal_id` label, see the [API Reference](https://api.telemetry.confluent.cloud/docs/descriptors/datasets/cloud). 1. 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. ```json { "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 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < principal_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @principal_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "metric.principal_id": "sa-abc123", "metric.principal_name": "Prod Service Account", "timestamp": "2022-01-01T00:00:00Z", "value": 430.99999999997 }, { "metric.principal_id": "u-def456", "metric.principal_name": "Mary Smith", "timestamp": "2022-01-01T00:00:00Z", "value": 427.93333333332 }, { "metric.principal_id": "u-abc123", "metric.principal_name": "Tom Nguyen", "timestamp": "2022-01-01T00:00:00Z", "value": 333.19999999997 } ], "meta": { "pagination": { "next_page_token": "eyJ2ZXJzaW9uIjoiMSIsInJlcXVlc3RI", "page_size": 5 } } } ``` Topics without reported metric values during the specified interval aren’t returned. ### Query for connection accept count to monitor eCKU scaling This query measures new connections accepted by the cluster. Each sample is the number of connections accepted since the previous data point, sampled every 60 seconds. This metric corresponds to the connection attempts dimension used to determine [eCKU](../billing/billing-dimensions.md#e-cku-definition) scaling for elastic cluster types. eCKU is a unit of capacity that scales elastically, so this metric helps you understand and monitor your eCKU usage. 1. Create a file named `connection_accept_count_query.json` using the following template. Be sure to change `lkc-xyz` and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.kafka.server/connection_accept_count" } ], "filter": { "field": "resource.kafka.id", "op": "EQ", "value": "lkc-xyz" }, "granularity": "PT1M", "intervals": [ "PT1H/now" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < connection_accept_count_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @connection_accept_count_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2026-04-14T20:30:00Z", "value": 12.0 }, { "timestamp": "2026-04-14T20:31:00Z", "value": 10.0 }, { "timestamp": "2026-04-14T20:32:00Z", "value": 13.0 } ] } ``` This metric provides visibility into all connection attempts, including failed ones, which are included in eCKU cost calculations. ### Query for throttled clients on a cluster This query returns the average throttle time applied to principals on a cluster, grouped by principal, violated limit type, and reason. For more information about this metric, see [Throttled clients metric](metrics-api.md#throttled-clients-metric). 1. Create a file named `throttled_clients_query.json` using the following template. Be sure to change `lkc-XXXXX` and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.kafka.server/client_limit_milliseconds" } ], "filter": { "field": "resource.kafka.id", "op": "EQ", "value": "lkc-XXXXX" }, "granularity": "PT1M", "group_by": [ "metric.principal_id", "metric.violated_limit", "metric.reason" ], "intervals": [ "PT1H/now" ], "limit": 25 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < throttled_clients_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @throttled_clients_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2026-05-04T14:00:00Z", "metric.principal_id": "u-23agh7", "metric.violated_limit": "produce_throughput_quota", "metric.reason": "cluster_quota_violation", "value": 150.0 }, { "timestamp": "2026-05-04T14:00:00Z", "metric.principal_id": "sa-abc123", "metric.violated_limit": "fetch_throughput_quota", "metric.reason": "principal_quota_violation", "value": 75.5 } ] } ``` ## Monitor Flink Monitoring Flink compute pools and SQL statements shows you how much data your statements are processing, whether the statements are keeping up with incoming data, and how much compute capacity they’re using. The following examples query metrics for Flink compute pools and SQL statements. ### Query for the total number of records a Flink SQL statement has received This metric represents the total number of records a Flink SQL statement has received as input. 1. Create a file named `num_records_in.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), statement name (`XXXXXXXX-XXXX-XXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/num_records_in" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.flink_statement.name", "op": "EQ", "value": "XXXXXXXX-XXXX-XXXX" }, { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" } ] }, "granularity": "PT1M", "intervals": [ "2023-10-23T16:30:00/2023-10-23T16:35:00" ], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < num_records_in.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @num_records_in.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-23T16:30:00Z", "value": 115.0 }, { "timestamp": "2023-10-23T16:31:00Z", "value": 116.0 }, { "timestamp": "2023-10-23T16:32:00Z", "value": 116.0 }, { "timestamp": "2023-10-23T16:33:00Z", "value": 131.0 }, { "timestamp": "2023-10-23T16:34:00Z", "value": 127.0 } ] } ``` ### Query for the total number of records a Flink SQL statement has emitted This metric represents the total number of records a Flink SQL statement has emitted as output. 1. Create a file named `num_records_out.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), statement name (`XXXXXXXX-XXXX-XXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/num_records_out" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.flink_statement.name", "op": "EQ", "value": "XXXXXXXX-XXXX-XXXX" }, { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" } ] }, "granularity": "PT1M", "intervals": [ "2023-10-23T16:30:00/2023-10-23T16:35:00" ], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < num_records_out.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @num_records_out.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-23T16:30:00Z", "value": 115.0 }, { "timestamp": "2023-10-23T16:31:00Z", "value": 116.0 }, { "timestamp": "2023-10-23T16:32:00Z", "value": 116.0 }, { "timestamp": "2023-10-23T16:33:00Z", "value": 131.0 }, { "timestamp": "2023-10-23T16:34:00Z", "value": 127.0 } ] } ``` ### Query for the backlog of a Flink SQL statement This metric represents the total number of available records after the consumer offset in a Kafka partition for a Flink SQL statement, across all operators. 1. Create a file named `pending_records.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), statement name (`XXXXXXXX-XXXX-XXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/pending_records" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.flink_statement.name", "op": "EQ", "value": "XXXXXXXX-XXXX-XXXX" }, { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" } ] }, "granularity": "PT1M", "intervals": [ "2023-10-23T16:30:00/2023-10-23T16:35:00" ], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < pending_records.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @pending_records.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-23T16:30:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:31:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:32:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:33:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:34:00Z", "value": 0.0 } ] } ``` A non-zero value indicates some backlog associated with the Flink statement. ### Query for the highest record lateness of a Flink SQL statement This metric represents the highest observed lateness, in milliseconds, across all records that a Flink SQL statement processed in the last minute. A record counts as late if its event-time timestamp is at or before the current [watermark](../flink/concepts/timely-stream-processing.md#flink-sql-event-time-lateness). This is the closest available metric to a per-record processing latency, in milliseconds. 1. Create a file named `max_input_lateness.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), statement name (`XXXXXXXX-XXXX-XXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/max_input_lateness_milliseconds" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.flink_statement.name", "op": "EQ", "value": "XXXXXXXX-XXXX-XXXX" }, { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" } ] }, "granularity": "PT1M", "intervals": [ "2023-10-23T16:30:00/2023-10-23T16:35:00" ], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < max_input_lateness.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @max_input_lateness.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-23T16:30:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:31:00Z", "value": 120.0 }, { "timestamp": "2023-10-23T16:32:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:33:00Z", "value": 0.0 }, { "timestamp": "2023-10-23T16:34:00Z", "value": 0.0 } ] } ``` A non-zero value indicates that at least one record arrived that many milliseconds after the statement’s watermark had already advanced past its event-time timestamp. To also count how many records were affected, query `io.confluent.flink/num_late_records_in` the same way. ### Query for records received by all Flink SQL statements in a Flink compute pool This metric represents the total number of records received across all Flink SQL statements running in a Flink compute pool. 1. Create a file named `pool_num_records_in.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/num_records_in" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < pool_num_records_in.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @pool_num_records_in.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-25T16:30:00Z", "value": 236.0 }, { "timestamp": "2023-10-25T16:31:00Z", "value": 228.0 }, { "timestamp": "2023-10-25T16:32:00Z", "value": 240.0 }, { "timestamp": "2023-10-25T16:33:00Z", "value": 230.0 }, { "timestamp": "2023-10-25T16:34:00Z", "value": 252.0 } ] } ``` ### Query for records emitted by all Flink SQL statements in a Flink compute pool This metric represents the total number of records emitted across all Flink SQL statements running in a Flink compute pool. 1. Create a file named `pool_num_records_out.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/num_records_out" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < pool_num_records_out.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @pool_num_records_out.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-25T16:30:00Z", "value": 236.0 }, { "timestamp": "2023-10-25T16:31:00Z", "value": 228.0 }, { "timestamp": "2023-10-25T16:32:00Z", "value": 240.0 }, { "timestamp": "2023-10-25T16:33:00Z", "value": 230.0 }, { "timestamp": "2023-10-25T16:34:00Z", "value": 252.0 } ] } ``` ### Query for the backlog of all Flink SQL statements in a Flink compute pool This metric represents the total number of available records after the consumer offset in a Kafka partition for all Flink SQL statements using a Flink compute pool, across all operators. 1. Create a file named `pool_pending_records.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/pending_records" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < pool_pending_records.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @pool_pending_records.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2023-10-25T16:30:00Z", "value": 0.0 }, { "timestamp": "2023-10-25T16:31:00Z", "value": 0.0 }, { "timestamp": "2023-10-25T16:32:00Z", "value": 0.0 }, { "timestamp": "2023-10-25T16:33:00Z", "value": 0.0 }, { "timestamp": "2023-10-25T16:34:00Z", "value": 0.0 } ] } ``` A non-zero value indicates the combined backlog associated with the Flink statements using the Flink compute pool in the query. ### Query for the current number of CFUs in a Flink compute pool This metric represents the absolute number of Confluent Flink Units (CFUs), or the current usage, at a given moment in a Flink compute pool. It’s a point-in-time gauge, so it’s best suited to usage alerts, for example, comparing it against a compute pool’s CFU limit. It isn’t the metric Confluent uses to calculate your bill; for that, see [Query for the CFU-minutes consumed by a Flink compute pool](#metrics-api-cfu-minutes-consumed) below. For more information about CFUs, see [CFU billing](../flink/concepts/flink-billing.md#flink-sql-billing-cfu). 1. Create a file named `current_cfus.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/compute_pool_utilization/current_cfus" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2024-05-15T14:00:00/2024-05-15T14:05:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < current_cfus.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @current_cfus.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2024-05-15T14:00:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:01:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:02:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:03:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:04:00Z", "value": 3.0 } ] } ``` ### Query for the CFU-minutes consumed by a Flink compute pool This metric represents the number of Confluent Flink Units (CFUs) consumed by a Flink compute pool, in CFU-minutes, over each measurement interval. Unlike `current_cfus`, which is a snapshot of usage at a single moment, `cfu_minutes_consumed` is a counter that accumulates usage between samples. Summing its values over a time range gives the total CFU-minutes consumed in that range, which is the same unit that Confluent bills against. For more information, see [CFU billing](../flink/concepts/flink-billing.md#flink-sql-billing-cfu). 1. Create a file named `cfu_minutes_consumed.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/compute_pool_utilization/cfu_minutes_consumed" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2024-05-15T14:00:00/2024-05-15T14:05:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < cfu_minutes_consumed.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @cfu_minutes_consumed.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2024-05-15T14:00:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:01:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:02:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:03:00Z", "value": 3.0 }, { "timestamp": "2024-05-15T14:04:00Z", "value": 3.0 } ] } ``` Summing these five one-minute values gives the total CFU-minutes consumed over the five-minute interval: 15 CFU-minutes. ### Query for the maximum number of CFUs assigned to a Flink compute pool This metric represents the maximum number of CFUs assigned to a Flink compute pool. When Flink statements are running, the compute pool is autoscaled up to this maximum number of CFUs assigned to a Flink compute pool. 1. Create a file named `cfu_limit.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/compute_pool_utilization/cfu_limit" } ], "filter": { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" }, "granularity": "PT1M", "intervals": ["2024-05-15T14:00:00/2024-05-15T14:05:00"], "limit": 5 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < cfu_limit.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @cfu_limit.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2024-05-15T14:00:00Z", "value": 10.0 }, { "timestamp": "2024-05-15T14:01:00Z", "value": 10.0 }, { "timestamp": "2024-05-15T14:02:00Z", "value": 10.0 }, { "timestamp": "2024-05-15T14:03:00Z", "value": 10.0 }, { "timestamp": "2024-05-15T14:04:00Z", "value": 10.0 } ] } ``` ### Query for the state size of a Flink statement This metric represents the [state size](../flink/concepts/statements.md#flink-sql-statements-state-size-limits) of a Flink SQL statement. 1. Create a file named `state_size.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/operator/state_size_bytes", "agg": "SUM" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.compute_pool.id", "op": "EQ", "value": "" }, { "field": "resource.flink_statement.name", "op": "EQ", "value": "" } ] }, "granularity": "PT5M", "intervals": [ "2026-01-23T00:00:00Z/2026-01-23T01:00:00Z" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < state_size.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @state_size.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "metadata": { "name": "", "self": "https://flink...confluent.cloud/sql/v3/organizations//environments//statements/" }, "status": { "phase": "RUNNING", "stateLimitStatus": { "stateLimitState": "APPROACHING_SOFT_LIMIT", "message": "Statement state size is approaching the soft state size limit.", "lastUpdatedAt": "2026-01-23T09:15:00Z" }, "scalingStatus": { "scalingState": "POOL_EXHAUSTED" } } } ``` ### Query for the statement status for a given Flink SQL statement This metric represents the status of a Flink SQL statement. 1. Create a file named `statement_status.json` using the following template. Be sure to change the compute pool ID (`lfcp-XXXXXX`), and the timestamp values to match your needs. ```json { "aggregations": [ { "metric": "io.confluent.flink/statement_status" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.flink_statement.name", "op": "EQ", "value": "" }, { "field": "resource.compute_pool.id", "op": "EQ", "value": "lfcp-XXXXXX" } ] }, "granularity": "PT1M", "intervals": ["now-6h/now"], "group_by": [ "resource.flink_statement.uid", "metric.status" ], "limit": 1000 } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < statement_status.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @statement_status.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "timestamp": "2025-03-10T09:27:00Z", "value": 1.0, "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", "metric.status": "RUNNING" }, { "timestamp": "2025-03-10T09:32:00Z", "value": 1.0, "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", "metric.status": "RUNNING" }, { "timestamp": "2025-03-10T09:34:00Z", "value": 1.0, "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", "metric.status": "RUNNING" }, { "timestamp": "2025-03-10T09:36:00Z", "value": 1.0, "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", "metric.status": "RUNNING" } ] } ``` ## Monitor ksqlDB Monitoring ksqlDB clusters shows you how much storage and compute capacity your queries are using and whether they’re processing data without errors. The following examples query metrics for ksqlDB clusters. ### Query for hourly streaming units for ksqlDB cluster `lksqlc-XXXXX` This query measures the number of ksqlDB streaming units a cluster uses per hour, which you can use to monitor capacity usage. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_streaming_unit_count.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_streaming_unit_count.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "value": 4.0 } ] } ``` ### Query for max storage percent used by a ksqlDB cluster `lksqlc-XXXXX` This query measures the maximum percentage of storage used across all Confluent Streaming Units (CSUs) for a ksqlDB cluster. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_storage_utilization.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_storage_utilization.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "value": 0.85 } ] } ``` ### Query for storage bytes used by a query on ksqlDB cluster `lksqlc-XXXXX` This query measures the number of storage bytes used by a specific ksqlDB query, grouped by query ID. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_query_storage.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_query_storage.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "metric.query_id": "CTAS_PAGEVIEWS_2", "timestamp": "2021-02-24T10:00:00Z", "value": 7688174488 } ] } ``` ### Query for storage bytes used by a task on ksqlDB cluster `lksqlc-XXXXX` This query measures the number of storage bytes used by an individual task within a ksqlDB query. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_task_storage.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_task_storage.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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` This query measures query saturation for a ksqlDB cluster. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_query_saturation.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_query_saturation.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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` This query measures the total number of bytes a ksqlDB cluster has consumed. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_bytes_consumed.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_bytes_consumed.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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` This query measures the total number of bytes a ksqlDB cluster has produced. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_bytes_produced.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_bytes_produced.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "value": 1024 } ] } ``` ### Query for topic offsets processed by a task on ksqlDB cluster `lksqlc-XXXXX` This query measures the total number of topic offsets processed by an individual task within a ksqlDB query. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_offsets_processed.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_offsets_processed.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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 total offsets by all tasks of a query on ksqlDB cluster `lksqlc-XXXXX` This query measures the total number of topic offsets processed by all tasks of a ksqlDB query, combined. 1. Create a file named `ksql_offsets_processed_by_query.json` using the following template. Be sure to change `lksqlc-XXXXX` and the timestamp values to match your needs. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_offsets_processed_by_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_offsets_processed_by_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "metric.query_id": "CTAS_PAGEVIEWS_2", "value": 123 } ] } ``` ### Query for committed offset lag by task on ksqlDB cluster `lksqlc-XXXXX` This query measures the current committed offset lag for an individual task within a ksqlDB query. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_offset_lag.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_offset_lag.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "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 total committed offset lag for a query on ksqlDB cluster `lksqlc-XXXXX` This query measures the current total committed offset lag across all tasks of a ksqlDB query, combined. 1. Create a file named `ksql_offset_lag_by_query.json` using the following template. Be sure to change `lksqlc-XXXXX` and the timestamp values to match your needs. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_offset_lag_by_query.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_offset_lag_by_query.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "metric.query_id": "CTAS_PAGEVIEWS_2", "value": 456 } ] } ``` ### Query for processing errors by query on ksqlDB cluster `lksqlc-XXXXX` This query measures the total number of processing errors for a ksqlDB query. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_processing_errors.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_processing_errors.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "metric.query_id": "CTAS_PAGEVIEWS_2", "value": 16 } ] } ``` ### Query for restarts due to failure by query on ksqlDB cluster `lksqlc-XXXXX` This query measures the total number of times a ksqlDB query has restarted due to failure. 1. 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. ```json { "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" ] } ``` 2. Submit the query as a `POST` using the following command. Be sure to change `API_KEY` and `SECRET` to match your environments. ### HTTPie ```bash http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth ':' < ksql_query_restarts.json ``` ### cURL ```bash curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u ':' -d @ksql_query_restarts.json -H 'Content-Type: application/json' ``` Your output should resemble: ```json { "data": [ { "resource.ksql.id": "lksqlc-XXXXX", "timestamp": "2021-02-24T10:00:00Z", "metric.query_id": "CTAS_PAGEVIEWS_2", "value": 3 } ] } ```