feat: add server runtime metrics aggregator

Author: anonrig

packages/bun/src/index.ts

@@ -67,6 +67,7 @@ export {
   startInactiveSpan,
   startSpanManual,
   continueTrace,
+  metrics,
 } from '@sentry/core';
 export type { SpanStatusType } from '@sentry/core';
 export { autoDiscoverNodePerformanceMonitoringIntegrations } from '@sentry/node';

packages/core/src/metrics/aggregator.ts

@@ -0,0 +1,118 @@
+import type {
+  MeasurementUnit,
+  Primitive,
+  MetricsAggregator as MetricsAggregatorBase,
+  ClientOptions, Client, MetricBucketItem
+} from '@sentry/types';
+import {DEFAULT_FLUSH_INTERVAL} from './constants';
+import type {MetricBucket, MetricType} from './types';
+import {timestampInSeconds} from '@sentry/utils';
+
+/**
+ * A metrics aggregator that aggregates metrics in memory and flushes them periodically.
+ */
+export class MetricsAggregator implements MetricsAggregatorBase {
+  private _buckets: MetricBucket;
+  private _bucketsTotalWeight;
+  private readonly _interval: ReturnType<typeof setInterval>;
+  private readonly _flushShift: number;
+  private _forceFlush: boolean;
+
+  public constructor(private readonly _client: Client<ClientOptions>) {
+    this._buckets = new Map();
+    this._bucketsTotalWeight = 0;
+    this._interval = setInterval(() => this._flush(), DEFAULT_FLUSH_INTERVAL);
+
+    // SDKs are required to shift the flush interval by random() * rollup_in_seconds.
+    // That shift is determined once per startup to create jittering.
+    this._flushShift = Math.random() * DEFAULT_FLUSH_INTERVAL;
+
+    this._forceFlush = false;
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public add(
+    metricType: MetricType,
+    unsanitizedName: string,
+    value: number | string,
+    unit: MeasurementUnit = 'none',
+    unsanitizedTags: Record<string, Primitive> = {},
+    maybeFloatTimestamp = timestampInSeconds(),
+  ): void {
+    // Do nothing
+  }
+
+  /**
+   * Flushes the current metrics to the transport via the transport.
+   */
+  public flush(): void {
+    this._forceFlush = true;
+    this._flush();
+  }
+
+  /**
+   * Shuts down metrics aggregator and clears all metrics.
+   */
+  public close(): void {
+    this._forceFlush = true;
+    clearInterval(this._interval);
+    this._flush();
+  }
+
+  /**
+   * Returns a string representation of the aggregator.
+   */
+  public toString(): string {
+    return '';
+  }
+
+  /**
+   * Flushes the buckets according to the internal state of the aggregator.
+   * If it is a force flush, which happens on shutdown, it will flush all buckets.
+   * Otherwise, it will only flush buckets that are older than the flush interval,
+   * and according to the flush shift.
+   *
+   * This function mutates `_forceFlush` and `_bucketsTotalWeight` properties.
+   */
+  private _flush(): void {
+    // This path eliminates the need for checking for timestamps since we're forcing a flush.
+    // Remember to reset the flag, or it will always flush all metrics.
+    if (this._forceFlush) {
+      this._forceFlush = false;
+      this._bucketsTotalWeight = 0;
+      this._captureMetrics(this._buckets);
+      this._buckets.clear();
+      return;
+    }
+    const cutoffSeconds = timestampInSeconds() - DEFAULT_FLUSH_INTERVAL - this._flushShift;
+    // TODO(@anonrig): Optimization opportunity.
+    // Convert this map to an array and store key in the bucketItem.
+    const flushedBuckets: MetricBucket = new Map();
+    for (const [key, bucket] of this._buckets) {
+      if (bucket.timestamp < cutoffSeconds) {
+        flushedBuckets.set(key, bucket);
+        this._bucketsTotalWeight -= bucket.metric.weight;
+      }
+    }
+
+    for (const [key] of flushedBuckets) {
+      this._buckets.delete(key);
+    }
+
+    this._captureMetrics(flushedBuckets);
+  }
+
+  /**
+   * Only captures a subset of the buckets passed to this function.
+   * @param flushedBuckets
+   */
+  private _captureMetrics(flushedBuckets: MetricBucket): void {
+    if (flushedBuckets.size > 0 && this._client.captureAggregateMetrics) {
+      // TODO(@anonrig): This copy operation can be avoided if we store the key in the bucketItem.
+      const buckets = Array.from(flushedBuckets).map(([, bucketItem]) => bucketItem);
+      this._client.captureAggregateMetrics(buckets);
+    }
+  }
+}

packages/core/src/metrics/simpleaggregator.ts renamed to packages/core/src/metrics/browser-aggregator.ts

@@ -1,27 +1,33 @@
-import type { Client, ClientOptions, MeasurementUnit, MetricsAggregator, Primitive } from '@sentry/types';
+import type {
+  Client,
+  ClientOptions,
+  MeasurementUnit,
+  MetricBucketItem,
+  MetricsAggregator,
+  Primitive
+} from '@sentry/types';
 import { timestampInSeconds } from '@sentry/utils';
 import {
-  DEFAULT_FLUSH_INTERVAL,
+  DEFAULT_BROWSER_FLUSH_INTERVAL,
   NAME_AND_TAG_KEY_NORMALIZATION_REGEX,
-  TAG_VALUE_NORMALIZATION_REGEX,
 } from './constants';
 import { METRIC_MAP } from './instance';
-import type { MetricType, SimpleMetricBucket } from './types';
-import { getBucketKey } from './utils';
+import type { MetricType, MetricBucket } from './types';
+import { getBucketKey, sanitizeTags } from './utils';
 
 /**
  * A simple metrics aggregator that aggregates metrics in memory and flushes them periodically.
  * Default flush interval is 5 seconds.
  *
  * @experimental This API is experimental and might change in the future.
  */
-export class SimpleMetricsAggregator implements MetricsAggregator {
-  private _buckets: SimpleMetricBucket;
+export class BrowserMetricsAggregator implements MetricsAggregator {
+  private _buckets: MetricBucket;
   private readonly _interval: ReturnType<typeof setInterval>;
 
   public constructor(private readonly _client: Client<ClientOptions>) {
     this._buckets = new Map();
-    this._interval = setInterval(() => this.flush(), DEFAULT_FLUSH_INTERVAL);
+    this._interval = setInterval(() => this.flush(), DEFAULT_BROWSER_FLUSH_INTERVAL);
   }
 
   /**
@@ -31,27 +37,33 @@ export class SimpleMetricsAggregator implements MetricsAggregator {
     metricType: MetricType,
     unsanitizedName: string,
     value: number | string,
-    unit: MeasurementUnit = 'none',
-    unsanitizedTags: Record<string, Primitive> = {},
-    maybeFloatTimestamp = timestampInSeconds(),
+    unit: MeasurementUnit | undefined = 'none',
+    unsanitizedTags: Record<string, Primitive> | undefined = {},
+    maybeFloatTimestamp: number | undefined = timestampInSeconds(),
   ): void {
     const timestamp = Math.floor(maybeFloatTimestamp);
     const name = unsanitizedName.replace(NAME_AND_TAG_KEY_NORMALIZATION_REGEX, '_');
     const tags = sanitizeTags(unsanitizedTags);
 
     const bucketKey = getBucketKey(metricType, name, unit, tags);
-    const bucketItem = this._buckets.get(bucketKey);
+    const bucketItem: MetricBucketItem | undefined = this._buckets.get(bucketKey);
     if (bucketItem) {
-      const [bucketMetric, bucketTimestamp] = bucketItem;
-      bucketMetric.add(value);
+      bucketItem.metric.add(value);
       // TODO(abhi): Do we need this check?
-      if (bucketTimestamp < timestamp) {
-        bucketItem[1] = timestamp;
+      if (bucketItem.timestamp < timestamp) {
+        bucketItem.timestamp = timestamp;
       }
     } else {
-      // @ts-expect-error we don't need to narrow down the type of value here, saves bundle size.
-      const newMetric = new METRIC_MAP[metricType](value);
-      this._buckets.set(bucketKey, [newMetric, timestamp, metricType, name, unit, tags]);
+      this._buckets.set(bucketKey, {
+        // @ts-expect-error we don't need to narrow down the type of value here, saves bundle size.
+        metric: new METRIC_MAP[metricType](value),
+        timestamp,
+        metricType,
+        name,
+        unit,
+        tags,
+      });
+
     }
   }
 
@@ -78,14 +90,3 @@ export class SimpleMetricsAggregator implements MetricsAggregator {
     this.flush();
   }
 }
-
-function sanitizeTags(unsanitizedTags: Record<string, Primitive>): Record<string, string> {
-  const tags: Record<string, string> = {};
-  for (const key in unsanitizedTags) {
-    if (Object.prototype.hasOwnProperty.call(unsanitizedTags, key)) {
-      const sanitizedKey = key.replace(NAME_AND_TAG_KEY_NORMALIZATION_REGEX, '_');
-      tags[sanitizedKey] = String(unsanitizedTags[key]).replace(TAG_VALUE_NORMALIZATION_REGEX, '_');
-    }
-  }
-  return tags;
-}

packages/core/src/metrics/constants.ts

@@ -27,4 +27,11 @@ export const TAG_VALUE_NORMALIZATION_REGEX = /[^\w\d_:/@.{}[\]$-]+/g;
  * This does not match spec in https://develop.sentry.dev/sdk/metrics
  * but was chosen to optimize for the most common case in browser environments.
  */
-export const DEFAULT_FLUSH_INTERVAL = 5000;
+export const DEFAULT_BROWSER_FLUSH_INTERVAL = 5000;
+
+/**
+ * SDKs are required to bucket into 10 second intervals (rollup in seconds)
+ * which is the current lower bound of metric accuracy.
+ */
+export const DEFAULT_FLUSH_INTERVAL = 10000;
+

packages/core/src/metrics/envelope.ts

@@ -30,7 +30,7 @@ export function createMetricEnvelope(
   return createEnvelope<StatsdEnvelope>(headers, [item]);
 }
 
-function createMetricEnvelopeItem(metricBucketItems: Array<MetricBucketItem>): StatsdItem {
+function createMetricEnvelopeItem(metricBucketItems: MetricBucketItem[]): StatsdItem {
   const payload = serializeMetricBuckets(metricBucketItems);
   const metricHeaders: StatsdItem[0] = {
     type: 'statsd',

packages/core/src/metrics/exports.ts

@@ -17,7 +17,7 @@ function addToMetricsAggregator(
   metricType: MetricType,
   name: string,
   value: number | string,
-  data: MetricData = {},
+  data: MetricData | undefined = {},
 ): void {
   const hub = getCurrentHub();
   const client = hub.getClient() as BaseClient<ClientOptions>;
@@ -50,7 +50,7 @@ function addToMetricsAggregator(
 /**
  * Adds a value to a counter metric
  *
- * @experimental This API is experimental and might having breaking changes in the future.
+ * @experimental This API is experimental and might have breaking changes in the future.
  */
 export function increment(name: string, value: number = 1, data?: MetricData): void {
   addToMetricsAggregator(COUNTER_METRIC_TYPE, name, value, data);
@@ -59,7 +59,7 @@ export function increment(name: string, value: number = 1, data?: MetricData): v
 /**
  * Adds a value to a distribution metric
  *
- * @experimental This API is experimental and might having breaking changes in the future.
+ * @experimental This API is experimental and might have breaking changes in the future.
  */
 export function distribution(name: string, value: number, data?: MetricData): void {
   addToMetricsAggregator(DISTRIBUTION_METRIC_TYPE, name, value, data);
@@ -68,7 +68,7 @@ export function distribution(name: string, value: number, data?: MetricData): vo
 /**
  * Adds a value to a set metric. Value must be a string or integer.
  *
- * @experimental This API is experimental and might having breaking changes in the future.
+ * @experimental This API is experimental and might have breaking changes in the future.
  */
 export function set(name: string, value: number | string, data?: MetricData): void {
   addToMetricsAggregator(SET_METRIC_TYPE, name, value, data);
@@ -77,7 +77,7 @@ export function set(name: string, value: number | string, data?: MetricData): vo
 /**
  * Adds a value to a gauge metric
  *
- * @experimental This API is experimental and might having breaking changes in the future.
+ * @experimental This API is experimental and might have breaking changes in the future.
  */
 export function gauge(name: string, value: number, data?: MetricData): void {
   addToMetricsAggregator(GAUGE_METRIC_TYPE, name, value, data);

packages/core/src/metrics/instance.ts

@@ -8,6 +8,11 @@ import { simpleHash } from './utils';
 export class CounterMetric implements MetricInstance {
   public constructor(private _value: number) {}
 
+  /** @inheritDoc */
+  public get weight(): number {
+    return 1;
+  }
+
   /** @inheritdoc */
   public add(value: number): void {
     this._value += value;
@@ -37,6 +42,11 @@ export class GaugeMetric implements MetricInstance {
     this._count = 1;
   }
 
+  /** @inheritDoc */
+  public get weight(): number {
+    return 5;
+  }
+
   /** @inheritdoc */
   public add(value: number): void {
     this._last = value;
@@ -66,6 +76,11 @@ export class DistributionMetric implements MetricInstance {
     this._value = [first];
   }
 
+  /** @inheritDoc */
+  public get weight(): number {
+    return this._value.length;
+  }
+
   /** @inheritdoc */
   public add(value: number): void {
     this._value.push(value);
@@ -87,21 +102,24 @@ export class SetMetric implements MetricInstance {
     this._value = new Set([first]);
   }
 
+  /** @inheritDoc */
+  public get weight(): number {
+    return this._value.size;
+  }
+
   /** @inheritdoc */
   public add(value: number | string): void {
     this._value.add(value);
   }
 
   /** @inheritdoc */
   public toString(): string {
-    return `${Array.from(this._value)
+    return Array.from(this._value)
       .map(val => (typeof val === 'string' ? simpleHash(val) : val))
-      .join(':')}`;
+      .join(':');
   }
 }
 
-export type Metric = CounterMetric | GaugeMetric | DistributionMetric | SetMetric;
-
 export const METRIC_MAP = {
   [COUNTER_METRIC_TYPE]: CounterMetric,
   [GAUGE_METRIC_TYPE]: GaugeMetric,

packages/core/src/metrics/integration.ts

@@ -1,6 +1,6 @@
 import type { ClientOptions, Integration } from '@sentry/types';
 import type { BaseClient } from '../baseclient';
-import { SimpleMetricsAggregator } from './simpleaggregator';
+import { BrowserMetricsAggregator } from './browser-aggregator';
 
 /**
  * Enables Sentry metrics monitoring.
@@ -33,6 +33,6 @@ export class MetricsAggregator implements Integration {
    * @inheritDoc
    */
   public setup(client: BaseClient<ClientOptions>): void {
-    client.metricsAggregator = new SimpleMetricsAggregator(client);
+    client.metricsAggregator = new BrowserMetricsAggregator(client);
   }
 }

packages/core/src/metrics/types.ts

@@ -7,4 +7,4 @@ export type MetricType =
   | typeof SET_METRIC_TYPE
   | typeof DISTRIBUTION_METRIC_TYPE;
 
-export type SimpleMetricBucket = Map<string, MetricBucketItem>;
+export type MetricBucket = Map<string, MetricBucketItem>;