--- title: WarpStream description: Monitor the health and performance of your WarpStream Agents breadcrumbs: Docs > Integrations > WarpStream --- # WarpStream Supported OS Integration version1.0.0 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 %} [Install the Datadog Agent](https://docs.datadoghq.com/developers/dogstatsd/) version `>=6.17` or `>=7.17`. ### Configuration{% #configuration %} Complete all of the following steps to ensure the WarpStream integration works properly. There are two parts of the WarpStream integration: - The **Datadog Agent** portion, which makes requests to a provided endpoint for the WarpStream agent to report whether it can connect and is healthy. - The **WarpStream StatsD** portion, where the WarpStream Agent can be configured 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 Datadog Agent WarpStream integration{% #configure-datadog-agent-warpstream-integration %} Configure the WarpStream check included in the [Datadog Agent](https://app.datadoghq.com/account/settings/agent/latest) package to collect health metrics and service checks. This can be done by editing the `url` within the `warpstream.d/conf.yaml` file, in the `conf.d/` folder at the root of your Agent's configuration directory, to start collecting your WarpStream service checks. 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. Ensure that `url` matches your WarpStream Agent HTTP server (port 8080 by default). ##### Connect WarpStream Agent to DogStatsD{% #connect-warpstream-agent-to-dogstatsd %} Start the 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://github.com/DataDog/integrations-extras/blob/master/warpstream/metadata.csv). 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 | | **warpstream.cpu\_utilization\_os**(gauge) | current process's CPU utilization from the operating system's perspective | | **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) | counts the number of records that were skipped during Tableflow ingestion while using the skip strategy tagged by `topic` | | **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.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 | ## 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 %} - [Kafka Is Dead, Long Live Kafka: The Case for Diskless, S3-Native Streaming](https://www.warpstream.com/blog/kafka-is-dead-long-live-kafka)