datadogJul 1, 2026

How we measure data completeness at scale (opens in new tab)

Datadog built a real-time data-completeness system to ensure that every customer’s telemetry is available for dashboards, alerts, queries, and AI-driven decisions. Because ingestion spans hundreds of distributed paths and customers may send delayed or retried data, global or watermark-based tracking is unreliable. The system instead tracks payloads segment by segment, using idempotent create and acknowledgment events to identify losses and calculate end-to-end completeness. ## Defining Completeness at Datadog’s Scale - Completeness means every ingested payload—metrics, logs, spans, or other telemetry—is ultimately available to customers. - The system must measure completeness: - Across hundreds of services and ingestion paths - For each individual customer - In real time - With enough detail to identify where degradation occurred - Customer traffic may take different routes because of partitioning, isolation, and traffic patterns. - Metrics and APM pipelines can each involve hundreds or tens of distinct paths, creating a large number of possible failure points. - The completeness system must remain independent of the services it monitors so it can provide trustworthy diagnostics during incidents. ## Tracking Completeness by Pipeline Segment - Datadog considered watermark-based tracking, but delayed customer data, replayed traffic, and pipeline loops made predictable watermarks impractical. - Pipelines are divided into segments representing steps within or between services. - For example, intake-in to intake-out is one segment. - Intake-out to processing-in is another. - Each segment is measured independently, allowing engineers to locate degradation within a service or between services. - Segment-level tracking also adapts to pipelines whose branches appear or disappear over time. ## Counting Creates and Acknowledgments - When a payload enters a segment, the system records a create event. - When it exits, the system records an acknowledgment using the payload’s unique identifier. - Comparing creates with acknowledgments reveals whether payloads were lost in that segment. - Events are organized into time buckets based on when the payload first entered Datadog, using a Datadog-controlled timestamp rather than the customer’s clock. - Each identifier has a state per segment: - Created - Acknowledged - Acknowledged before the create event arrived - Duplicate create or acknowledgment events are ignored, making the system idempotent despite retries and event reordering. ## Calculating End-to-End Completeness - Segment completeness is the ratio of payloads exiting a segment to those entering it. - For sequential services, overall completeness is calculated by multiplying segment ratios. - Parallel branches require a different approach: - Treating branches as one pipeline would make completeness wait for the slowest branch. - Instead, Datadog uses a weighted average, giving each branch influence proportional to the volume it processes. - In the example, one branch reaches 94% completeness by multiplying 98% and 96% across two sequential services, while another branch reaches 100%. - Combining these branch measurements produces a more accurate view of currently available data without incorrectly marking all data incomplete because one branch is slower. ## Practical Conclusion Segment-level, identifier-based tracking gives Datadog a real-time and customer-specific view of data completeness. It both supports reliable end-to-end calculations and helps humans or automated systems quickly determine where ingestion problems are occurring.

kafkaobservabilitydistributed-systemsapm+3

datadogJun 23, 2026

How we migrated a live routing system using AI-assisted refactoring (opens in new tab)

