--- title: WarpStream description: Monitor the health and performance of your WarpStream Agents breadcrumbs: Docs > Integrations > WarpStream --- > For the complete documentation index, see [llms.txt](https://docs.datadoghq.com/llms.txt). # WarpStream Supported OS Integration version1.0.0 {% callout %} # Important note for users on the following Datadog sites: us2.ddog-gov.com {% alert level="info" %} To find out if this integration is available in your organization, see your [Datadog Integrations](https://app.datadoghq.com/integrations) page or ask your organization administrator. To initiate an exception request to enable this integration for your organization, email [support@ddog-gov.com](mailto:support@ddog-gov.com). {% /alert %} {% /callout %} Datadog dashboardWarpStream architectureWarpStream console ## Overview{% #overview %} WarpStream is a data streaming platform compatible with Apache Kafka®, designed to run directly on object storage. This integration provides visibility into WarpStream agent performance by exposing key metrics, helping users monitor both health and performance. ## Setup{% #setup %} ### Installation{% #installation %} 1. [Install the Datadog Agent](https://docs.datadoghq.com/getting_started/agent.md). 1. Manually install the WarpStream integration. For environment-specific instructions, see [Use Community Integrations](https://docs.datadoghq.com/agent/guide/community-integrations-installation-with-docker-agent.md). ### Configuration{% #configuration %} Complete the following steps to set up the WarpStream integration. The WarpStream integration has two components: - The **Datadog Agent** component, which makes requests to an endpoint to check whether WarpStream can connect and is healthy. - The **WarpStream StatsD** component, which you configure to send metrics to the Datadog Agent. The WarpStream integration's [metrics](https://github.com/warpstreamlabs/integrations-extras/tree/epot/tier/warpstream#metrics) come from both the Agent and StatsD portions. #### Configure the WarpStream integration for the Datadog Agent{% #configure-the-warpstream-integration-for-the-datadog-agent %} Edit the `url` in `warpstream.d/conf.yaml`, located in the `conf.d/` folder at the root of your Agent's configuration directory. See the [sample `warpstream.d/conf.yaml`](https://github.com/DataDog/integrations-extras/blob/master/warpstream/datadog_checks/warpstream/data/conf.yaml.example) for all available configuration options. The `url` value must match your WarpStream Agent HTTP server (port 8080 by default). ##### Connect WarpStream Agent to DogStatsD{% #connect-warpstream-agent-to-dogstatsd %} Start the WarpStream Agent with the flag `-enableDatadogMetrics` or set the environment variable `WARPSTREAM_ENABLE_DATADOG_METRICS` to `true`. ##### Restart Datadog Agent and WarpStream{% #restart-datadog-agent-and-warpstream %} 1. [Restart the Agent](https://docs.datadoghq.com/agent/configuration/agent-commands.md#start-stop-and-restart-the-agent). 1. Restart the WarpStream Agent to start sending your WarpStream metrics to the Agent DogStatsD endpoint. ## Data Collected{% #data-collected %} ### Metrics{% #metrics %} | | | | | **warpstream.can_connect**(gauge) | 1 if can connect to WarpStream Agent, otherwise 0 | | **warpstream.agent_connection_throttle_count**(count) | counts the number of new connections opened to the WarpStream Agent throttled | | **warpstream.agent_deadscanner_optimistic_queue_delete_file_outcome**(count) | counts the number of files processed by a WarpStream Agent in its deadscanner optimistic queue, tagged by 'outcome' | | **warpstream.agent_deadscanner_optimistic_queue_submit_outcome**(count) | counts the number of requests to the WarpStream Agent deadscanner optimistic queue in its deadscanner optimistic deletion process, tagged by 'outcome' | | **warpstream.agent_deadscanner_outcomes**(count) | counts the number of files processed by a deadscanner job execution, tagged by 'outcome' | | **warpstream.agent_file_cache_client_fetch_local_or_remote_counter**(count) | counts the number of times a WarpStream Agent file cache fetch was served locally or remotely, tagged by `outcome` and `source` | | **warpstream.agent_file_cache_client_fetch_local_or_remote_num_bytes_distribution**(gauge) | distribution of the number of bytes fetched from a WarpStream Agent file cache, tagged by `source`*Shown as byte* | | **warpstream.agent_file_cache_client_get_stream_range_latency**(gauge) | latency of requests made to a WarpStream Agent file cache, tagged by `outcome`*Shown as second* | | **warpstream.agent_file_cache_client_get_stream_range_outcome**(count) | counts the number of requests made to a WarpStream Agent file cache, tagged by `outcome` | | **warpstream.agent_file_cache_client_get_stream_range_partitioned_requests_counter**(count) | counts the number of partitioned requests made to a WarpStream Agent file cache | | **warpstream.agent_file_cache_client_get_stream_range_partitioned_requests_distribution**(gauge) | distribution of the number of partitioned requests made to a WarpStream Agent file cache | | **warpstream.agent_file_cache_concurrent_fetches**(gauge) | distribution of the number of concurrent request made to a WarpStream Agent file cache | | **warpstream.agent_file_cache_context_remaining_timeout**(gauge) | timeout of a request to fetch to a WarpStream Agent file cache*Shown as second* | | **warpstream.agent_file_cache_get_range_num_chunks**(gauge) | distribution of the number of chunks fetched in a request to a WarpStream Agent file cache | | **warpstream.agent_file_cache_server_blob_store_fetcher_counter**(count) | counts the number of times a WarpStream Agent file cache service fetched from object storage tagged by `outcome` | | **warpstream.agent_file_cache_server_blob_store_fetcher_num_bytes_counter**(count) | counts the number of bytes fetched from object storage by a WarpStream Agent*Shown as byte* | | **warpstream.agent_file_cache_server_blob_store_fetcher_num_bytes_distribution**(gauge) | distribution of the number of bytes fetched from object storage by a WarpStream Agent*Shown as byte* | | **warpstream.agent_file_cache_server_fetch_size_counter**(count) | counts the number of bytes fetched from a WarpStream Agent file cache*Shown as byte* | | **warpstream.agent_file_cache_server_fetch_size_num_bytes_counter**(count) | counts the number of bytes fetched from a WarpStream Agent file cache*Shown as byte* | | **warpstream.agent_file_cache_server_get_range_copy_chunk**(count) | counts the number of chunks copied in a request to a WarpStream Agent file cache, tagged by `outcome` | | **warpstream.agent_file_cache_server_get_range_copy_chunk_num_bytes_copied**(count) | counts the number of bytes copied in a request to a WarpStream Agent file cache*Shown as byte* | | **warpstream.agent_file_cache_server_get_stream_range_latency**(gauge) | latency of requests served by a WarpStream Agent file cache, tagged by `outcome`*Shown as second* | | **warpstream.agent_file_cache_server_get_stream_range_num_bytes_count**(count) | counts the number of bytes served by a WarpStream Agent file cache*Shown as byte* | | **warpstream.agent_file_cache_server_get_stream_range_num_bytes_distribution**(gauge) | distribution of the number of bytes served by a WarpStream Agent file cache*Shown as byte* | | **warpstream.agent_file_cache_server_get_stream_range_num_ranges_distribution**(gauge) | distribution of the number of ranges served by a WarpStream Agent file cache | | **warpstream.agent_file_cache_server_get_stream_range_outcome**(count) | counts the number of requests served by a WarpStream Agent file cache, tagged by `outcome` | | **warpstream.agent_flushtuner_flush_size_rec**(gauge) | current value of the size autotuner used to flush data to the object storage when producing data*Shown as byte* | | **warpstream.agent_job_runner_fetched_jobs_num_jobs**(gauge) | distribution of the number of jobs fetched when polling jobs from WarpStream control plane | | **warpstream.agent_job_runner_fetched_jobs_outcome**(count) | counts the number of jobs poll request emitted by the WarpStream Agent job runner, tagged by 'outcome' | | **warpstream.agent_job_runner_no_jobs_found**(count) | counts the number of times the WarpStream Agent job runner polled for jobs but found none available | | **warpstream.agent_job_runner_num_running_jobs_gauge**(gauge) | number of jobs currently running on a given WarpStream Agent | | **warpstream.agent_job_runner_num_running_slots_distribution**(gauge) | distribution of the job slots currently running on a given WarpStream Agent | | **warpstream.agent_job_runner_num_running_slots_gauge**(gauge) | number of job slots currently occupied on a given WarpStream Agent | | **warpstream.agent_job_runner_slots_utilization_distribution**(gauge) | distribution of the ratio (between 0 and 1) of the job slots utilization | | **warpstream.agent_job_runner_slots_utilization_gauge**(gauge) | distribution of the ratio (between 0 and 1) of the job slots utilization | | **warpstream.agent_kafka_fetch_bytes_max**(gauge) | distribution of the max bytes of Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_bytes_min**(gauge) | distribution of the min bytes of Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_bytes_remaining**(gauge) | distribution of the number of bytes remaining when processing Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_compressed_bytes**(gauge) | distribution of the number of compressed bytes read when processing Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_compressed_bytes_counter**(count) | counts the number of compressed bytes read when processing Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_num_pointers_counter**(count) | counts the number of pointers read when processing Kafka fetch requests issued to a WarpStream Agent | | **warpstream.agent_kafka_fetch_num_pointers_distribution**(gauge) | distribution of the number of pointers read when processing Kafka fetch requests issued to a WarpStream Agent | | **warpstream.agent_kafka_fetch_partial_response_error_scenario_counter**(count) | counts the number of partial fetch requests processed by a WarpStream Agent | | **warpstream.agent_kafka_fetch_partial_response_error_scenario_num_bytes_distribution**(gauge) | distribution of the bytes in partial fetch requests processed by a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_partitions_count**(gauge) | number of partitions read when processing Kafka fetch requests issued to a WarpStream Agent | | **warpstream.agent_kafka_fetch_prefetch_concurrency_distribution**(gauge) | distribution of the concurrency used to prefetch data | | **warpstream.agent_kafka_fetch_request_latency**(gauge) | latency of Kafka fetch requests issued to a WarpStream Agent*Shown as second* | | **warpstream.agent_kafka_fetch_single_attempt_latency**(gauge) | latency of a single attempt to process Kafka fetch requests issued to a WarpStream Agent*Shown as second* | | **warpstream.agent_kafka_fetch_single_attempt_outcome**(count) | counts the number of attempts to process Kafka fetch requests issued to a WarpStream Agent, tagged by `outcome` | | **warpstream.agent_kafka_fetch_topics_count**(gauge) | distribution of the number of topics read when processing Kafka fetch requests issued to a WarpStream Agent | | **warpstream.agent_kafka_fetch_uncompressed_bytes**(gauge) | distribution of the number of uncompressed bytes read when processing Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_fetch_uncompressed_bytes_counter**(count) | counts the number of uncompressed bytes read when processing Kafka fetch requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_inflight_connections**(gauge) | gauge of the number of inflight Kafka connections opened to a WarpStream Agent | | **warpstream.agent_kafka_inflight_request_per_connection**(gauge) | distribution of the number of requests per Kafka connections open to a WarpStream Agent | | **warpstream.agent_kafka_inflight_request_per_connection_on_close**(gauge) | distribution of the number of inflight requests per Kafka connection when it was closed by a WarpStream Agent | | **warpstream.agent_kafka_inflight_requests**(gauge) | gauge of the number of inflight Kafka requests issued to a WarpStream Agent | | **warpstream.agent_kafka_joingroup_poll_iterations**(gauge) | distribution of the number of iterations to poll when processing a Kafka JoinGroup request | | **warpstream.agent_kafka_produce_compressed_bytes**(gauge) | distribution of the number of compressed bytes written when processing Kafka produce requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_produce_compressed_bytes_counter**(count) | counts the number of compressed bytes written when processing Kafka produce requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_produce_uncompressed_bytes**(gauge) | distribution of the number of uncompressed bytes written when processing Kafka produce requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_produce_uncompressed_bytes_counter**(count) | counts the number of uncompressed bytes written when processing Kafka produce requests issued to a WarpStream Agent*Shown as byte* | | **warpstream.agent_kafka_produce_with_offset_uncompressed_bytes**(gauge) | distribution of the number of uncompressed bytes written when processing Kafka produce with offset requests issued to a WarpStream Agent | | **warpstream.agent_kafka_produce_with_offset_uncompressed_bytes_counter**(count) | number of uncompressed bytes written when processing Kafka produce with offset requests issued to a WarpStream Agent, tagged by `outcome`*Shown as byte* | | **warpstream.agent_kafka_request_count**(count) | counts the number of Kafka requests handled by a WarpStream Agent, tagged by `kafka_key` | | **warpstream.agent_kafka_request_latency**(gauge) | latency of Kafka requests issued to a WarpStream Agent, tagged by `kafka_key` and `outcome`*Shown as second* | | **warpstream.agent_kafka_request_outcome**(count) | counts the number of Kafka requests handled by a WarpStream Agent, tagged by `kafka_key` and `outcome` | | **warpstream.agent_kafka_request_process_latency**(gauge) | latency of processing Kafka requests issued to a WarpStream Agent, tagged by `kafka_key`*Shown as second* | | **warpstream.agent_kafka_source_cluster_connections_counter**(count) | number of connections opened from Orbit to a source cluster | | **warpstream.agent_net_kafka_tcp_idle_conn_close**(count) | counts the number of times a WarpStream Agent closed an idle TCP connection to a Kafka client due to inactivity | | **warpstream.agent_net_kafka_tcp_limit_listener_acquire_count**(count) | counts the number of times a WarpStream Agent accepted a new Kafka active connection | | **warpstream.agent_net_kafka_tcp_limit_listener_acquire_latency**(gauge) | latency to accept a new Kafka active connection*Shown as second* | | **warpstream.agent_produce_backpressure_count**(count) | counts the number of times a WarpStream Agent had to backpressure a produce request, tagged by 'kafka_method' and 'backpressure_source' | | **warpstream.agent_produce_outcome**(count) | counts the number of produce requests handled by a WarpStream Agent, tagged by 'outcome' | | **warpstream.agent_run_and_ack_job_latency**(gauge) | latency of jobs run and acknowledged by the WarpStream Agent job runner*Shown as second* | | **warpstream.agent_run_and_ack_job_outcome**(count) | counts the number of jobs run and acknowledged by the WarpStream Agent job runner, tagged by 'outcome' | | **warpstream.agent_segment_batcher_flush_file_size_compressed_bytes**(gauge) | distribution of the compressed size of files batched in-memory before flushing to the object storage when processing a produce request*Shown as byte* | | **warpstream.agent_segment_batcher_flush_file_size_uncompressed_bytes**(gauge) | distribution of the uncompressed size of files batched in-memory before flushing to the object storage when processing a produce request*Shown as byte* | | **warpstream.agent_segment_batcher_flush_new_file_duration**(gauge) | distribution of the latency to batch in-memory before flushing to the object storage when processing a produce request*Shown as second* | | **warpstream.agent_segment_batcher_flush_num_batches**(gauge) | distribution of the number of batches in a file flushed to object storage | | **warpstream.agent_segment_batcher_flush_num_records_counter**(count) | counts the number of records in a file flushed to object storage | | **warpstream.agent_segment_batcher_flush_num_records_distribution**(gauge) | distribution of the number of records in a file flushed to object storage | | **warpstream.agent_segment_batcher_flush_outcome**(count) | counts the number of files flushed to object storage, tagged by 'outcome' | | **warpstream.agent_segment_batcher_flush_put_file_duration**(gauge) | latency to advertise a new file to WarpStream control plane*Shown as second* | | **warpstream.agent_segment_batcher_flush_total_duration**(gauge) | total time to flush a file and advertise it to WarpStream control plane*Shown as second* | | **warpstream.agent_segment_batcher_flush_write_to_duration**(gauge) | time to upload a new file to object storage*Shown as second* | | **warpstream.blob_store_get_stream_range_num_bytes_distribution**(gauge) | latency of the number of bytes read from object storage when reading ranges from a streams*Shown as byte* | | **warpstream.blob_store_operation_latency**(gauge) | latency of requests to object storage, tagged by `operation` and `outcome`*Shown as second* | | **warpstream.build_age_unix_ms**(gauge) | time since the WarpStream Agent binary was built | | **warpstream.circuit_breaker_count**(count) | counts the number of times a circuit breaker was tripped, tagged by `name` and `state` | | **warpstream.circuit_breaker_hits**(count) | counts the number of hits to a circuit breaker, tagged by `name` and `outcome` | | **warpstream.circuit_breaker_permit**(count) | counts the number of permits granted by a circuit breaker, tagged by `name` and `outcome` | | **warpstream.circuit_breaker_state_set**(gauge) | gauge of the current state of a circuit breaker, tagged by `name` and `state` | | **warpstream.consumer_group_estimated_lag_very_coarse_do_not_use_to_measure_e2e_seconds**(gauge) | consumer time lag estimate, its tagging depends on the WarpStream Agent settings*Shown as second* | | **warpstream.consumer_group_generation_id**(gauge) | indicates the generation number of the consumer group, incrementing by one with each rebalance. It serves as an effective indicator for detecting occurrences of rebalances. | | **warpstream.consumer_group_lag**(gauge) | sum of the consumer group lag across the topic's partitions | | **warpstream.consumer_group_num_members**(gauge) | number of members in each consumer group | | **warpstream.consumer_group_num_partitions**(gauge) | number of partitions in each consumer group | | **warpstream.consumer_group_num_topics**(gauge) | number of topics in each consumer group | | **warpstream.consumer_group_state**(gauge) | indicates state of consumer group, tagged by 'consumer_group' and 'group_state' | | **warpstream.container_running**(gauge) | gauge set to 1 to indicate that the container is running | | **warpstream.cpu_utilization_go_runtime**(gauge) | current process's CPU utilization from the Go runtime's perspective, recorded as a ratio between 0 and 1 | | **warpstream.cpu_utilization_os**(gauge) | current process's CPU utilization from the operating system's perspective, recorded as a ratio between 0 and 1 | | **warpstream.diagnostic_status**(gauge) | diagnostic status, tagged by 'diagnostic_name', 'diagnostic_type', 'successful' and 'muted'. This metric is deprecated, you should use 'diagnostic_failure' instead. | | **warpstream.error_count**(count) | counts the number of errors logged | | **warpstream.fast_retrier_execution**(count) | counts the number of times the fast retrier was used | | **warpstream.fast_retrier_p99_ms**(gauge) | p99 latency tracked by a fast retrier*Shown as millisecond* | | **warpstream.fast_retrier_speculative_retry**(count) | counts the number of times a speculative retry was performed by the fast retrier | | **warpstream.file_cache_batcher_batches_count**(count) | number of batches to the file cache | | **warpstream.file_cache_batcher_batches_distribution**(gauge) | distribution of the number of batches to the file cache | | **warpstream.file_cache_batcher_called**(count) | number of times a WarpStream Agent batched requests to the file cache | | **warpstream.files_count**(gauge) | number of files tracked by WarpStream control plane, tagged by 'level' of compaction | | **warpstream.get_records_batch_batches_count**(count) | number of batches when issuing a GetRecords request to WarpStream control plane | | **warpstream.get_records_batch_batches_distribution**(gauge) | distribution of number of batches when issuing a GetRecords request to WarpStream control plane | | **warpstream.get_records_batch_called**(count) | number of times a WarpStream Agent batched GetRecords requests | | **warpstream.global_ratelimit_fetch_compressed_bytes**(count) | number of fetch request bytes rate limited*Shown as byte* | | **warpstream.kafka_batcher_batches_count**(count) | number of batches when issuing Kafka requests (not produce nor fetch) to WarpStream control plane | | **warpstream.kafka_batcher_batches_distribution**(gauge) | distribution of number of batches when issuing Kafka requests (not produce nor fetch) to WarpStream control plane | | **warpstream.kafka_batcher_called**(count) | number of times a WarpStream Agent processed a Kafka request (not produce nor fetch) | | **warpstream.list_streams_batcher_batches_count**(count) | number of batches when issuing a ListStream request to WarpStream control plane | | **warpstream.list_streams_batcher_batches_distribution**(gauge) | distribution of the number of batches when issuing a ListStream request to WarpStream control plane | | **warpstream.list_streams_batcher_called**(count) | number of times an Agent issued a ListStream request | | **warpstream.loading_cache_loader_outcome**(count) | counts the number of loading cache load operations, tagged by 'cache_name' and 'outcome' | | **warpstream.loading_cache_refresh_completed**(count) | counts the number of completed loading cache refresh operations, tagged by 'cache_name' and 'outcome' | | **warpstream.max_offset**(gauge) | max Kafka offset | | **warpstream.net_http_tcp_limit_listener_acquire_count**(count) | number of times a WarpStream Agent accepts a new HTTP active connection | | **warpstream.net_http_tcp_limit_listener_acquire_latency**(gauge) | latency to accept a new HTTP active connection*Shown as second* | | **warpstream.num_records**(gauge) | number of records | | **warpstream.partitions_count**(gauge) | number of partitions | | **warpstream.partitions_count_limit**(gauge) | max number of partitions | | **warpstream.schema_versions_count**(gauge) | number of schema registry versions | | **warpstream.schema_versions_limit**(gauge) | max number of schema registry versions | | **warpstream.server_starting**(count) | counts the number of times a Warpstream Agent has started | | **warpstream.server_tick_running**(count) | counts the number of Warpstream Agent server ticks | | **warpstream.target_release_server_middleware**(count) | counts the number of times a WarpStream Agent checked that a HTTP request was intended for this WarpStream cluster, tagged by `outcome` | | **warpstream.topics_count**(gauge) | number of topics | | **warpstream.topics_count_limit**(gauge) | max number of topics | | **warpstream.agent_ripcord_delete_file**(count) | counts the number of files deleted by the ripcord replay job tagged by `outcome` and `reason` | | **warpstream.agent_ripcord_replayed_file**(count) | counts the number of files replayed by the ripcord replay job tagged by `live` | | **warpstream.query_engine_active_queries_count**(gauge) | distribution of the number of active queries being currently processed by the query engine | | **warpstream.diagnostic_failure**(gauge) | diagnostic failures, tagged by 'diagnostic_name', 'diagnostic_type', 'severity_low', 'severity_medium', 'severity_high', 'muted' | | **warpstream.events_processor_produce_success_count**(count) | counts the number of events successfully produced from the WarpStream Agent events processor to Kafka | | **warpstream.events_processor_produce_error_count**(count) | counts the number of errors encountered when producing events from the WarpStream Agent events processor to Kafka | | **warpstream.events_processor_submit_bytes_count**(count) | counts the number of bytes submitted to the WarpStream Agent events processor | | **warpstream.events_processor_submit_success_count**(count) | counts the number of events successfully submitted to the WarpStream Agent events processor | | **warpstream.orbit_handler_num_fetches**(count) | counts the number of fetch operations performed by Orbit | | **warpstream.orbit_handler_num_polls**(count) | counts the number of Orbit fetch and produce loops executed | | **warpstream.orbit_handler_bytes_fetched_during_poll**(count) | counts the number of bytes fetched during an Orbit fetch and produce loop*Shown as byte* | | **warpstream.orbit_handler_bytes_in_fetch**(count) | counts the number of bytes fetched from the source cluster*Shown as byte* | | **warpstream.agent_kafka_fetch_num_pointers_counter_by_level**(count) | counts the number of pointers read when processing Kafka fetch requests issued to a WarpStream Agent tagged by `compaction_level` | | **warpstream.agent_kafka_fetch_num_pointers_distribution_by_level**(gauge) | distribution of the number of pointers read when processing Kafka fetch requests issued to a WarpStream Agent tagged by `compaction_level` | | **warpstream.agent_control_plane_operation_latency**(gauge) | latency of control plane operations performed by a WarpStream Agent, tagged by `operation` and `outcome`*Shown as second* | | **warpstream.agent_kafka_produce_records_counter**(count) | counts the number of events successfully produced | | **warpstream.agent_ripcord_deleted_file_queue**(gauge) | latency between the time a file was queued for deletion and it actually got removed*Shown as second* | | **warpstream.loading_cache_refresh_throttled**(count) | counts the number of times a loading cache refresh was throttled, tagged by 'cache_name' | | **warpstream.agent_kafka_connection_state_cache_outcome**(count) | counts the number of connection state cache lookups in a WarpStream Agent, tagged by `outcome` | | **warpstream.logs_async_queue_push_count**(count) | counts the number of logs pushed to the async queue | | **warpstream.node_heartbeat**(count) | counts the number of heartbeats sent by a Warpstream Agent node, tagged by `outcome` and `mechanism` | | **warpstream.single_value_cache_loader_outcome**(count) | counts the number of single value cache load operations, tagged by 'cache_name' and 'outcome' | | **warpstream.agent_kafka_fetch_request_num_top_level_fetch_had_no_pointers_in_a_row**(gauge) | distribution of the number of fetches that did not retrieve any data consecutively | | **warpstream.agent_segment_batcher_flush_to_start_commit_duration**(gauge) | distribution of the latency to batch in-memory before committing the new files to the WarpStream control plane*Shown as second* | | **warpstream.embedded_node_pool_tracker_refresh_by_mechanism_agent**(count) | counts the number of times the embedded node pool tracker was refreshed in a Warpstream Agent | | **warpstream.logs_async_queue_size_bytes**(gauge) | gauge of the current size in bytes of the logs async queue*Shown as byte* | | **warpstream.control_plane_utilization**(gauge) | value between 0 and 1 to represent the utilization of the control plane for the virtual cluster. Getting close to 1 means that the control plane will start throttling requests. | | **warpstream.agent_control_plane_operation_counter**(count) | counts the number of control plane operations performed by a WarpStream Agent, tagged by `operation` and `outcome` | | **warpstream.agent_kafka_fetch_records_counter**(count) | counts the number of records present in all of the data pages the Warpstream Agent fetched to respond to fetches | | **warpstream.consistent_assignments_cache_invalidation**(count) | counts the number times a WarpStream Agent had to invalidate one of its metadata cache | | **warpstream.agent_kafka_metadata_preserve_previous_partition_assignments**(count) | counts the number of times we were able to re-use a previous partition assignment while handling a metadata request tagged by `outcome` | | **warpstream.agent_kafka_request_latency_before_response**(gauge) | latency to process a Kafka request before starting to write the response in the TCP socket tagged by `outcome` and `kafka_key`*Shown as second* | | **warpstream.query_engine_runnable_tasks_count**(gauge) | distribution of the number of runnable tasks in the query engine | | **warpstream.schema_validation_invalid_record_count**(count) | counts the number of invalid records found during schema validation | | **warpstream.logs_forwarder_error**(count) | counts the number of errors encountered when forwarding logs to WarpStream control plane | | **warpstream.agent_ripcord_ingestion_latency**(gauge) | latency to ingest data in lightning topics or ripcord mode*Shown as second* | | **warpstream.query_engine_queue_runnable_task**(count) | counts the number of runnable tasks queued in the query engine | | **warpstream.query_engine_query_started**(count) | counts the number of queries the query engine has started to process | | **warpstream.query_engine_exec_task**(count) | counts the number of tasks being executed by the query engine | | **warpstream.query_engine_task_completed_count**(count) | counts the number of tasks completed by the query engine tagged by `outcome` | | **warpstream.query_engine_task_completed_duration**(gauge) | latency of task completion in the query engine tagged by `outcome`*Shown as nanosecond* | | **warpstream.agent_schema_registry_response_bytes_counter**(count) | counts the number of bytes of schema registry responses tagged by `outcome` and `operation` | | **warpstream.schema_linking_source_subject_versions_count**(gauge) | total number of subject versions synced by schema linking tagged by `sync_id` and `config_id` | | **warpstream.schema_linking_newly_migrated_subject_versions**(gauge) | number of subject versions just migrated by schema linking tagged by `sync_id` and `config_id` | | **warpstream.agent_schema_registry_inflight_connections**(gauge) | gauge of the number of inflight Schema Registry connections opened to a Warpstream Agent | | **warpstream.agent_schema_registry_request_bytes_counter**(count) | counts the number of uncompressed bytes read when processing Schema Registry fetch requests | | **warpstream.agent_schema_registry_request_latency**(gauge) | latency of Schema Registry fetch requests issued to a Warpstream Agent*Shown as second* | | **warpstream.agent_schema_registry_request_outcome**(count) | counts the number of schema registry requests handled by the agent, tagged by `schema_registry_operation` and `outcome` | | **warpstream.tableflow_snapshots_limit**(gauge) | max number of Tableflow snapshots on a given cluster | | **warpstream.tableflow_tables_limit**(gauge) | max number of Tableflow tables on a given cluster | | **warpstream.tableflow_files_limit**(gauge) | max number of Tableflow files on a given cluster | | **warpstream.tableflow_partitions_limit**(gauge) | max number of Tableflow partitions on a given cluster | | **warpstream.agent_tableflow_files_written**(count) | number of files written by WarpStream Agents while ingesting from Kafka to Tableflow tagged by `source_topic` | | **warpstream.ingestion_job_kafka_time_lag_ms**(gauge) | time lag of the Tableflow ingestion job that is consuming from Kafka and producing Tableflow files - this is deprecated please use `warpstream.tableflow_partition_time_lag_seconds`*Shown as millisecond* | | **warpstream.ingestion_job_dlq_skip_strategy_counter**(count) | DEPRECATED: use warpstream.tableflow_dlq_records_counter with strategy tag instead | | **warpstream.tableflow_dlq_records_counter**(count) | counts the number of records handled by the Tableflow DLQ during ingestion tagged by `topic` and `strategy` (skip or keep) | | **warpstream.tableflow_tables_count**(count) | counts the number of Tableflow tables created by WarpStream Agents | | **warpstream.tableflow_snapshots_count**(count) | counts the number of Tableflow snapshots created by WarpStream Agents | | **warpstream.tableflow_files_count**(count) | counts the number of Tableflow files created by WarpStream Agents | | **warpstream.tableflow_partitions_count**(count) | counts the number of Tableflow partitions created by WarpStream Agents | | **warpstream.tableflow_partition_time_lag_seconds**(gauge) | time lag of the Tableflow ingestion job that is consuming from Kafka and producing Tableflow files*Shown as second* | | **warpstream.tableflow_partition_offset_lag**(gauge) | offset lag (number of records) of the Tableflow ingestion job that is consuming from Kafka and producing Tableflow files | | **warpstream.batcher_batches_count**(count) | counts the number of items in each batch produced by a batcher | | **warpstream.batcher_batches_distribution**(gauge) | distribution of the number of items in each batch produced by a batcher | | **warpstream.batcher_called**(count) | counts the number of times a batcher was called | | **warpstream.agent_file_cache_too_many_concurrent_fetches**(count) | counts the number of times a WarpStream Agent filecache had to backpressure because there were too many concurrent fetches in parallel | | **warpstream.agent_tableflow_bytes_written**(count) | number of bytes written by WarpStream Agents while ingesting from Kafka to Tableflow tagged by `source_topic` | | **warpstream.logs_forwarder_channel_full**(count) | counts the number of times the log forwarder channel was full | | **warpstream.http_instrumented_middleware_count**(count) | counts the number of HTTP requests processed by the service | | **warpstream.logs_async_queue_closed_overflow_count**(count) | counts the number of logs dropped due to the async queue being closed | | **warpstream.agent_acl_denied**(count) | counts the number of ACL denials on a WarpStream Agent, tagged by `resource_type` and `operation` | | **warpstream.ratelimiter_nola_local_refill_request_size**(gauge) | gauge of the local refill request size for the NOLA rate limiter | | **warpstream.embedded_node_pool_tracker_read_through_requests_agent_by_mechanism**(count) | counts the number of read-through requests made by the embedded node pool tracker in a Warpstream Agent, tagged by `mechanism` | | **warpstream.agent_ripcord_outstanding_replays_count**(gauge) | number of replays that are currently opened in the ripcord folder | | **warpstream.http_instrumented_middleware_latency**(gauge) | latency of HTTP requests processed by the service*Shown as second* | | **warpstream.http_ratelimiter_latency**(gauge) | latency of HTTP requests processed by the HTTP rate limiter*Shown as second* | | **warpstream.agent_query_engine_active_queries_limit_hit**(count) | counts the number of times the active queries limit was hit in the agent query engine | | **warpstream.events_processor_submit_error_count**(count) | counts the number of errors encountered when submitting events to the WarpStream Agent events processor | | **warpstream.logs_async_queue_overflow_count**(count) | counts the number of logs dropped due to the async queue being full | | **warpstream.ratelimiter_nola_acquire_ratelimited_wait**(count) | counts the number of token acquire attempts that were rate limited and waited | | **warpstream.agent_active_pipeline_instances**(gauge) | gauge of the number of active pipeline instances running on a WarpStream Agent | | **warpstream.ratelimiter_nola_refill_tokens_num_tokens_requested_counter**(count) | counts the number of tokens requested during refill attempts | | **warpstream.ratelimiter_nola_init_timeout**(count) | counts the number of token acquire attempts that timed out during initialization | | **warpstream.ratelimiter_nola_local_refill_response_size**(gauge) | gauge of the local refill response size for the NOLA rate limiter | | **warpstream.publish_all**(count) | counts the total number of times we have published all contexts, tagged by `outcome` | | **warpstream.invariant_count**(count) | counts the number of times an invariant was violated | | **warpstream.blob_store_bytes_written**(count) | counts the total number of bytes written to object storage*Shown as byte* | | **warpstream.ingestion_job_transform_drop_counter**(count) | counts the number of records that were dropped during Tableflow ingestion tagged by `topic` | | **warpstream.agent_ripcord_sequence_closed**(count) | counts the number of sequences closed by the agent | | **warpstream.ratelimiter_nola_acquire_success**(count) | counts the number of successful token acquire attempts | | **warpstream.agent_kafka_produce_uncompressed_bytes_per_topic_single_produce_request_distribution**(gauge) | distribution of the number of uncompressed bytes per topic in a single produce request processed by a WarpStream Agent*Shown as byte* | | **warpstream.logs_payload_too_large**(count) | counts the number of times a log payload was too large to be forwarded to WarpStream control plane | | **warpstream.truncated_logs_count**(count) | counts the number of logs that were truncated before being written | | **warpstream.ratelimiter_nola_refill_tokens_outcome**(count) | counts the outcome of token refill attempts | | **warpstream.ratelimiter_nola_refill_tokens_num_tokens_received_counter**(count) | counts the number of tokens received during refill attempts | | **warpstream.embedded_node_pool_tracker_read_through_requests_agent**(count) | counts the number of read-through requests made by the embedded node pool tracker in a Warpstream Agent | | **warpstream.agent_ripcord_oldest_replay_age**(gauge) | age of the oldest replay in the ripcord folder in seconds*Shown as second* | | **warpstream.blob_store_resume_get**(count) | counts the number of times a blob store had to resume a get operation that was interrupted | | **warpstream.truncated_logs_count_bytes**(count) | counts the total number of bytes of logs that were truncated before being written*Shown as byte* | | **warpstream.ratelimiter_nola_acquire_bypass**(count) | counts the number of token acquire attempts that bypassed the rate limiter | | **warpstream.ratelimiter_nola_acquire_ratelimited**(count) | counts the number of token acquire attempts that were rate limited | | **warpstream.agent_query_engine_query_started**(count) | counts the number of queries started in the agent query engine | | **warpstream.blob_store_bytes_read**(count) | counts the total number of bytes read from object storage*Shown as byte* | | **warpstream.embedded_node_pool_tracker_requests_agent**(count) | counts the number of requests made by the embedded node pool tracker in a WarpStream Agent | | **warpstream.unauthenticated_http_ratelimiter_latency**(gauge) | latency of unauthenticated HTTP requests processed by the unauthenticated HTTP rate limiter*Shown as second* | | **warpstream.events_processor_event_too_large_count**(count) | counts the number of events dropped by the WarpStream Agent events processor due to exceeding the maximum allowed event size | | **warpstream.data_lake_file_exists_verifier_run_outcome**(count) | counts the number of times the data lake file exists verifier verified a file, tagged by `outcome` | | **warpstream.bento_rate_limit_checked**(count) | counts the number of times Bento's rate limit was checked in Managed Data Pipelines | | **warpstream.bento_processor_batch_received**(count) | counts the number of times a batch was received by a Bento's processor in Managed Data Pipelines | | **warpstream.bento_batch_created**(count) | counts the number of times a batch was created in Managed Data Pipelines | | **warpstream.bento_input_received**(count) | counts the number of times an input was received in Managed Data Pipelines | | **warpstream.bento_processor_sent**(count) | counts the number of times a processor was sent in Managed Data Pipelines | | **warpstream.bento_output_batch_sent**(count) | counts the number of times a batch was sent in Managed Data Pipelines | | **warpstream.bento_processor_latency_ns**(gauge) | latency of Managed Data Pipelines processor (naming is confusing because Bento's metrics are forwarded as is)*Shown as second* | | **warpstream.bento_output_latency_ns**(gauge) | latency of Managed Data Pipelines output (naming is confusing because Bento's metrics are forwarded as is)*Shown as second* | | **warpstream.bento_input_latency_ns**(gauge) | latency of Managed Data Pipelines input (naming is confusing because Bento's metrics are forwarded as is)*Shown as second* | | **warpstream.bento_output_sent**(count) | counts the number of times an output was sent in Managed Data Pipelines | | **warpstream.bento_processor_received**(count) | counts the number of times a processor was received in Managed Data Pipelines | | **warpstream.bento_processor_batch_sent**(count) | counts the number of times a processor batch was sent in Managed Data Pipelines | | **warpstream.pressure_cpu_some_pct**(gauge) | CPU pressure: percent of time some tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.pressure_cpu_full_pct**(gauge) | CPU pressure: percent of time all tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.pressure_memory_some_pct**(gauge) | memory pressure: percent of time some tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.pressure_memory_full_pct**(gauge) | memory pressure: percent of time all tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.pressure_io_some_pct**(gauge) | I/O pressure: percent of time some tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.pressure_io_full_pct**(gauge) | I/O pressure: percent of time all tasks were stalled over the sampling interval*Shown as percent* | | **warpstream.sched_goroutines_runnable**(gauge) | number of goroutines ready to run, waiting for a P | | **warpstream.sched_goroutines_running**(gauge) | number of goroutines currently executing on a thread | | **warpstream.sched_goroutines_waiting**(gauge) | number of goroutines blocked on I/O, channels, locks, etc. | | **warpstream.sched_goroutines_not_in_go**(gauge) | number of goroutines in syscalls or cgo | | **warpstream.sched_goroutines_created**(gauge) | cumulative total goroutines ever created | | **warpstream.sched_threads_total**(gauge) | total OS threads managed by the Go runtime | | **warpstream.agent_produce_out_of_sequence_soft**(count) | counts the number of out of sequence soft produce requests handled by a WarpStream Agent | | **warpstream.bento_input_connection_up**(count) | counts the number of times the input has successfully established a connection to the target source in Managed Data Pipelines | | **warpstream.bento_input_connection_lost**(count) | counts the number of times the input has lost a previously established connection to the target source in Managed Data Pipelines | | **warpstream.bento_input_connection_failed**(count) | counts the number of times the input has failed to establish a connection to the target source in Managed Data Pipelines | | **warpstream.bento_rate_limit_triggered**(count) | counts the number of times the rate limit has been triggered by a probe in Managed Data Pipelines | | **warpstream.bento_output_connection_up**(count) | counts the number of the times the output has successfully established a connection to the target sink in Managed Data Pipelines | | **warpstream.bento_output_connection_lost**(count) | counts the number of times the output has lost a previously established connection to the target sink in Managed Data Pipelines | | **warpstream.bento_output_connection_failed**(count) | counts the number of times the output has failed to establish a connection to the target sink in Managed Data Pipelines | | **warpstream.tableflow_ingestion_lag_seconds**(gauge) | measures how far behind the ingestion process is from the source Kafka topic*Shown as second* | | **warpstream.tableflow_query_lag_seconds**(gauge) | measures the time from when data is produced to Kafka until it is queryable (ingestion lag and catalog sync delay)*Shown as second* | | **warpstream.bento_output_error**(count) | counts the number of send attempts that have failed in Managed Data Pipelines. On failed batched sends this count is incremented once only | | **warpstream.cpu_utilization_go_runtime_moving_average_sample**(count) | counts each value appended to the Go runtime CPU utilization moving average | | **warpstream.cpu_utilization_go_runtime_moving_average_sample_utilization_distribution**(gauge) | distribution of the values appended to the Go runtime CPU utilization moving average, recorded as ratios between 0 and 1 | | **warpstream.agent_kafka_produce_forwarded_records_counter**(count) | counts the number of records successfully forwarded to the source Kafka cluster during auto migration | | **warpstream.agent_kafka_produce_forwarded_compressed_bytes**(gauge) | distribution of the number of compressed bytes successfully forwarded to the source Kafka cluster during auto migration*Shown as byte* | | **warpstream.agent_kafka_produce_forwarded_compressed_bytes_counter**(count) | counts the number of compressed bytes successfully forwarded to the source Kafka cluster during auto migration*Shown as byte* | | **warpstream.agent_cpu**(gauge) | latest CPU usage percentage reported by a live agent heartbeat*Shown as percent* | | **warpstream.min_offset**(gauge) | oldest offset still retained for a topic partition | | **warpstream.max_offset**(gauge) | high watermark offset (next offset to be written) for a topic partition | | **warpstream.consumer_group_commit_ts**(gauge) | timestamp (nanoseconds since epoch) of the last offset commit for a partition consumed by a Kafka consumer group*Shown as nanosecond* | | **warpstream.partition_size_estimated_compressed_bytes**(gauge) | estimated compressed size in bytes of a topic partition's retained data*Shown as byte* | | **warpstream.agent_num_vcpus**(gauge) | number of vCPUs reported by a live agent heartbeat | | **warpstream.produced_records**(count) | monotonically increasing count of records produced to a topic partition | | **warpstream.partition_size_uncompressed_bytes**(gauge) | uncompressed size in bytes of a topic partition's retained data*Shown as byte* | | **warpstream.consumer_group_max_offset**(gauge) | high watermark offset for a partition consumed by a Kafka consumer group | | **warpstream.agent_heartbeat**(gauge) | one sample per live agent heartbeat. Always 1; labels identify the agent, role set, group, availability zone, and other placement metadata | ## Uninstallation{% #uninstallation %} 1. Remove the integration yaml config file and restart the Datadog Agent. 1. Unset the `-enableDatadogMetrics` flag and restart the WarpStream Agent. ## Support{% #support %} Need help? Contact us through the following channels: - Slack: [https://console.warpstream.com/socials/slack](https://console.warpstream.com/socials/slack) - Discord: [https://discord.com/invite/rSFx8vqjVY](https://discord.com/invite/rSFx8vqjVY) ## Further Reading{% #further-reading %} Additional helpful documentation, links, and articles: - [Kafka Is Dead, Long Live Kafka: The Case for Diskless, S3-Native Streaming](https://www.warpstream.com/blog/kafka-is-dead-long-live-kafka)