Add timing metric and beforeMetric callback (part 3) (#1954)

* added SentryOptions.beforeMetricCallback
* added beforeMetricCallback logic in metrics_aggregator.dart
* added timing metric api with span auto start
* timing api uses span duration as value for the emitted metric if possible

* Feat/metrics span summary p4 (#1958)
* added local_metrics_aggregator.dart to spans
* metrics_aggregator.dart now adds to current span's localMetricsAggregator
* added metric_summary.dart
* added metricSummary to spans and transaction JSONs

dart/lib/src/metrics/local_metrics_aggregator.dart

@@ -0,0 +1,37 @@
+import 'dart:core';
+import 'package:meta/meta.dart';
+import '../protocol/metric_summary.dart';
+import 'metric.dart';
+
+@internal
+class LocalMetricsAggregator {
+  // format: <export key, <metric key, gauge>>
+  final Map<String, Map<String, GaugeMetric>> _buckets = {};
+
+  void add(final Metric metric, final num value) {
+    final bucket =
+        _buckets.putIfAbsent(metric.getSpanAggregationKey(), () => {});
+
+    bucket.update(metric.getCompositeKey(), (m) => m..add(value),
+        ifAbsent: () => Metric.fromType(
+            type: MetricType.gauge,
+            key: metric.key,
+            value: value,
+            unit: metric.unit,
+            tags: metric.tags) as GaugeMetric);
+  }
+
+  Map<String, List<MetricSummary>> getSummaries() {
+    final Map<String, List<MetricSummary>> summaries = {};
+    for (final entry in _buckets.entries) {
+      final String exportKey = entry.key;
+
+      final metricSummaries = entry.value.values
+          .map((gauge) => MetricSummary.fromGauge(gauge))
+          .toList();
+
+      summaries[exportKey] = metricSummaries;
+    }
+    return summaries;
+  }
+}

dart/lib/src/metrics/metric.dart

@@ -112,6 +112,10 @@ abstract class Metric {
     return ('${type.statsdType}_${key}_${unit.name}_$serializedTags');
   }
 
+  /// Return a key created by [key], [type] and [unit].
+  /// This key should be used to aggregate the metric locally in a span.
+  String getSpanAggregationKey() => '${type.statsdType}:$key@${unit.name}';
+
   /// Remove forbidden characters from the metric key and tag key.
   String _normalizeKey(String input) =>
       input.replaceAll(forbiddenKeyCharsRegex, '_');
@@ -186,13 +190,9 @@ class GaugeMetric extends Metric {
 
   @visibleForTesting
   num get last => _last;
-  @visibleForTesting
   num get minimum => _minimum;
-  @visibleForTesting
   num get maximum => _maximum;
-  @visibleForTesting
   num get sum => _sum;
-  @visibleForTesting
   int get count => _count;
 }
 

dart/lib/src/metrics/metrics_aggregator.dart

@@ -59,6 +59,30 @@ class MetricsAggregator {
       return;
     }
 
+    // run before metric callback if set
+    if (_options.beforeMetricCallback != null) {
+      try {
+        final shouldEmit = _options.beforeMetricCallback!(key, tags: tags);
+        if (!shouldEmit) {
+          _options.logger(
+            SentryLevel.info,
+            'Metric was dropped by beforeMetric',
+          );
+          return;
+        }
+      } catch (exception, stackTrace) {
+        _options.logger(
+          SentryLevel.error,
+          'The BeforeMetric callback threw an exception',
+          exception: exception,
+          stackTrace: stackTrace,
+        );
+        if (_options.automatedTestMode) {
+          rethrow;
+        }
+      }
+    }
+
     final bucketKey = _getBucketKey(_options.clock());
     final bucket = _buckets.putIfAbsent(bucketKey, () => {});
     final metric = Metric.fromType(
@@ -76,6 +100,13 @@ class MetricsAggregator {
       ifAbsent: () => metric,
     );
 
+    // For sets, we only record that a value has been added to the set but not which one.
+    // See develop docs: https://develop.sentry.dev/sdk/metrics/#sets
+    _hub
+        .getSpan()
+        ?.localMetricsAggregator
+        ?.add(metric, metricType == MetricType.set ? addedWeight : value);
+
     // Schedule the metrics flushing.
     _scheduleFlush();
   }

dart/lib/src/metrics/metrics_api.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:convert';
 import '../../sentry.dart';
 import '../utils/crc32_utils.dart';
@@ -115,4 +116,65 @@ class MetricsApi {
       map.putIfAbsent(key, () => value);
     }
   }
+
+  /// Emits a Distribution metric, identified by [key], with the time it takes
+  /// to run [function].
+  /// You can set the [unit] and the optional [tags] to associate to the metric.
+  void timing(final String key,
+      {required FutureOr<void> Function() function,
+      final DurationSentryMeasurementUnit unit =
+          DurationSentryMeasurementUnit.second,
+      final Map<String, String>? tags}) async {
+    // Start a span for the metric
+    final span = _hub.getSpan()?.startChild('metric.timing', description: key);
+    // Set the user tags to the span as well
+    if (span != null && tags != null) {
+      for (final entry in tags.entries) {
+        span.setTag(entry.key, entry.value);
+      }
+    }
+    final before = _hub.options.clock();
+    try {
+      if (function is Future<void> Function()) {
+        await function();
+      } else {
+        function();
+      }
+    } finally {
+      final after = _hub.options.clock();
+      Duration duration = after.difference(before);
+      // If we have a span, we use its duration as value for the emitted metric
+      if (span != null) {
+        await span.finish();
+        duration =
+            span.endTimestamp?.difference(span.startTimestamp) ?? duration;
+      }
+      final value = _convertMicrosTo(unit, duration.inMicroseconds);
+
+      _hub.metricsAggregator?.emit(MetricType.distribution, key, value, unit,
+          _enrichWithDefaultTags(tags));
+    }
+  }
+
+  double _convertMicrosTo(
+      final DurationSentryMeasurementUnit unit, final int micros) {
+    switch (unit) {
+      case DurationSentryMeasurementUnit.nanoSecond:
+        return micros * 1000;
+      case DurationSentryMeasurementUnit.microSecond:
+        return micros.toDouble();
+      case DurationSentryMeasurementUnit.milliSecond:
+        return micros / 1000.0;
+      case DurationSentryMeasurementUnit.second:
+        return micros / 1000000.0;
+      case DurationSentryMeasurementUnit.minute:
+        return micros / 60000000.0;
+      case DurationSentryMeasurementUnit.hour:
+        return micros / 3600000000.0;
+      case DurationSentryMeasurementUnit.day:
+        return micros / 86400000000.0;
+      case DurationSentryMeasurementUnit.week:
+        return micros / 86400000000.0 / 7.0;
+    }
+  }
 }