Stream Router evolved from a small configuration file into a critical control-plane service routing Datadog’s massive metrics workload. Its original FoundationDB key-value model eventually hit transaction-size and performance limits because relational relationships were reconstructed in application code. Datadog redesigned the system around PostgreSQL and DuckDB, using AI-assisted, test-driven refactoring to accelerate the migration without disrupting production traffic. ## Stream Router’s Role in Datadog’s Metrics Pipeline - Datadog processes more than a hundred trillion events per day. - Stream Router determines which Kafka cluster, topic, partitions, and sharding strategy should handle each datapoint. - It serves both producers and queriers but does not process Kafka messages itself. - Routing decisions change frequently as infrastructure evolves, making correctness and historical tracking essential. ## From Configuration File to Control Plane - In 2016, routing was managed through a small configuration file distributed to services. - As the platform grew, the file expanded to thousands of lines and required manual edits and rollouts. - Stream Router replaced this workflow with: - A centralized gRPC service - API-managed routes - Automated, gradual rollouts - The write path used FoundationDB, while the read path served static RocksDB snapshots restored into memory. - This eventually became a bottleneck as routing tables and operational changes grew larger. ## Why the Key-Value Model Stopped Scaling - Routes reference streams and sharding strategies, while rules reference routes. - These relationships are inherently relational and require cross-entity validation. - The KV implementation loaded tens of thousands of records into application processes and reconstructed database-like relationships in code. - Some operations exceeded FoundationDB transaction-size limits. - Moving to PostgreSQL without changing the access patterns would not solve the issue; certain operations were estimated to require 45 minutes because of thousands of sequential database round trips. - The fundamental problem was the data model and application logic, not simply the choice of database. ## Designing the New Storage Architecture - The team redesigned the schema manually before using AI tools. - The relational model introduced explicit foreign keys between: - Streams - Sharding strategies - Routes - Rules - PostgreSQL was selected for the write path because it provided the required relational semantics and transaction model. - DuckDB was selected for the read path because: - It is embeddable and suitable for snapshot-based serving - It supports array columns - Its SQL dialect is closely compatible with PostgreSQL - Shared query logic could therefore work across both storage engines. ## AI-Assisted Refactoring - Claude and Cursor were used to accelerate a systematic, test-driven migration. - For each method, developers supplied: - The old implementation - The new schema - A failing test - AI generated an initial implementation, while tests determined whether it was correct. - The models assisted with method-level refactoring rather than autonomously designing the architecture. - Human expertise remained central to schema design, migration strategy, and evaluating system-level risks. ## Foundations for a Safe Migration - The migration benefited from infrastructure already present at Datadog. - Stream Router’s storage layer was isolated behind an internal `Controller` interface. - This modularity helped contain storage changes and enabled incremental refactoring. - Existing tests and clear boundaries provided confidence in generated implementations while production traffic continued. The central lesson is that AI was most effective as an accelerator inside a disciplined engineering process. A well-designed relational schema, modular storage abstraction, and failing tests provided the safety mechanisms; AI helped implement the resulting changes faster, but did not replace human architectural judgment.

kafkadatabase-designpostgresqlgrpc+4

datadogApr 7, 2026

How we built a real-world evaluation platform for autonomous SRE agents at scale (opens in new tab)

Bits AI SRE improved in isolated scenarios but lacked a way to detect regressions across the broader range of production incidents. The team found that tool-level tests and live replays could not capture failures caused by multi-step reasoning or changing telemetry. They built a replayable evaluation platform combining realistic investigation labels, scalable orchestration, and longitudinal performance tracking. ## Subtle Regressions from Well-Intentioned Features - Adding a monitor’s service name to the agent’s initial context improved some internal investigations. - Across broader scenarios, it introduced irrelevant signals that confused the agent and degraded unrelated investigations. - Because there was no representative evaluation set, the team could not measure the change’s wider impact before internal misses exposed it. - The incident demonstrated the need to evaluate every change across diverse investigation types. ## Limits of Tool Tests and Live Replay - Testing tools individually failed to capture errors caused by incorrect interactions between valid tool outputs. - Live investigation replay was difficult to scale because: - Results were not consistently aggregated. - Production environments changed. - Telemetry expired, making investigations unreplayable. - Standard evaluation frameworks assumed clean inputs and static datasets, unlike agents operating over production telemetry. - The team needed controlled, offline replay of realistic end-to-end investigations. ## Evaluation Labels and World Snapshots - Each label represents one production-style investigation. - It contains: - **Ground truth:** the issue’s actual root cause. - **World snapshot:** the queries and signals available when the issue occurred. - The agent is never shown the root cause directly; it must reason from the preserved signals. - Labels must cover varied technologies and failure modes, including: - Kubernetes pod failures - Kafka lag - Bad-code deployments - Complex multi-service business failures - A narrow or overly clean dataset would make performance appear better than it really is. ## Orchestrating Evaluations at Scale - The platform runs Bits against labels, scores the outcomes, and tracks quality over time. - It supports comparisons across: - Investigation categories - Model variants - Configuration versions - Evaluation runs - The architecture consists of a shared label set, an orchestration layer, and reporting infrastructure. - This allows teams to determine whether improvements in one domain, such as Kafka, regress another, such as Kubernetes. ## Scaling Label Creation - The team initially created labels manually from Datadog alerts. - Manual labeling provided early coverage but consumed engineering time and remained far from representative. - They embedded label generation into Bits itself: - Customer feedback and investigation data are used to derive root causes. - Relevant queries are preserved as the world snapshot. - Each user interaction becomes a potential evaluation case. - This increased label creation rates by an order of magnitude and allowed coverage to grow with product usage. ## Agent-Assisted Validation - Early labels required extensive human review, especially when feedback was ambiguous or reconstructed signals were uncertain. - As ingestion grew, manual review became a bottleneck. - Bits was then used to assist with validation by aggregating related signals, identifying relationships, and resolving ambiguous feedback before human review. ## Practical Conclusion Reliable agent improvement requires more than testing individual tools or replaying live incidents. A representative, production-derived label set combined with reproducible end-to-end evaluations makes regressions visible and enables safer iteration.

