Metric summaries on span

* add a `LocalAggregator` instance on spans to duplicate metrics on the
  span as a gauge metric
* proxy the main aggregator add calls to the local aggregator if a span
  is running
* start a `metrics.timing` span in the `Sentry::Metrics.timing` API

Author: sl0thentr0py

CHANGELOG.md

@@ -8,6 +8,7 @@
 - Add [Metrics](https://docs.sentry.io/product/metrics/) support
   - Add main APIs and `Aggregator` thread [#2247](https://github.com/getsentry/sentry-ruby/pull/2247)
   - Add `Sentry::Metrics.timing` API for measuring block duration [#2254](https://github.com/getsentry/sentry-ruby/pull/2254)
+  - Add metric summaries on spans [#2255](https://github.com/getsentry/sentry-ruby/pull/2255)
 
     The SDK now supports recording and aggregating metrics. A new thread will be started
     for aggregation and will flush the pending data to Sentry every 5 seconds.
@@ -39,6 +40,7 @@
     Sentry::Metrics.set('user_view', 'jane')
 
     # timing - measure duration of code block, defaults to seconds
+    # will also automatically create a `metric.timing` span
     Sentry::Metrics.timing('how_long') { sleep(1) }
     # timing - measure duration of code block in other duraton units
     Sentry::Metrics.timing('how_long_ms', unit: 'millisecond') { sleep(0.5) }

sentry-ruby/lib/sentry/metrics.rb

@@ -14,6 +14,8 @@ module Metrics
     INFORMATION_UNITS = %w[bit byte kilobyte kibibyte megabyte mebibyte gigabyte gibibyte terabyte tebibyte petabyte pebibyte exabyte exbibyte]
     FRACTIONAL_UNITS = %w[ratio percent]
 
+    OP_NAME = 'metric.timing'
+
     class << self
       def increment(key, value = 1.0, unit: 'none', tags: {}, timestamp: nil)
         Sentry.metrics_aggregator&.add(:c, key, value, unit: unit, tags: tags, timestamp: timestamp)
@@ -32,15 +34,18 @@ def gauge(key, value, unit: 'none', tags: {}, timestamp: nil)
       end
 
       def timing(key, unit: 'second', tags: {}, timestamp: nil, &block)
-        return unless Sentry.metrics_aggregator
         return unless block_given?
         return unless DURATION_UNITS.include?(unit)
 
-        start = Timing.send(unit.to_sym)
-        yield
-        value = Timing.send(unit.to_sym) - start
+        Sentry.with_child_span(op: OP_NAME, description: key) do |span|
+          tags.each { |k, v| span.set_tag(k, v.is_a?(Array) ? v.join(', ') : v.to_s) } if span
+
+          start = Timing.send(unit.to_sym)
+          yield
+          value = Timing.send(unit.to_sym) - start
 
-        Sentry.metrics_aggregator.add(:d, key, value, unit: unit, tags: tags, timestamp: timestamp)
+          Sentry.metrics_aggregator&.add(:d, key, value, unit: unit, tags: tags, timestamp: timestamp)
+        end
       end
     end
   end

sentry-ruby/lib/sentry/metrics/aggregator.rb

@@ -59,15 +59,23 @@ def add(type,
         serialized_tags = serialize_tags(get_updated_tags(tags))
         bucket_key = [type, key, unit, serialized_tags]
 
-        @mutex.synchronize do
+        added = @mutex.synchronize do
           @buckets[bucket_timestamp] ||= {}
 
-          if @buckets[bucket_timestamp][bucket_key]
-            @buckets[bucket_timestamp][bucket_key].add(value)
+          if (metric = @buckets[bucket_timestamp][bucket_key])
+            old_weight = metric.weight
+            metric.add(value)
+            metric.weight - old_weight
           else
-            @buckets[bucket_timestamp][bucket_key] = METRIC_TYPES[type].new(value)
+            metric = METRIC_TYPES[type].new(value)
+            @buckets[bucket_timestamp][bucket_key] = metric
+            metric.weight
           end
         end
+
+        # for sets, we pass on if there was a new entry to the local gauge
+        local_value = type == :s ? added : value
+        process_span_aggregator(bucket_key, local_value)
       end
 
       def flush(force: false)
@@ -180,6 +188,14 @@ def get_updated_tags(tags)
 
         updated_tags
       end
+
+      def process_span_aggregator(key, value)
+        scope = Sentry.get_current_scope
+        return nil unless scope && scope.span
+        return nil if scope.transaction_source_low_quality?
+
+        scope.span.metrics_local_aggregator.add(key, value)
+      end
     end
   end
 end

sentry-ruby/lib/sentry/metrics/local_aggregator.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Metrics
+    class LocalAggregator
+      # exposed only for testing
+      attr_reader :buckets
+
+      def initialize
+        @buckets = {}
+      end
+
+      def add(key, value)
+        if @buckets[key]
+          @buckets[key].add(value)
+        else
+          @buckets[key] = GaugeMetric.new(value)
+        end
+      end
+
+      def to_hash
+        return nil if @buckets.empty?
+
+        @buckets.map do |bucket_key, metric|
+          type, key, unit, tags = bucket_key
+
+          payload_key = "#{type}:#{key}@#{unit}"
+          payload_value = {
+            tags: deserialize_tags(tags),
+            min: metric.min,
+            max: metric.max,
+            count: metric.count,
+            sum: metric.sum
+          }
+
+          [payload_key, payload_value]
+        end.to_h
+      end
+
+      private
+
+      def deserialize_tags(tags)
+        tags.inject({}) do |h, tag|
+          k, v = tag
+          old = h[k]
+          # make it an array if key repeats
+          h[k] = old ? (old.is_a?(Array) ? old << v : [old, v]) : v
+          h
+        end
+      end
+    end
+  end
+end

sentry-ruby/lib/sentry/span.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require "securerandom"
+require "sentry/metrics/local_aggregator"
 
 module Sentry
   class Span
@@ -149,7 +150,7 @@ def to_baggage
 
     # @return [Hash]
     def to_hash
-      {
+      hash = {
         trace_id: @trace_id,
         span_id: @span_id,
         parent_span_id: @parent_span_id,
@@ -161,6 +162,11 @@ def to_hash
         tags: @tags,
         data: @data
       }
+
+      summary = metrics_summary
+      hash[:_metrics_summary] = summary if summary
+
+      hash
     end
 
     # Returns the span's context that can be used to embed in an Event.
@@ -268,5 +274,14 @@ def set_data(key, value)
     def set_tag(key, value)
       @tags[key] = value
     end
+
+    # Collects gauge metrics on the span for metric summaries.
+    def metrics_local_aggregator
+      @metrics_local_aggregator ||= Sentry::Metrics::LocalAggregator.new
+    end
+
+    def metrics_summary
+      @metrics_local_aggregator&.to_hash
+    end
   end
 end

sentry-ruby/lib/sentry/transaction_event.rb

@@ -17,6 +17,9 @@ class TransactionEvent < Event
     # @return [Hash, nil]
     attr_accessor :profile
 
+    # @return [Hash, nil]
+    attr_accessor :metrics_summary
+
     def initialize(transaction:, **options)
       super(**options)
 
@@ -29,6 +32,7 @@ def initialize(transaction:, **options)
       self.tags = transaction.tags
       self.dynamic_sampling_context = transaction.get_baggage.dynamic_sampling_context
       self.measurements = transaction.measurements
+      self.metrics_summary = transaction.metrics_summary
 
       finished_spans = transaction.span_recorder.spans.select { |span| span.timestamp && span != transaction }
       self.spans = finished_spans.map(&:to_hash)
@@ -49,6 +53,7 @@ def to_hash
       data[:spans] = @spans.map(&:to_hash) if @spans
       data[:start_timestamp] = @start_timestamp
       data[:measurements] = @measurements
+      data[:_metrics_summary] = @metrics_summary if @metrics_summary
       data
     end
 

sentry-ruby/spec/sentry/client_spec.rb

@@ -195,6 +195,16 @@ def sentry_context
       event = subject.event_from_transaction(transaction)
       expect(event.contexts).to include({ foo: { bar: 42 } })
     end
+
+    it 'adds metric summary on transaction if any' do
+      key = [:c, 'incr', 'none', []]
+      transaction.metrics_local_aggregator.add(key, 10)
+      hash = subject.event_from_transaction(transaction).to_hash
+
+      expect(hash[:_metrics_summary]).to eq({
+        'c:incr@none' => { count: 1, max: 10.0, min: 10.0, sum: 10.0, tags: {} }
+      })
+    end
   end
 
   describe "#event_from_exception" do

sentry-ruby/spec/sentry/metrics/aggregator_spec.rb

@@ -187,6 +187,57 @@
       expect(metric).to be_a(Sentry::Metrics::SetMetric)
       expect(metric.value).to eq(Set[1])
     end
+
+    describe 'local aggregation for span metric summaries' do
+      it 'does nothing without an active scope span' do
+        expect_any_instance_of(Sentry::Metrics::LocalAggregator).not_to receive(:add)
+        subject.add(:c, 'incr', 1)
+      end
+
+      context 'with running transaction and active span' do
+        let(:span) { Sentry.start_transaction }
+
+        before do
+          Sentry.get_current_scope.set_span(span)
+          Sentry.get_current_scope.set_transaction_name('metric', source: :view)
+        end
+
+        it 'does nothing if transaction name is low quality' do
+          expect_any_instance_of(Sentry::Metrics::LocalAggregator).not_to receive(:add)
+
+          Sentry.get_current_scope.set_transaction_name('/123', source: :url)
+          subject.add(:c, 'incr', 1)
+        end
+
+        it 'proxies bucket key and value to local aggregator' do
+          expect(span.metrics_local_aggregator).to receive(:add).with(
+            array_including(:c, 'incr', 'none'),
+            1
+          )
+          subject.add(:c, 'incr', 1)
+        end
+
+        context 'for set metrics' do
+          before { subject.add(:s, 'set', 'foo') }
+
+          it 'proxies bucket key and value 0 when existing element' do
+            expect(span.metrics_local_aggregator).to receive(:add).with(
+              array_including(:s, 'set', 'none'),
+              0
+            )
+            subject.add(:s, 'set', 'foo')
+          end
+
+          it 'proxies bucket key and value 1 when new element' do
+            expect(span.metrics_local_aggregator).to receive(:add).with(
+              array_including(:s, 'set', 'none'),
+              1
+            )
+            subject.add(:s, 'set', 'bar')
+          end
+        end
+      end
+    end
   end
 
   describe '#flush' do

sentry-ruby/spec/sentry/metrics/local_aggregator_spec.rb

@@ -0,0 +1,83 @@
+require 'spec_helper'
+
+RSpec.describe Sentry::Metrics::LocalAggregator do
+  let(:tags) { [['foo', 1], ['foo', 2], ['bar', 'baz']] }
+  let(:key) { [:c, 'incr', 'second', tags] }
+  let(:key2) { [:s, 'set', 'none', []] }
+
+  describe '#add' do
+    it 'creates new GaugeMetric and adds it to bucket if key not existing' do
+      expect(Sentry::Metrics::GaugeMetric).to receive(:new).with(10).and_call_original
+
+      subject.add(key, 10)
+
+      metric = subject.buckets[key]
+      expect(metric).to be_a(Sentry::Metrics::GaugeMetric)
+      expect(metric.last).to eq(10.0)
+      expect(metric.min).to eq(10.0)
+      expect(metric.max).to eq(10.0)
+      expect(metric.sum).to eq(10.0)
+      expect(metric.count).to eq(1)
+    end
+
+    it 'adds value to existing GaugeMetric' do
+      subject.add(key, 10)
+
+      metric = subject.buckets[key]
+      expect(metric).to be_a(Sentry::Metrics::GaugeMetric)
+      expect(metric).to receive(:add).with(20).and_call_original
+      expect(Sentry::Metrics::GaugeMetric).not_to receive(:new)
+
+      subject.add(key, 20)
+      expect(metric.last).to eq(20.0)
+      expect(metric.min).to eq(10.0)
+      expect(metric.max).to eq(20.0)
+      expect(metric.sum).to eq(30.0)
+      expect(metric.count).to eq(2)
+    end
+  end
+
+  describe '#to_hash' do
+    it 'returns nil if empty buckets' do
+      expect(subject.to_hash).to eq(nil)
+    end
+
+    context 'with filled buckets' do
+      before do
+        subject.add(key, 10)
+        subject.add(key, 20)
+        subject.add(key2, 1)
+      end
+
+      it 'has the correct payload keys in the hash' do
+        expect(subject.to_hash.keys).to eq([
+          'c:incr@second',
+          's:set@none'
+        ])
+      end
+
+      it 'has the tags deserialized correctly with array values' do
+        expect(subject.to_hash['c:incr@second'][:tags]).to eq({
+          'foo' => [1, 2],
+          'bar' => 'baz'
+        })
+      end
+
+      it 'has the correct gauge metric values' do
+        expect(subject.to_hash['c:incr@second']).to include({
+          min: 10.0,
+          max: 20.0,
+          count: 2,
+          sum: 30.0
+        })
+
+        expect(subject.to_hash['s:set@none']).to include({
+          min: 1.0,
+          max: 1.0,
+          count: 1,
+          sum: 1.0
+        })
+      end
+    end
+  end
+end