dart/lib/src/noop_sentry_span.dart

@@ -1,3 +1,4 @@
+import 'metrics/local_metrics_aggregator.dart';
 import 'protocol.dart';
 import 'tracing.dart';
 import 'utils.dart';
@@ -95,4 +96,7 @@ class NoOpSentrySpan extends ISentrySpan {
 
   @override
   void scheduleFinish() {}
+
+  @override
+  LocalMetricsAggregator? get localMetricsAggregator => null;
 }

dart/lib/src/protocol.dart

@@ -8,6 +8,7 @@ export 'protocol/sentry_device.dart';
 export 'protocol/dsn.dart';
 export 'protocol/sentry_gpu.dart';
 export 'protocol/mechanism.dart';
+export 'protocol/metric_summary.dart';
 export 'protocol/sentry_message.dart';
 export 'protocol/sentry_operating_system.dart';
 export 'protocol/sentry_request.dart';

dart/lib/src/protocol/metric_summary.dart

@@ -0,0 +1,43 @@
+import '../metrics/metric.dart';
+
+class MetricSummary {
+  final num min;
+  final num max;
+  final num sum;
+  final int count;
+  final Map<String, String>? tags;
+
+  MetricSummary.fromGauge(GaugeMetric gauge)
+      : min = gauge.minimum,
+        max = gauge.maximum,
+        sum = gauge.sum,
+        count = gauge.count,
+        tags = gauge.tags;
+
+  const MetricSummary(
+      {required this.min,
+      required this.max,
+      required this.sum,
+      required this.count,
+      required this.tags});
+
+  /// Deserializes a [MetricSummary] from JSON [Map].
+  factory MetricSummary.fromJson(Map<String, dynamic> data) => MetricSummary(
+        min: data['min'],
+        max: data['max'],
+        count: data['count'],
+        sum: data['sum'],
+        tags: data['tags']?.cast<String, String>(),
+      );
+
+  /// Produces a [Map] that can be serialized to JSON.
+  Map<String, dynamic> toJson() {
+    return <String, dynamic>{
+      'min': min,
+      'max': max,
+      'count': count,
+      'sum': sum,
+      if (tags?.isNotEmpty ?? false) 'tags': tags,
+    };
+  }
+}

dart/lib/src/protocol/sentry_span.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import '../hub.dart';
+import '../metrics/local_metrics_aggregator.dart';
 import '../protocol.dart';
 
 import '../sentry_tracer.dart';