kafkak8sdatadogsre+3

datadogNov 4, 2025

Replication redefined: How we built a low-latency, multi-tenant data replication platform (opens in new tab)

Datadog built a managed, multi-tenant data replication platform to move data reliably across thousands of services without brittle, point-to-point integrations. The effort began by separating analytical search workloads from a shared PostgreSQL database, then evolved into automated pipeline provisioning with Temporal. The platform favors asynchronous replication to improve scalability and resilience, accepting limited replication lag in exchange for lower application latency and reduced operational coupling. ## Scaling Search Beyond PostgreSQL - A shared PostgreSQL database initially provided low-latency access, ACID guarantees, and low operational cost. - As data volumes grew, complex joins and aggregations became increasingly slow. - Datadog’s Metrics Summary page had to join: - 82,000 active metrics - 817,000 metric configurations - Page latency reached approximately 7 seconds at p90, while repeated facet changes generated additional expensive queries. - Index and disk bloat, memory pressure, VACUUM and ANALYZE overhead, and rising I/O wait further reduced throughput. - Rather than continuing to optimize PostgreSQL for analytical search, Datadog moved search and aggregation workloads to a dedicated search platform. - Data was denormalized during replication, producing document-oriented indexes better suited to faceted search. - The resulting system reduced page-load times by as much as 97%—from roughly 30 seconds to 1 second—while maintaining about 500 ms of replication lag. ## Automating Pipeline Provisioning with Temporal Provisioning a replication pipeline required coordinating multiple systems and configuration steps: - Enabling PostgreSQL logical replication with `wal_level`. - Creating users and assigning replication permissions. - Configuring publishers and replication slots. - Deploying Debezium instances to capture PostgreSQL changes. - Creating Kafka topics and mapping them to Debezium instances. - Adding heartbeat tables to monitor replication and prevent excessive WAL retention. - Configuring sink connectors to write Kafka data into the search platform. Manual management became increasingly difficult across many pipelines and data centers. Datadog used Temporal workflows to split provisioning into modular, repeatable tasks and combine them into higher-level orchestrations. This reduced errors, improved consistency, and allowed engineers to create and modify pipelines without repeating complex operational procedures. ## Choosing Asynchronous Replication - Synchronous replication provides strong consistency by waiting for replicas to acknowledge each write. - However, it increases latency and operational complexity, particularly across distributed environments. - Asynchronous replication allows the primary system to acknowledge writes immediately while replicas catch up afterward. - Datadog selected the asynchronous model because it decouples application performance from network latency and replica availability. - The trade-off is temporary replication lag during failures or periods of pressure, but the model offers better scalability and resilience for high-throughput systems. Datadog’s experience suggests that replication should be treated as a managed platform rather than a collection of custom integrations. Separating workloads, automating provisioning, and choosing asynchronous delivery can improve performance and reliability while reducing the operational burden on individual engineering teams.

kafkadatabase-designpostgresqlelasticsearch+4

datadogAug 12, 2025

Scaling down to speed up: How we improved efficiency of live process metrics by 100x (opens in new tab)

Datadog redesigned its real-time Processes and Containers pipeline to avoid collecting high-frequency metrics that users never see. By limiting 2-second collection to hosts actively viewed and using standard 10-second data for sorting, the company reduced real-time traffic by over 100x, cut infrastructure costs by 98%, and lowered Agent resource usage. The approach also improved scalability without sacrificing the live investigation experience. ## Original Real-Time Collection Model - Datadog Agents normally collect process and container metrics every 10 seconds. - When a user opened a live Processes or Containers view, all hosts in that tenant switched to 2-second collection. - This supported near-real-time monitoring similar to `htop`, but across distributed infrastructure. - As tenants grew, the pipeline had to process millions of processes per second, even though users typically viewed only around 50 processes or containers. - Live sorting required keeping all tenant data in memory on a single server, limiting horizontal scaling and forcing vertical scaling. ## Refocusing on User-Visible Data - Most collected metrics were never displayed to users. - Datadog determined that real-time collection only needed to be enabled for hosts running the processes or containers currently in view—up to roughly 50 hosts per user. - Internal telemetry suggested this could reduce traffic by more than 100x. - This required tracking active host subscriptions and updating them as users navigated the product. - Because sorting occurred every 10 seconds, it did not need 2-second data. Datadog switched live views to use the existing 10-second metrics, aligning live and historical sorting logic. ## Host Subscription Filtering - A proof of concept added host subscriptions to the live data servers. - Servers filtered Kafka payloads and discarded data for hosts without active subscriptions. - This immediately reduced: - Memory usage by 85% - CPU usage by 33% - The improvement came from storing fewer live metrics and processing fewer incoming payloads. - The prototype confirmed that filtering preserved product behavior while simplifying sorting. ## Moving Filtering Earlier in the Pipeline - Late filtering improved live data servers but still left unnecessary work for the rest of the system and customer-side Datadog Agents. - Datadog therefore planned to propagate subscription state to the intake service. - Live data servers publish users’ active host sets over Kafka once per second. - The intake service consumes this information and decides which hosts should activate 2-second process and container metric collection. - This allows real-time collection to be restricted to hosts users are actively investigating while maintaining responsive live views. Datadog’s redesign demonstrates that real-time systems scale more effectively when they prioritize data users can actually see. Filtering at intake, limiting high-frequency collection to subscribed hosts, and reusing standard-resolution data for sorting provide a simpler and more economical architecture without eliminating live functionality.

