- Refactor metric reporting for improved cardinality and aggregation logic

This commit message was generated with the help of LLMs.

Author: nateberkopec

README.md

@@ -168,7 +168,7 @@ We report the following metrics:
 QueueLatency - Time job spent waiting in queue before execution (Seconds)
 ```
 
-This metric includes JobClass and QueueName dimensions.
+This metric includes QueueName dimension and is aggregated per interval using CloudWatch StatisticSets.
 
 ### Rails
 
@@ -218,3 +218,9 @@ If for some reason you want to disable an integration _after_ this registration,
 ```ruby
 
 ```
+
+## Comparisons and Design Notes
+
+For a deep comparison with related libraries and concrete recommendations informed by their approaches, see:
+
+- docs/cloudwatch_libs_comparison.md

docs/metrics.md

@@ -132,6 +132,6 @@ This document explains each metric collected by Speedshop::Cloudwatch, how it's
 
 ### QueueLatency
 - **Unit:** Seconds
-- **Dimensions:** JobClass, QueueName
+- **Dimensions:** QueueName
 - **Source:** `Time.now.to_f - job.enqueued_at`
-- **Description:** Time a job spent waiting in the queue before execution started. Measured when the job begins executing. High latency indicates jobs are backing up for this job class and queue combination.
+- **Description:** Time a job spent waiting in the queue before execution started. Measured when the job begins executing. Values are aggregated per queue into CloudWatch StatisticSets for each reporting interval.

lib/speedshop/cloudwatch/active_job.rb

@@ -11,7 +11,8 @@ def report_job_metrics
         begin
           if enqueued_at
             queue_time = Time.now.to_f - enqueued_at.to_f
-            Reporter.instance.report(metric: :QueueLatency, value: queue_time, dimensions: {JobClass: self.class.name, QueueName: queue_name}, integration: :active_job)
+            # Drop JobClass to reduce time series cardinality and allow aggregation into StatisticSets per queue
+            Reporter.instance.report(metric: :QueueLatency, value: queue_time, dimensions: {QueueName: queue_name}, integration: :active_job)
           end
         rescue => e
           Speedshop::Cloudwatch.log_error("Failed to collect ActiveJob metrics: #{e.message}", e)

lib/speedshop/cloudwatch/puma.rb

@@ -12,6 +12,7 @@ def collect
               metric_name = m.to_s.split("_").map(&:capitalize).join.to_sym
               Reporter.instance.report(metric: metric_name, value: stats[m] || 0)
             end
+            report_aggregate_worker_stats(stats)
           end
 
           workers = stats[:worker_status] ? worker_statuses(stats) : [[stats, 0]]
@@ -32,6 +33,33 @@ def report_worker(stats, idx)
             Reporter.instance.report(metric: metric_name, value: stats[m] || 0, dimensions: {WorkerIndex: idx.to_s})
           end
         end
+
+        def report_aggregate_worker_stats(stats)
+          statuses = stats[:worker_status].map { |w| w[:last_status] || {} }
+          metrics = %i[running backlog pool_capacity max_threads]
+
+          metrics.each do |m|
+            values = statuses.map { |s| s[m] }.compact
+            next if values.empty?
+
+            sample_count = values.length
+            sum = values.inject(0) { |acc, v| acc + v.to_f }
+            minimum = values.min.to_f
+            maximum = values.max.to_f
+
+            metric_name = m.to_s.split("_").map(&:capitalize).join.to_sym
+            Reporter.instance.report(
+              metric: metric_name,
+              statistic_values: {
+                sample_count: sample_count,
+                sum: sum,
+                minimum: minimum,
+                maximum: maximum
+              },
+              integration: :puma
+            )
+          end
+        end
       end
     end
   end

lib/speedshop/cloudwatch/reporter.rb

@@ -54,7 +54,7 @@ def stop!
         thread_to_join&.join
       end
 
-      def report(metric:, value:, dimensions: {}, namespace: nil, integration: nil)
+      def report(metric:, value: nil, statistic_values: nil, dimensions: {}, namespace: nil, integration: nil)
         metric_name = metric.to_sym
 
         int = integration || find_integration_for_metric(metric_name)
@@ -68,11 +68,16 @@ def report(metric:, value:, dimensions: {}, namespace: nil, integration: nil)
         dimensions_array = dimensions.map { |k, v| {name: k.to_s, value: v.to_s} }
         all_dimensions = dimensions_array + custom_dimensions
 
-        @mutex.synchronize do
-          @queue << {metric_name: metric_name.to_s, value: value, namespace: ns, unit: unit,
-                     dimensions: all_dimensions, timestamp: Time.now}
+        datum = {metric_name: metric_name.to_s, namespace: ns, unit: unit,
+                 dimensions: all_dimensions, timestamp: Time.now}
+        if statistic_values
+          datum[:statistic_values] = statistic_values
+        else
+          datum[:value] = value
         end
 
+        @mutex.synchronize { @queue << datum }
+
         start! unless started?
       end
 
@@ -137,15 +142,99 @@ def flush_metrics
         metrics = @mutex.synchronize { @queue.empty? ? nil : @queue.dup.tap { @queue.clear } }
         return unless metrics
 
+        high_resolution = config.interval.to_i < 60
+
         metrics.group_by { |m| m[:namespace] }.each do |namespace, ns_metrics|
           config.logger.debug "Speedshop::Cloudwatch: Sending #{ns_metrics.size} metrics to namespace #{namespace}"
-          metric_data = ns_metrics.map { |m| m.slice(:metric_name, :value, :unit, :timestamp, :dimensions) }
-          config.client.put_metric_data(namespace: namespace, metric_data: metric_data)
+
+          aggregated = aggregate_namespace_metrics(ns_metrics)
+
+          metric_data = aggregated.map do |m|
+            datum = {
+              metric_name: m[:metric_name],
+              unit: m[:unit],
+              timestamp: m[:timestamp],
+              dimensions: m[:dimensions]
+            }
+            if m[:statistic_values]
+              datum[:statistic_values] = m[:statistic_values]
+            else
+              datum[:value] = m[:value]
+            end
+            datum[:storage_resolution] = 1 if high_resolution
+            datum
+          end
+
+          metric_data.each_slice(20) do |batch|
+            config.client.put_metric_data(namespace: namespace, metric_data: batch)
+          end
         end
       rescue => e
         Speedshop::Cloudwatch.log_error("Failed to send metrics: #{e.message}", e)
       end
 
+      def aggregate_namespace_metrics(ns_metrics)
+        # Group by metric_name + unit + dimensions (order-insensitive) to aggregate within a flush window.
+        groups = {}
+
+        ns_metrics.each do |m|
+          key = [m[:metric_name], m[:unit], normalized_dimensions_key(m[:dimensions])]
+          (groups[key] ||= []) << m
+        end
+
+        aggregated = []
+        groups.each do |(_name, _unit, _dims_key), items|
+          # If only one item and it already has value/statistic_values, keep as-is
+          if items.size == 1
+            aggregated << items.first
+            next
+          end
+
+          # Aggregate values and statistic_values into a single StatisticSet
+          sample_count = 0.0
+          sum = 0.0
+          minimum = Float::INFINITY
+          maximum = -Float::INFINITY
+
+          items.each do |it|
+            if it[:statistic_values]
+              sv = it[:statistic_values]
+              sc = sv[:sample_count].to_f
+              sample_count += sc
+              sum += sv[:sum].to_f
+              minimum = [minimum, sv[:minimum].to_f].min
+              maximum = [maximum, sv[:maximum].to_f].max
+            elsif it.key?(:value)
+              v = it[:value].to_f
+              sample_count += 1.0
+              sum += v
+              minimum = [minimum, v].min
+              maximum = [maximum, v].max
+            end
+          end
+
+          aggregated << {
+            metric_name: items.first[:metric_name],
+            unit: items.first[:unit],
+            dimensions: items.first[:dimensions],
+            # Use flush time to represent the interval
+            timestamp: Time.now,
+            statistic_values: {
+              sample_count: sample_count,
+              sum: sum,
+              minimum: minimum.finite? ? minimum : 0.0,
+              maximum: maximum.finite? ? maximum : 0.0
+            }
+          }
+        end
+
+        aggregated
+      end
+
+      def normalized_dimensions_key(dims)
+        (dims || []).sort_by { |d| d[:name].to_s }.map { |d| "#{d[:name]}=#{d[:value]}" }.join("|")
+      end
+
       def metric_allowed?(integration, metric_name)
         config.metrics[integration].include?(metric_name.to_sym)
       end