Making Metrics Queryable at Scale

Building a system that can query metrics across every host and service, over any time period since inception, took about a day of engineering work. The immediate payoff was a mid-seven-figure optimization that shipped, and within a year the system had driven mid-eight-figures in annual cost savings. Internally, it's called LongTermMetrics (LTM), a deliberately boring name for a system that does one useful thing: it makes historical metrics data accessible to arbitrary SQL queries.

The project started with a simple question about JVM survivor space utilization. Survivor space acts as a fixed-size buffer in the JVM's default garbage collector. For any individual large service, the buffer is often either too small—causing performance problems under load—or too large, wasting memory and money. The obvious step was to check every service at once and rank them by potential improvement. That turned out to be harder than expected, primarily because the existing metrics infrastructure wasn't built for general queries.

The Data Problem

Twitter's MetricsDB was designed for monitoring, dashboards, and alerts, not arbitrary analytics. Running general SQL queries against it wasn't an option. The metrics data did get copied to HDFS in Parquet format for batch processing, but there were two serious limitations:

  • A single metric can have dozens of names—JVM survivor space alone had 94 variants found by a simple search.
  • The same metric name could mean different things for different services: counter vs. gauge, bytes vs. MB, microseconds vs. milliseconds.

Beyond naming inconsistencies, metrics could be tagged with incorrect service names, and zombie shards—old instances still reporting after being replaced—created duplicate, conflicting data. Even when the data was clean enough to use, querying it all was impractical: the sheer number of metric names prevented a columnar format with one column per name, since Presto can't handle that many columns. Without columnar storage, non-trivial queries timed out, and the alternative MapReduce jobs via Scalding took anywhere from three to twenty hours per day of data—far too slow for exploratory analysis.

The solution was a filtering job. Most metrics—perhaps 90% or more—are rarely useful to anyone. The ones that matter for performance and capacity analysis are a much smaller set, roughly 0.1% to 0.01% of the total. A Scalding job was written to extract that relevant subset and rewrite it into a proper columnar format, making it queryable with Presto. A side benefit: this filtered dataset is small enough to store indefinitely, unlike the full metrics dump, which is deleted after a few weeks due to storage costs. Long-term history becomes available for capacity planning and trend analysis.

What Gets Stored

The dataset captures, for each shard of each service:

  • Buffer sizes and utilizations
  • CPU, memory, and other resource utilization
  • Thread counts, context switches, core migrations
  • Queue depths and network statistics
  • JVM version, feature flags, and configuration
  • Garbage collection statistics
  • Finagle metrics

Host-level data includes procfs metrics like iowait and idle time, cluster membership, hardware details such as NIC speed and core count, health indicators like thermal throttling and machine checks, and OS/software versions. Rezolus metrics are also included. Most of this is stored at the same granularity as the original metrics, though infrequently-changing host properties are recorded daily instead.

Finding Misconfigurations with SQL

Once the data is in a queryable format, identifying problems becomes straightforward. Finding services with very high p90 survivor space utilization—a strong signal of under-provisioning—is a simple SQL query across all services:

with results as (
  select servicename,
    approx_distinct(source, 0.1) as approx_sources, -- number of shards for the service
    -- real query uses [coalesce and nullif](https://prestodb.io/docs/current/functions/conditional.html) to handle edge cases, omitted for brevity
    approx_percentile(jvmSurvivorUsed / jvmSurvivorMax, 0.90) as p90_used,
    approx_percentile(jvmSurvivorUsed / jvmSurvivorMax, 0.50) as p50_used,
  from ltm_service 
  where ds >= '2020-02-01' and ds <= '2020-02-28'
  group by servicename)
select * from results
where approx_sources > 100
order by p90_used desc

This query pattern works for any utilization metric or rate: memory usage, GC frequency, queue depths, and more. The output can feed directly into scripts that generate configuration-change diffs for the relevant teams. This approach uncovered a single service wasting enough RAM to cover a decade of a senior engineer's salary.

Percentile thresholds are a reasonable starting point, though more sophisticated analysis is possible once the infrastructure exists.

Cross-Cutting Analyses and Rollout Monitoring

The same infrastructure supports queries across hosts, not just services, and can join against host properties and feature flags. When the compute platform team considered changing network egress limits, LTM provided the data to assess the risk: many services were hitting network caps even when host-level network utilization appeared low. During the gradual rollout of higher caps, queries monitored queue depths to confirm there was no performance degradation. The query below checks queue depths, and extending it to queue size in bytes or packet drops is just a matter of adding more metric names.

with rolled as (
 select
   -- rollout was fixed for all hosts during the time period, can pick an arbitrary element from the time period
   arbitrary(element_at(misc, 'egress_rate_limit_increase')) as rollout,
   hostId
 from ltm_deploys
 where ds = '2019-10-10'
 and zone = 'foo'
 group by ipAddress
), host_info as(
 select
   arbitrary(nicSpeed) as nicSpeed,
   hostId
 from ltm_host
 where ds = '2019-10-10'
 and zone = 'foo'
 group by ipAddress
), host_rolled as (
 select
   rollout,
   nicSpeed,
   rolled.hostId
 from rolled
 join host_info on rolled.ipAddress = host_info.ipAddress
), container_metrics as (
 select
   service,
   netTxQlen,
   hostId
 from ltm_container
 where ds >= '2019-10-10' and ds <= '2019-10-14'
 and zone = 'foo'
)
select
 service,
 nicSpeed,
 approx_percentile(netTxQlen, 1, 0.999, 0.0001) as p999_qlen,
 approx_percentile(netTxQlen, 1, 0.99, 0.001) as p99_qlen,
 approx_percentile(netTxQlen, 0.9) as p90_qlen,
 approx_percentile(netTxQlen, 0.68) as p68_qlen,
 rollout,
 count(*) as cnt
from container_metrics
join host_rolled on host_rolled.hostId = container_metrics.hostId
group by service, nicSpeed, rollout

That rollout successfully reduced queue depths and packet drops. The general capability is significant: any platform or service-level feature change can be evaluated against real service metrics across the entire fleet.

Additional Questions the System Makes Easy

  • What's the latency, CPU, or IPC impact of a kernel change, an OS release, or a change in oversubscription?
  • What is the distribution of a given metric across the fleet, in general or under specific conditions?
  • Which hosts show poor performance for every service running on them, after accounting for load? (Often a hardware fault or misconfiguration.)
  • Which services degrade their neighbors' performance beyond general host-level load?
  • What is the latency impact of a failover event, and what load levels should be expected after considering current growth?
  • For any fixed-size buffer, what configuration change improves performance without costing more, or reduces cost without degrading performance?
  • If a host-level health problem appears N times, what's the probability of recurrence?

Deliberately Boring Design Decisions

Every choice in LTM was the path of least resistance. It uses Scalding because it's standard at Twitter, and the integration was free. Presto was chosen over an analytics database like Druid because joins with other datasets are trivial in Presto; a contemporaneous project on tracing analytics was expected to benefit from the same query layer. Druid's interactive query features looked attractive, but the join story would have required significant planning.

A traditional relational database was out of the question—the data volume, even filtered, is a big-data problem. Parquet was chosen over a more bespoke format because it was the most suitable of the formats already supported by the infrastructure. Introducing a new format would have ballooned the project's scope.

The system is batch, not real-time, with at least an hour of latency. This was a deliberate trade-off:

  • The batch pipeline was already built, requiring only a day to learn and adapt.
  • Real-time analytics would have taken one to four quarters for an MVP, requiring headcount and roadmap commitments.
  • Misconfigurations and optimization opportunities are rarely transient; they persist for months or years.
  • Batch processing has negligible operational overhead—about one to two days of engineering time per month—and never pages anyone.

This approach contrasts with a similar effort at another company that started with deep learning instead of simple heuristics. The former returned significant value in days; the latter took person-years. Deep learning has better optics for resumes, promotions, and conference talks, but the ROI of boring, practitioner-guided analysis is far higher as a starting point.

Mistakes Worth Admitting

A few decisions made in the interest of speed created avoidable problems later:

  • Not using doubles universally. The metrics pipeline converts everything to double at some stage, so storing integral types as longs causes precision loss on the round trip. Some small values were stored as integers, losing precision unnecessarily. Fixing this requires a full data migration, estimated at half a week to a week of work—which thirty seconds of forethought would have avoided.
  • Using asserts instead of error logging. When an unexpected input arrived—like a clock-skew incident of 3600 years causing a timestamp overflow—the entire job crashed. Writing bad entries to a separate table, and alerting if it's non-empty, would have preserved correct output and avoided the operational interruption.
  • Inconsistent capitalization. Longterm vs. LongTerm in the codebase. Trivial, but a grep and cleanup at the time would have prevented the pointless inconsistency.

The scalability of the original design isn't aspirational—it's proven. A filter job that extracts a fraction of a percent of metrics data, stored in a columnar format and queried with SQL, has driven eight figures in annual savings at Twitter's scale. The most impressive part might be how unimpressive each individual piece is, and how much value can be unlocked by connecting them.