kafkadatabase-designdatadogdocker+4

datadogJun 23, 2025

How we built reliable log delivery to thousands of unpredictable endpoints (opens in new tab)

Datadog’s Log Forwarding system resembles a package delivery network: it must move large volumes of data efficiently and reliably to many unpredictable destinations. Kafka provides ordered transport, but its FIFO behavior creates difficult tradeoffs when endpoints are slow or unavailable. The central challenge is preserving delivery guarantees without losing logs, creating duplicates, blocking unrelated destinations, or overwhelming customer infrastructure. ## What Log Forwarding Does - Datadog forwards processed, enriched logs as schemaless JSON records. - Destinations can include: - Elasticsearch - Splunk - Generic HTTP endpoints accepting JSON `POST` requests - The system must support thousands of tenants and external endpoints with widely varying reliability and performance. ## Kafka as the Distribution Network - Logs move through Datadog on Kafka topics, analogous to packages traveling on conveyor belts. - Each Kafka partition provides strict FIFO ordering: - Records are read in the order they were written. - Kafka offsets must be committed in that same order. - Logs for different destinations are spread across multiple partitions, so records for a single destination may need to be regrouped during delivery. - Assigning a dedicated Kafka partition to every destination would be simple conceptually but infeasible at scale. ## Reliability Challenges - External endpoints may be: - Temporarily unavailable - Slow or unstable - Unreachable for hours or days - The system must avoid: - Losing customer logs - Sending duplicate logs - Delaying all destinations because one endpoint is unhealthy - Excessive resource usage - Overwhelming or effectively DDoSing a customer endpoint - Sending one HTTP request per log would be inefficient, so logs should be buffered and delivered in batches, much like packages going to the same address. ## Kafka Ordering and Blocked Progress - Waiting for each forwarding request to succeed before reading more Kafka data protects against data loss but can halt progress. - Continuing to read and acknowledge Kafka records before successful delivery risks losing logs. - Because offsets must be committed in order, one unavailable destination can block later records in the same partition—even if those records belong to healthy destinations. - This makes coordination between Kafka consumption, retries, batching, and concurrent delivery especially complex in a multi-tenant system. ## Lessons from Log Archives - Datadog had prior experience with similar delivery problems in its Log Archives feature. - Archiving was easier because: - Cloud object storage endpoints are generally more reliable. - Archiving has lower latency requirements. - Those lessons helped the team anticipate reliability and ordering pitfalls in Log Forwarding. ## Dedicated Kafka Topics per Destination - A possible solution would be to assign one or more Kafka partitions to each destination. - This would isolate destinations so that one slow endpoint could not block others. - However, the approach would require an impractically large number of Kafka topics or partitions as the number of customers and destinations grows.

kafkadatabase-designdatadogjson+4

datadogJun 17, 2025

How we scaled fast, reliable configuration distribution to thousands of workload containers (opens in new tab)

