|
| 1 | +import type { |
| 2 | + Client, |
| 3 | + ClientOptions, |
| 4 | + MeasurementUnit, |
| 5 | + MetricsAggregator as MetricsAggregatorBase, |
| 6 | + Primitive, |
| 7 | +} from '@sentry/types'; |
| 8 | +import { timestampInSeconds } from '@sentry/utils'; |
| 9 | +import { DEFAULT_FLUSH_INTERVAL, MAX_WEIGHT, NAME_AND_TAG_KEY_NORMALIZATION_REGEX } from './constants'; |
| 10 | +import { METRIC_MAP } from './instance'; |
| 11 | +import type { MetricBucket, MetricType } from './types'; |
| 12 | +import { getBucketKey, sanitizeTags } from './utils'; |
| 13 | + |
| 14 | +/** |
| 15 | + * A metrics aggregator that aggregates metrics in memory and flushes them periodically. |
| 16 | + */ |
| 17 | +export class MetricsAggregator implements MetricsAggregatorBase { |
| 18 | + // TODO(@anonrig): Use FinalizationRegistry to have a proper way of flushing the buckets |
| 19 | + // when the aggregator is garbage collected. |
| 20 | + // Ref: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/FinalizationRegistry |
| 21 | + private _buckets: MetricBucket; |
| 22 | + |
| 23 | + // Different metrics have different weights. We use this to limit the number of metrics |
| 24 | + // that we store in memory. |
| 25 | + private _bucketsTotalWeight; |
| 26 | + |
| 27 | + private readonly _interval: ReturnType<typeof setInterval>; |
| 28 | + |
| 29 | + // SDKs are required to shift the flush interval by random() * rollup_in_seconds. |
| 30 | + // That shift is determined once per startup to create jittering. |
| 31 | + private readonly _flushShift: number; |
| 32 | + |
| 33 | + // An SDK is required to perform force flushing ahead of scheduled time if the memory |
| 34 | + // pressure is too high. There is no rule for this other than that SDKs should be tracking |
| 35 | + // abstract aggregation complexity (eg: a counter only carries a single float, whereas a |
| 36 | + // distribution is a float per emission). |
| 37 | + // |
| 38 | + // Force flush is used on either shutdown, flush() or when we exceed the max weight. |
| 39 | + private _forceFlush: boolean; |
| 40 | + |
| 41 | + public constructor(private readonly _client: Client<ClientOptions>) { |
| 42 | + this._buckets = new Map(); |
| 43 | + this._bucketsTotalWeight = 0; |
| 44 | + this._interval = setInterval(() => this._flush(), DEFAULT_FLUSH_INTERVAL); |
| 45 | + this._flushShift = Math.floor((Math.random() * DEFAULT_FLUSH_INTERVAL) / 1000); |
| 46 | + this._forceFlush = false; |
| 47 | + } |
| 48 | + |
| 49 | + /** |
| 50 | + * @inheritDoc |
| 51 | + */ |
| 52 | + public add( |
| 53 | + metricType: MetricType, |
| 54 | + unsanitizedName: string, |
| 55 | + value: number | string, |
| 56 | + unit: MeasurementUnit = 'none', |
| 57 | + unsanitizedTags: Record<string, Primitive> = {}, |
| 58 | + maybeFloatTimestamp = timestampInSeconds(), |
| 59 | + ): void { |
| 60 | + const timestamp = Math.floor(maybeFloatTimestamp); |
| 61 | + const name = unsanitizedName.replace(NAME_AND_TAG_KEY_NORMALIZATION_REGEX, '_'); |
| 62 | + const tags = sanitizeTags(unsanitizedTags); |
| 63 | + |
| 64 | + const bucketKey = getBucketKey(metricType, name, unit, tags); |
| 65 | + let bucketItem = this._buckets.get(bucketKey); |
| 66 | + if (bucketItem) { |
| 67 | + bucketItem.metric.add(value); |
| 68 | + // TODO(abhi): Do we need this check? |
| 69 | + if (bucketItem.timestamp < timestamp) { |
| 70 | + bucketItem.timestamp = timestamp; |
| 71 | + } |
| 72 | + } else { |
| 73 | + bucketItem = { |
| 74 | + // @ts-expect-error we don't need to narrow down the type of value here, saves bundle size. |
| 75 | + metric: new METRIC_MAP[metricType](value), |
| 76 | + timestamp, |
| 77 | + metricType, |
| 78 | + name, |
| 79 | + unit, |
| 80 | + tags, |
| 81 | + }; |
| 82 | + this._buckets.set(bucketKey, bucketItem); |
| 83 | + } |
| 84 | + |
| 85 | + // We need to keep track of the total weight of the buckets so that we can |
| 86 | + // flush them when we exceed the max weight. |
| 87 | + this._bucketsTotalWeight += bucketItem.metric.weight; |
| 88 | + |
| 89 | + if (this._bucketsTotalWeight >= MAX_WEIGHT) { |
| 90 | + this.flush(); |
| 91 | + } |
| 92 | + } |
| 93 | + |
| 94 | + /** |
| 95 | + * Flushes the current metrics to the transport via the transport. |
| 96 | + */ |
| 97 | + public flush(): void { |
| 98 | + this._forceFlush = true; |
| 99 | + this._flush(); |
| 100 | + } |
| 101 | + |
| 102 | + /** |
| 103 | + * Shuts down metrics aggregator and clears all metrics. |
| 104 | + */ |
| 105 | + public close(): void { |
| 106 | + this._forceFlush = true; |
| 107 | + clearInterval(this._interval); |
| 108 | + this._flush(); |
| 109 | + } |
| 110 | + |
| 111 | + /** |
| 112 | + * Flushes the buckets according to the internal state of the aggregator. |
| 113 | + * If it is a force flush, which happens on shutdown, it will flush all buckets. |
| 114 | + * Otherwise, it will only flush buckets that are older than the flush interval, |
| 115 | + * and according to the flush shift. |
| 116 | + * |
| 117 | + * This function mutates `_forceFlush` and `_bucketsTotalWeight` properties. |
| 118 | + */ |
| 119 | + private _flush(): void { |
| 120 | + // TODO(@anonrig): Add Atomics for locking to avoid having force flush and regular flush |
| 121 | + // running at the same time. |
| 122 | + // Ref: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Atomics |
| 123 | + |
| 124 | + // This path eliminates the need for checking for timestamps since we're forcing a flush. |
| 125 | + // Remember to reset the flag, or it will always flush all metrics. |
| 126 | + if (this._forceFlush) { |
| 127 | + this._forceFlush = false; |
| 128 | + this._bucketsTotalWeight = 0; |
| 129 | + this._captureMetrics(this._buckets); |
| 130 | + this._buckets.clear(); |
| 131 | + return; |
| 132 | + } |
| 133 | + const cutoffSeconds = Math.floor(timestampInSeconds()) - DEFAULT_FLUSH_INTERVAL / 1000 - this._flushShift; |
| 134 | + // TODO(@anonrig): Optimization opportunity. |
| 135 | + // Convert this map to an array and store key in the bucketItem. |
| 136 | + const flushedBuckets: MetricBucket = new Map(); |
| 137 | + for (const [key, bucket] of this._buckets) { |
| 138 | + if (bucket.timestamp <= cutoffSeconds) { |
| 139 | + flushedBuckets.set(key, bucket); |
| 140 | + this._bucketsTotalWeight -= bucket.metric.weight; |
| 141 | + } |
| 142 | + } |
| 143 | + |
| 144 | + for (const [key] of flushedBuckets) { |
| 145 | + this._buckets.delete(key); |
| 146 | + } |
| 147 | + |
| 148 | + this._captureMetrics(flushedBuckets); |
| 149 | + } |
| 150 | + |
| 151 | + /** |
| 152 | + * Only captures a subset of the buckets passed to this function. |
| 153 | + * @param flushedBuckets |
| 154 | + */ |
| 155 | + private _captureMetrics(flushedBuckets: MetricBucket): void { |
| 156 | + if (flushedBuckets.size > 0 && this._client.captureAggregateMetrics) { |
| 157 | + // TODO(@anonrig): Optimization opportunity. |
| 158 | + // This copy operation can be avoided if we store the key in the bucketItem. |
| 159 | + const buckets = Array.from(flushedBuckets).map(([, bucketItem]) => bucketItem); |
| 160 | + this._client.captureAggregateMetrics(buckets); |
| 161 | + } |
| 162 | + } |
| 163 | +} |
0 commit comments