Metric summaries on span (#2255)

* 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 `metric.timing` span in the `Sentry::Metrics.timing` API
* add `before_emit` callback


part of #2246

CHANGELOG.md

@@ -8,6 +8,8 @@
 - 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)
+  - Add `config.metrics.before_emit` callback [#2258](https://github.com/getsentry/sentry-ruby/pull/2258)
 
     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,9 +41,22 @@
     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) }
+
+    # add a before_emit callback to filter keys or update tags
+    Sentry.init do |config|
+      # ...
+      config.metrics.enabled = true
+      config.metrics.before_emit = lambda do |key, tags|
+        return nil if key == 'foo'
+        tags[:bar] = 42
+        tags.delete(:baz)
+        true
+      end
+    end
     ```
 
 ### Bug Fixes

sentry-ruby/lib/sentry/configuration.rb

@@ -576,12 +576,6 @@ def error_messages
 
     private
 
-    def check_callable!(name, value)
-      unless value == nil || value.respond_to?(:call)
-        raise ArgumentError, "#{name} must be callable (or nil to disable)"
-      end
-    end
-
     def init_dsn(dsn_string)
       return if dsn_string.nil? || dsn_string.empty?
 

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

@@ -24,6 +24,7 @@ class Aggregator
       def initialize(configuration, client)
         @client = client
         @logger = configuration.logger
+        @before_emit = configuration.metrics.before_emit
 
         @default_tags = {}
         @default_tags['release'] = configuration.release if configuration.release
@@ -55,19 +56,30 @@ def add(type,
         # this is integer division and thus takes the floor of the division
         # and buckets into 10 second intervals
         bucket_timestamp = (timestamp / ROLLUP_IN_SECONDS) * ROLLUP_IN_SECONDS
+        updated_tags = get_updated_tags(tags)
 
-        serialized_tags = serialize_tags(get_updated_tags(tags))
+        return if @before_emit && !@before_emit.call(key, updated_tags)
+
+        serialized_tags = serialize_tags(updated_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)
@@ -179,6 +191,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/configuration.rb

@@ -3,15 +3,38 @@
 module Sentry
   module Metrics
     class Configuration
+      include ArgumentCheckingHelper
+
       # Enable metrics usage
       # Starts a new {Sentry::Metrics::Aggregator} instance to aggregate metrics
       # and a thread to aggregate flush every 5 seconds.
       # @return [Boolean]
       attr_accessor :enabled
 
+      # Optional Proc, called before emitting a metric to the aggregator.
+      # Use it to filter keys (return false/nil) or update tags.
+      # Make sure to return true at the end.
+      #
+      # @example
+      #   config.metrics.before_emit = lambda do |key, tags|
+      #     return nil if key == 'foo'
+      #     tags[:bar] = 42
+      #     tags.delete(:baz)
+      #     true
+      #   end
+      #
+      # @return [Proc, nil]
+      attr_reader :before_emit
+
       def initialize
         @enabled = false
       end
+
+      def before_emit=(value)
+        check_callable!("metrics.before_emit", value)
+
+        @before_emit = 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/lib/sentry/utils/argument_checking_helper.rb

@@ -15,5 +15,11 @@ def check_argument_includes!(argument, values)
         raise ArgumentError, "expect the argument to be one of #{values.map(&:inspect).join(' or ')}, got #{argument.inspect}"
       end
     end
+
+    def check_callable!(name, value)
+      unless value == nil || value.respond_to?(:call)
+        raise ArgumentError, "#{name} must be callable (or nil to disable)"
+      end
+    end
   end
 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