Datadog’s seemingly simple tenant-configuration CRUD system must propagate updates rapidly and reliably to thousands of containers processing millions of logs per second. Loading configuration on every log is too expensive, while periodic caching introduces stale data and delayed updates. Datadog initially used database-backed caches invalidated through Kafka, but growing scale exposed reliability and resilience problems tied to repeated workload access to the central database. ## The Challenge of Propagating Context Data - Datadog calls tenant-specific settings—such as log parsing rules, Sensitive Data Scanner settings, and storage quotas—“context data.” - Configuration changes are expected to take effect almost immediately, including in Live Tail. - The same context data may be consumed by thousands of containers handling traffic for many tenants. - Because configuration directly affects customer-data processing, propagation must be both low-latency and highly reliable. - The system must assume that failures can occur anywhere in a large distributed environment. ## Why On-Demand Fetching and Simple Caching Fail - Fetching configuration from a database for every incoming log would create an impractical read load. - Large tenants can generate hundreds of thousands of logs per second. - Each processing instance could require thousands of database reads per second. - Multiplying this across many instances would require extensive, highly performant database replicas. - Caching configuration in each workload container reduces reads but does not eliminate the scaling problem. - Many workload instances still cache data for a high number of tenants. - Increasing the cache interval reduces database load but delays configuration updates. - With periodic invalidation, the average propagation delay is roughly half the cache interval. ## Context Loading v1: Database-Backed Caches and Kafka Datadog’s first successful architecture kept tenant configuration in a central durable database while allowing workload containers to cache entries indefinitely. - A user changes a log-processing configuration. - The central context database stores the update. - Kafka publishes an invalidation message after the database write. - Every workload container receives the notification. - Each container reloads the affected tenant’s configuration from the database. - This minimized routine database reads while preserving low-latency updates. ## Why the Initial Architecture Needed Reconsideration - The design required every workload instance to reach the central context database whenever a configuration changed. - As Datadog added more workloads and containers, update-related database traffic grew substantially. - Internal game days and production incidents showed that problems affecting the context database could spread to downstream processing workloads. - Database failures could prevent configuration updates from propagating and potentially make it impossible for new workload containers to initialize their context. - These reliability concerns demonstrated that Kafka-based invalidation alone did not sufficiently isolate workload processing from context-database failures. Datadog’s experience shows that configuration propagation at large scale requires more than a durable database and cache invalidation. The system must also reduce dependency on the central database during updates and startup, while continuing to provide near-immediate, reliable propagation.

kafkadatabase-designdatadogobservability+4

datadogFeb 19, 2025

Achieving relentless Kafka reliability at scale with the Streaming Platform (opens in new tab)

Datadog built a Streaming Platform to make Kafka resilient and dynamically manageable across hundreds of clusters, thousands of topics, and millions of partitions. By decoupling applications from physical Kafka resources, the platform can reroute traffic, rebalance workloads, and fail over in seconds without redeployments. Its design trades strict processing order for parallelism and durability, while adding mechanisms to prevent poisoned events and backlogs from blocking entire partitions. ## A Fluid Control Plane for Multi-Cluster Orchestration - Kafka infrastructure is treated as interchangeable commodity hardware rather than a fixed application dependency. - Workloads can shift between clusters, availability zones, and Kubernetes environments automatically. - The control plane continuously replaces unhealthy components and maintains uninterrupted data flow. ### Resilient Pipelines with Streams - Streams abstract away Kafka topics and clusters behind stable identifiers. - A single Stream can span multiple Kafka clusters, availability zones, and Kubernetes clusters. - Producers and consumers do not need to know the underlying Kafka topology. - Infrastructure can be reconfigured in real time without application changes or redeployments. ### Live Traffic Failovers - When a cluster becomes unhealthy, the platform creates a replacement topic and redirects new traffic to it. - Consumers drain the backlog from the original topic before transitioning to the replacement. - The same mechanism supports proactive traffic redistribution, cluster decommissioning, capacity changes, and partition adjustments. - Operations that traditionally take hours can be completed in seconds. ### Consumer Semantics for Scale and Durability - Datadog uses at-least-once delivery rather than strict processing order. - Events can be processed independently and in parallel. - Ordering is restored later in the event store. - This trade-off enables efficient processing of petabytes of data across distributed infrastructure. ### The Assigner Coordinator - Kafka’s default group coordinator depends on session timeouts, making failover detection take tens of seconds or minutes. - Datadog’s Assigner monitors cluster health, resource usage, and workload distribution continuously. - It reacts in seconds to failures, traffic spikes, and capacity changes. - Workloads are balanced using real metrics such as CPU load and available resources, allowing heterogeneous environments to be used efficiently. ## Preventing Head-of-Line Blocking Kafka’s strict per-partition ordering means a single unprocessable event can block all later events. Datadog addresses this reliability problem through independent Stream lanes and a more flexible commit log. ### Stream Lanes and Quality of Service - Streams contain separate lanes for different priority levels and traffic requirements. - High-priority real-time traffic can be isolated from slower bursts or late-arriving data. - A dedicated dead-letter queue lane receives poison pills that cannot be processed. - Consumers can commit progress after moving failed events to the DLQ, preventing one bad event from blocking the partition without losing data. ### Advanced Commit Logging - Kafka traditionally maintains one committed pointer per partition. - That model prevents consumers from advancing past older events that are delayed or still being processed. - Datadog uses Kafka’s commit metadata to record multiple offsets or offset ranges simultaneously. - This allows consumers to continue processing live traffic while older events are handled separately, reducing backlog-related blocking. ## Overall Design Philosophy - The Streaming Platform combines stable logical Streams, real-time orchestration, flexible consumer semantics, QoS isolation, and enhanced offset tracking. - Together, these components make Kafka more self-healing and suitable for Datadog’s extreme scale. - The approach prioritizes uninterrupted processing, rapid recovery, and operational flexibility over Kafka’s default assumptions of fixed topology and strict ordering. Datadog’s design demonstrates that Kafka at very large scale requires a control layer around the broker infrastructure. Abstracting resources, automating failovers, and allowing independent progress through partitions can provide substantially better reliability and throughput than relying solely on Kafka’s native coordination model.

