-
-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Add server runtime metrics aggregator (#9894)
## Implements - [x] 10 Second Bucketing: SDKs are required to bucket into 10 second intervals (rollup in seconds) which is the current lower bound of metric accuracy. - [x] Flush Shift: SDKs are required to shift the flush interval by random() * rollup_in_seconds. That shift is determined once per startup to create jittering. - [ ] Force flush: an SDK is required to perform force flushing ahead of scheduled time if the memory pressure is too high. There is no rule for this other than that SDKs should be tracking abstract aggregation complexity (eg: a counter only carries a single float, whereas a distribution is a float per emission). ## Caveats - Force flush requires Node.js 14+ support (FinalizationRegistry). I recommend leaving it after v8 release to make the implementation a lot easier. --------- Co-authored-by: Abhijeet Prasad <[email protected]>
- Loading branch information
1 parent
5d16aae
commit 7f8eca7
Showing
20 changed files
with
540 additions
and
137 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,163 @@ | ||
| import type { | ||
| Client, | ||
| ClientOptions, | ||
| MeasurementUnit, | ||
| MetricsAggregator as MetricsAggregatorBase, | ||
| Primitive, | ||
| } from '@sentry/types'; | ||
| import { timestampInSeconds } from '@sentry/utils'; | ||
| import { DEFAULT_FLUSH_INTERVAL, MAX_WEIGHT, NAME_AND_TAG_KEY_NORMALIZATION_REGEX } from './constants'; | ||
| import { METRIC_MAP } from './instance'; | ||
| import type { MetricBucket, MetricType } from './types'; | ||
| import { getBucketKey, sanitizeTags } from './utils'; | ||
|
|
||
| /** | ||
| * A metrics aggregator that aggregates metrics in memory and flushes them periodically. | ||
| */ | ||
| export class MetricsAggregator implements MetricsAggregatorBase { | ||
| // TODO(@anonrig): Use FinalizationRegistry to have a proper way of flushing the buckets | ||
| // when the aggregator is garbage collected. | ||
| // Ref: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/FinalizationRegistry | ||
| private _buckets: MetricBucket; | ||
|
|
||
| // Different metrics have different weights. We use this to limit the number of metrics | ||
| // that we store in memory. | ||
| private _bucketsTotalWeight; | ||
|
|
||
| private readonly _interval: ReturnType<typeof setInterval>; | ||
|
|
||
| // SDKs are required to shift the flush interval by random() * rollup_in_seconds. | ||
| // That shift is determined once per startup to create jittering. | ||
| private readonly _flushShift: number; | ||
|
|
||
| // An SDK is required to perform force flushing ahead of scheduled time if the memory | ||
| // pressure is too high. There is no rule for this other than that SDKs should be tracking | ||
| // abstract aggregation complexity (eg: a counter only carries a single float, whereas a | ||
| // distribution is a float per emission). | ||
| // | ||
| // Force flush is used on either shutdown, flush() or when we exceed the max weight. | ||
| 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); | ||
| this._flushShift = Math.floor((Math.random() * DEFAULT_FLUSH_INTERVAL) / 1000); | ||
| this._forceFlush = false; | ||
| } | ||
|
|
||
| /** | ||
| * @inheritDoc | ||
| */ | ||
| public add( | ||
| metricType: MetricType, | ||
| unsanitizedName: string, | ||
| value: number | string, | ||
| unit: MeasurementUnit = 'none', | ||
| unsanitizedTags: Record<string, Primitive> = {}, | ||
| maybeFloatTimestamp = 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); | ||
| let bucketItem = this._buckets.get(bucketKey); | ||
| if (bucketItem) { | ||
| bucketItem.metric.add(value); | ||
| // TODO(abhi): Do we need this check? | ||
| if (bucketItem.timestamp < timestamp) { | ||
| bucketItem.timestamp = timestamp; | ||
| } | ||
| } else { | ||
| bucketItem = { | ||
| // @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, | ||
| }; | ||
| this._buckets.set(bucketKey, bucketItem); | ||
| } | ||
|
|
||
| // We need to keep track of the total weight of the buckets so that we can | ||
| // flush them when we exceed the max weight. | ||
| this._bucketsTotalWeight += bucketItem.metric.weight; | ||
|
|
||
| if (this._bucketsTotalWeight >= MAX_WEIGHT) { | ||
| this.flush(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * 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(); | ||
| } | ||
|
|
||
| /** | ||
| * 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 { | ||
| // TODO(@anonrig): Add Atomics for locking to avoid having force flush and regular flush | ||
| // running at the same time. | ||
| // Ref: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics | ||
|
|
||
| // 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 = Math.floor(timestampInSeconds()) - DEFAULT_FLUSH_INTERVAL / 1000 - 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): Optimization opportunity. | ||
| // 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); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| import type { | ||
| Client, | ||
| ClientOptions, | ||
| MeasurementUnit, | ||
| MetricBucketItem, | ||
| MetricsAggregator, | ||
| Primitive, | ||
| } from '@sentry/types'; | ||
| import { timestampInSeconds } from '@sentry/utils'; | ||
| import { DEFAULT_BROWSER_FLUSH_INTERVAL, NAME_AND_TAG_KEY_NORMALIZATION_REGEX } from './constants'; | ||
| import { METRIC_MAP } from './instance'; | ||
| import type { MetricBucket, MetricType } 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 BrowserMetricsAggregator implements MetricsAggregator { | ||
| // TODO(@anonrig): Use FinalizationRegistry to have a proper way of flushing the buckets | ||
| // when the aggregator is garbage collected. | ||
| // Ref: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/FinalizationRegistry | ||
| 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_BROWSER_FLUSH_INTERVAL); | ||
| } | ||
|
|
||
| /** | ||
| * @inheritDoc | ||
| */ | ||
| public add( | ||
| metricType: MetricType, | ||
| unsanitizedName: string, | ||
| value: number | string, | ||
| 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: MetricBucketItem | undefined = this._buckets.get(bucketKey); | ||
| if (bucketItem) { | ||
| bucketItem.metric.add(value); | ||
| // TODO(abhi): Do we need this check? | ||
| if (bucketItem.timestamp < timestamp) { | ||
| bucketItem.timestamp = timestamp; | ||
| } | ||
| } else { | ||
| 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, | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * @inheritDoc | ||
| */ | ||
| public flush(): void { | ||
| // short circuit if buckets are empty. | ||
| if (this._buckets.size === 0) { | ||
| return; | ||
| } | ||
| if (this._client.captureAggregateMetrics) { | ||
| // TODO(@anonrig): Use Object.values() when we support ES6+ | ||
| const metricBuckets = Array.from(this._buckets).map(([, bucketItem]) => bucketItem); | ||
| this._client.captureAggregateMetrics(metricBuckets); | ||
| } | ||
| this._buckets.clear(); | ||
| } | ||
|
|
||
| /** | ||
| * @inheritDoc | ||
| */ | ||
| public close(): void { | ||
| clearInterval(this._interval); | ||
| this.flush(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.