@@ -12,6 +13,7 @@ typedef OnFinishedCallback = Future<void> Function({DateTime? endTimestamp});
 class SentrySpan extends ISentrySpan {
   final SentrySpanContext _context;
   DateTime? _endTimestamp;
+  Map<String, List<MetricSummary>>? _metricSummaries;
   late final DateTime _startTimestamp;
   final Hub _hub;
 
@@ -22,6 +24,7 @@ class SentrySpan extends ISentrySpan {
   SpanStatus? _status;
   final Map<String, String> _tags = {};
   OnFinishedCallback? _finishedCallback;
+  late final LocalMetricsAggregator? _localMetricsAggregator;
 
   @override
   final SentryTracesSamplingDecision? samplingDecision;
@@ -37,6 +40,9 @@ class SentrySpan extends ISentrySpan {
     _startTimestamp = startTimestamp?.toUtc() ?? _hub.options.clock();
     _finishedCallback = finishedCallback;
     _origin = _context.origin;
+    _localMetricsAggregator = _hub.options.enableSpanLocalMetricAggregation
+        ? LocalMetricsAggregator()
+        : null;
   }
 
   @override
@@ -65,6 +71,7 @@ class SentrySpan extends ISentrySpan {
     if (_throwable != null) {
       _hub.setSpanContext(_throwable, this, _tracer.name);
     }
+    _metricSummaries = _localMetricsAggregator?.getSummaries();
     await _finishedCallback?.call(endTimestamp: _endTimestamp);
     return super.finish(status: status, endTimestamp: _endTimestamp);
   }
@@ -154,6 +161,9 @@ class SentrySpan extends ISentrySpan {
   @override
   set origin(String? origin) => _origin = origin;
 
+  @override
+  LocalMetricsAggregator? get localMetricsAggregator => _localMetricsAggregator;
+
   Map<String, dynamic> toJson() {
     final json = _context.toJson();
     json['start_timestamp'] =
@@ -174,6 +184,16 @@ class SentrySpan extends ISentrySpan {
     if (_origin != null) {
       json['origin'] = _origin;
     }
+
+    final metricSummariesMap = _metricSummaries?.entries ?? Iterable.empty();
+    if (metricSummariesMap.isNotEmpty) {
+      final map = <String, dynamic>{};
+      for (final entry in metricSummariesMap) {
+        final summary = entry.value.map((e) => e.toJson());
+        map[entry.key] = summary.toList(growable: false);
+      }
+      json['_metrics_summary'] = map;
+    }
     return json;
   }
 

dart/lib/src/protocol/sentry_transaction.dart

@@ -13,6 +13,7 @@ class SentryTransaction extends SentryEvent {
   @internal
   final SentryTracer tracer;
   late final Map<String, SentryMeasurement> measurements;
+  late final Map<String, List<MetricSummary>>? metricSummaries;
   late final SentryTransactionInfo? transactionInfo;
 
   SentryTransaction(
@@ -37,6 +38,7 @@ class SentryTransaction extends SentryEvent {
     super.request,
     String? type,
     Map<String, SentryMeasurement>? measurements,
+    Map<String, List<MetricSummary>>? metricSummaries,
     SentryTransactionInfo? transactionInfo,
   }) : super(
           timestamp: timestamp ?? tracer.endTimestamp,
@@ -52,6 +54,8 @@ class SentryTransaction extends SentryEvent {
     final spanContext = tracer.context;
     spans = tracer.children;
     this.measurements = measurements ?? {};
+    this.metricSummaries =
+        metricSummaries ?? tracer.localMetricsAggregator?.getSummaries();
 
     contexts.trace = spanContext.toTraceContext(
       sampled: tracer.samplingDecision?.sampled,
@@ -85,6 +89,16 @@ class SentryTransaction extends SentryEvent {
       json['transaction_info'] = transactionInfo.toJson();
     }
 
+    final metricSummariesMap = metricSummaries?.entries ?? Iterable.empty();
+    if (metricSummariesMap.isNotEmpty) {
+      final map = <String, dynamic>{};
+      for (final entry in metricSummariesMap) {
+        final summary = entry.value.map((e) => e.toJson());
+        map[entry.key] = summary.toList(growable: false);
+      }
+      json['_metrics_summary'] = map;
+    }
+
     return json;
   }
 
@@ -123,6 +137,7 @@ class SentryTransaction extends SentryEvent {
     List<SentryThread>? threads,
     String? type,
     Map<String, SentryMeasurement>? measurements,
+    Map<String, List<MetricSummary>>? metricSummaries,
     SentryTransactionInfo? transactionInfo,
   }) =>
       SentryTransaction(
@@ -148,6 +163,9 @@ class SentryTransaction extends SentryEvent {
         type: type ?? this.type,
         measurements: (measurements != null ? Map.from(measurements) : null) ??
             this.measurements,
+        metricSummaries:
+            (metricSummaries != null ? Map.from(metricSummaries) : null) ??
+                this.metricSummaries,
         transactionInfo: transactionInfo ?? this.transactionInfo,
       );
 }