kafkadatabase-designk8sload-balancing+4

datadogJun 28, 2024

Timeseries indexing at scale (opens in new tab)

Datadog’s metrics volume grew 30× from 2017 to 2022, while customers began running increasingly complex queries. This growth exposed limitations in the Timeseries Index service, whose original indexing approach became a performance and maintenance bottleneck. The post introduces Datadog’s metrics architecture and explains how its indexing strategy evolved to handle large-scale workloads more reliably. ## Metrics Platform Architecture - **Intake** - Datadog Agents send data points through a load balancer to metrics intake. - Each point contains a metric name, timestamp, numerical value, and optional tags. - Tags such as `env`, `host`, and `service` provide dimensions for filtering, aggregation, and comparison. - Data is written to Kafka, allowing multiple consumers to process it for storage, indexing, analysis, and archiving. - **Storage** - The short-term storage layer has two services: - The Timeseries Database stores tuples of `<timeseries_id, timestamp, float64>`. - The Timeseries Index stores `<timeseries_id, tags>` mappings. - The custom Timeseries Index database is built on RocksDB and supports filtering and grouping during queries. - **Query Processing** - The distributed query layer contacts index nodes, retrieves intermediate results from the timeseries database, and combines them. - Filters such as `env:prod AND service:event-consumer` restrict results to matching data points. - Grouping by tags, such as `service`, produces separate timeseries for each group. - Aggregators such as `avg` combine values within each group. ## Why Timeseries Indexing Matters - Indexes prevent queries from scanning every timeseries associated with a metric, much like database indexes avoid full table scans. - Poorly designed or insufficient indexes can make queries slow and consume excessive CPU and memory. - As Datadog’s data volume and query complexity increased, the indexing system became a critical scalability concern. ## Automatically Generated Indexes - The original system generated indexes from live query behavior. - Slow or resource-intensive queries were recorded in a query log and analyzed periodically. - Index selection considered: - Query frequency - Execution time - Number of input timeseries identifiers scanned - Number of output identifiers returned - Highly selective queries—with a high input-to-output ratio—received indexes. - Obsolete indexes that no longer received queries were removed. - These indexes acted as materialized views, replacing expensive scans with efficient key-value lookups. ## Original Indexing Service Design - The service was written in Go and used embedded SQLite and RocksDB databases. - SQLite stored metadata, including: - Index definitions - Query logs - Query counts and timestamps - Input and output cardinalities - Query durations - Index definitions were read frequently, updated rarely, and cached entirely in memory. - Query logs were bulk-written in the background, keeping them out of the ingestion and query paths. - SQLite’s SQL interface made the metadata easy to inspect and modify manually. - RocksDB handled the high-volume write workload required to index trillions of events per day. Datadog’s experience shows that indexing strategies that work at smaller scale can become bottlenecks as data volume and query sophistication grow. Effective timeseries systems therefore need adaptive indexing, careful separation of query and ingestion workloads, and storage technologies suited to extremely high write rates.

kafkadatabase-designdatadoggo+4

datadogFeb 22, 2023

Husky: Exactly-once ingestion and multi-tenancy at scale (opens in new tab)

Husky, Datadog’s distributed, time-series-oriented event store, is optimized for large scans and aggregations rather than high-volume, low-latency point lookups. This makes exactly-once ingestion challenging, especially at Datadog’s multi-tenant scale. Datadog addresses the problem with deterministic, locality-aware routing that limits deduplication scope, improves storage efficiency, and supports autoscaling ingestion pipelines. ## Husky’s Ingestion Challenge - Husky separates storage and compute, allowing each to scale independently. - Its storage engine is designed primarily for large analytical scans and aggregations. - It is not optimized for massive numbers of low-latency point lookups, complicating duplicate detection during ingestion. - The ingestion system must guarantee that every event is stored exactly once while maintaining: - Multi-tenant scalability - Reasonable ingestion latency - Controlled infrastructure and storage costs ## Routing Events to Storage Shards - Datadog uses an upstream **Shard Router** to introduce locality into Kafka pipelines. - Events are deterministically assigned to shards based on their tenant, timestamp, and event ID. - Each tenant receives a list of shards rather than being permanently assigned to one shard. - A deterministic choice from that list distributes the tenant’s events while keeping the number of active shards as small as practical. - Downstream workers consume one or more shards and perform exactly-once ingestion into Husky. ## Benefits of Data Locality - **Simpler deduplication** - An event with the same timestamp and ID always reaches the same shard. - Deduplication only needs to occur within that shard. - Workers handle smaller sets of event IDs, making in-memory deduplication more efficient. - **Lower storage costs and better performance** - Each shard processes a relatively small set of tenants. - Husky stores each tenant in a separate table and does not mix tenants within files. - More tenants per writer produce more output files, increasing blob-storage costs and compaction work. - Restricting tenant cardinality reduces file creation and improves writer and compactor efficiency. ## Challenges in Deterministic Routing - **Changing shard assignments** - Tenant traffic can increase by one or two orders of magnitude. - Assignments may change when scaling a tenant across more shards or rebalancing traffic among existing shards. - **Distributed router consensus** - Every Shard Router node must make the same routing decision for a given event. - Inconsistent decisions could send duplicates to different shards and undermine exactly-once ingestion. - **Load balancing** - Shards must receive roughly equal traffic so downstream ingestion workers remain balanced. ## Time-Bounded Shard Placements - A simple deterministic mapping can select a shard using a hash of the event ID: ```text shard = shards[hash(event_id) % num_shards] ``` - This approach is cheap and stateless when all routers know the same shard list. - However, changing the shard list can cause the same event ID to map to a different shard, so assignment changes require additional coordination. - The article introduces **time-bounded Shard Placements** to preserve consistent routing while allowing tenant assignments to evolve, though the supplied excerpt ends before explaining the mechanism in detail. Datadog’s core recommendation is to combine deterministic, tenant-aware routing with carefully coordinated assignment changes. This narrows the scope of deduplication while reducing storage overhead and enabling balanced, scalable exactly-once ingestion.

kafkadatabase-designdistributed-systemsload-balancing+4

datadogMay 17, 2022

Introducing Husky, Datadog's third-generation event store (opens in new tab)

Datadog built Husky, a new event-storage system, after its original log architecture struggled with multi-tenant reliability, rapid platform growth, and evolving product requirements. The post explains how Datadog moved from metrics-oriented storage to event storage, introduced custom sharding and routing, and eventually recognized the need for a more flexible system. Husky emerged from these lessons about isolation, scalability, and retaining high-cardinality event data. ## From Metrics to Logs - Metrics systems store pre-aggregated tuples such as `<timeseries_id, timestamp, float64>`. - Aggregation makes metrics extremely efficient: millions of events in a second can become one compact datapoint, often requiring less than two bytes with delta-of-delta encoding. - This model is poorly suited to logs because logs must preserve individual events and their full context. - Metrics typically favor long-lived, low-cardinality dimensions such as: - Datacenter - Service - Pod name - Short-lived, high-cardinality fields such as transaction IDs and packet IDs are usually pre-aggregated or omitted. - Logs instead need to support: - Multi-kilobyte events - High-cardinality values such as UUIDs and stack traces - Arbitrary aggregations performed at query time ## Limitations of the Initial Logs System - Datadog’s first Logs architecture initially worked well but became vulnerable in a multi-tenant environment. - A single unhealthy or overloaded node could degrade service for every tenant in the cluster. - Scaling overloaded clusters could worsen the situation because nodes began streaming data to one another while already handling excessive read and write workloads. - Diagnosing and mitigating these cascading failures was difficult. ## Separating Storage from Clustering Datadog’s second architecture retained the same single-node storage engine but moved clustering responsibilities into dedicated services. - Storage nodes no longer knew about one another and behaved like independent one-node clusters. - Failures were isolated to the tenants assigned to a particular shard instead of spreading across the entire cluster. - A Shard Router: - Read events from Kafka - Reorganized them into shard-based Kafka partitions - Dynamically assigned tenants to an appropriate number of shards based on their recent five-minute data volume - Each shard was consumed by two storage-node replicas for redundancy. - A custom query engine tracked tenant-to-shard assignments, queried the relevant replicas, merged partial aggregates, and produced final results. ## Growth of the Event Platform - The new architecture substantially improved reliability and reduced operational burden. - Datadog expanded the platform beyond Logs to support products including: - Network Performance Monitoring - Real User Monitoring - Continuous Profiler - These products generated structured, multi-kilobyte events with storage and indexing requirements similar to logs. - As usage grew, new problems appeared: - A tenant producing a sudden burst of events could degrade query performance for other tenants sharing its shard. - Product teams requested longer retention for important but infrequently queried data, while still requiring it to remain immediately queryable. - The existing architecture was increasingly difficult to adapt to these isolation, scalability, and retention requirements, motivating the development of Husky. Datadog’s progression shows that event storage cannot simply reuse metrics-oriented designs. Systems must preserve event-level context, isolate tenants from one another, and support changing retention and query requirements as products and workloads evolve.

kafkadatabase-designdatadogmetrics+4

datadogAug 13, 2018

Introducing Kafka-Kit: Tools for scaling Kafka (opens in new tab)

Datadog operates Kafka at extreme scale, ingesting trillions of data points daily and requiring petabytes of NVMe storage. To manage frequent data movement caused by scaling, recovery, and capacity changes, the company built Kafka-Kit, a set of operational tools that improve partition placement and replication control. Its primary tools, `topicmappr` and `autothrottle`, automate safer and more predictable Kafka operations. ## Kafka-Kit - Kafka-Kit addresses two major operational areas: - Data placement across brokers - Replication auto-throttling - Its main tools are: - `topicmappr`, for generating partition-to-broker mappings - `autothrottle`, for automatically controlling replication bandwidth ## Partition Placement with `topicmappr` `topicmappr` replaces Kafka’s `kafka-reassign-partitions.sh --generate` functionality while adding operational safeguards and placement controls. - Produces deterministic output: identical inputs generate the same partition map. - Supports minimal-movement broker replacement: - Failed brokers can be replaced without unnecessarily moving healthy partitions. - Partitions with complete in-sync replicas are normally left untouched. - Provides rack-aware placement using Kafka’s `broker.rack` metadata and ZooKeeper. - Supports placement based on: - Partition count - Storage size, enabling bin-packing and storage rebalancing - Allows replication factors to be increased or decreased while topics are running. - Generates clear summaries of: - Brokers being removed or added - Partition-level changes - Broker distribution before and after reassignment - Warnings and resulting partition-map files The tool is written in Go and can run from any system with access to Kafka’s ZooKeeper cluster. It requires topic names and broker IDs, then verifies that the brokers are live, sufficiently numerous, and properly distributed across configured localities. ## Replacing Failed Brokers For a failed broker, `topicmappr` can rebuild affected topics while limiting movement to the necessary partitions. - Existing replicas are preserved whenever possible. - Replacement brokers fill the gaps left by failed brokers. - The generated report makes the proposed changes visible before execution. - The example replaces broker `1002` with brokers `1003` and `1004`, showing the updated replica assignments and broker totals. ## Placement Strategies `topicmappr` offers multiple strategies for deciding where replicas should live, including `count` and tunable `storage` placement. ### Count Placement Strategy - The default strategy. - Balances leadership and the number of partitions held by each broker. - Works well when traffic is expected to be distributed evenly across partitions. - Does not require metrics data, allowing maps to be generated quickly. - Also attempts to maximize the number of distinct broker-to-broker replica relationships. - This avoids concentrating a broker’s partitions with the same small subset of peers, improving distribution across the cluster and its racks. ## Storage-Aware Placement - The storage strategy uses partition size when assigning replicas. - It supports storage bin-packing and rebalancing, which is important when brokers have uneven disk utilization. - This is particularly useful for Datadog’s large Kafka clusters, where storage capacity—not just partition count—can determine when data must be moved. Datadog’s approach demonstrates that Kafka’s flexible primitives can be extended with purpose-built tooling. For large deployments, deterministic assignments, rack awareness, minimal movement, and storage-based balancing can make scaling and failure recovery substantially safer and more predictable.

kafkadatabase-designzookeeperkafka-kit+4