diff --git a/package.json b/package.json
index 4fa76ae58198..e1d8309637c6 100644
--- a/package.json
+++ b/package.json
@@ -18,12 +18,12 @@
     "clean:deps": "lerna clean --yes && rm -rf node_modules && yarn",
     "clean:all": "run-s clean:build clean:caches clean:deps",
     "codecov": "codecov",
-    "fix": "run-s fix:lerna fix:biome",
+    "fix": "run-p fix:lerna fix:biome",
     "fix:lerna": "lerna run fix",
     "fix:biome": "biome check --apply .",
     "changelog": "ts-node ./scripts/get-commit-list.ts",
     "link:yarn": "lerna exec yarn link",
-    "lint": "run-s lint:lerna lint:biome",
+    "lint": "run-p lint:lerna lint:biome",
     "lint:lerna": "lerna run lint",
     "lint:biome": "biome check .",
     "validate:es5": "lerna run validate:es5",
diff --git a/packages/bun/src/index.ts b/packages/bun/src/index.ts
index 0166b43d137d..bc29dcd908b5 100644
--- a/packages/bun/src/index.ts
+++ b/packages/bun/src/index.ts
@@ -70,6 +70,7 @@ export {
   startInactiveSpan,
   startSpanManual,
   continueTrace,
+  metrics,
 } from '@sentry/core';
 export type { SpanStatusType } from '@sentry/core';
 export { autoDiscoverNodePerformanceMonitoringIntegrations } from '@sentry/node';
diff --git a/packages/core/src/baseclient.ts b/packages/core/src/baseclient.ts
index 865fdcbd14b5..75b736bbf803 100644
--- a/packages/core/src/baseclient.ts
+++ b/packages/core/src/baseclient.ts
@@ -407,6 +407,7 @@ export abstract class BaseClient<O extends ClientOptions> implements Client<O> {
    * @inheritDoc
    */
   public captureAggregateMetrics(metricBucketItems: Array<MetricBucketItem>): void {
+    DEBUG_BUILD && logger.log(`Flushing aggregated metrics, number of metrics: ${metricBucketItems.length}`);
     const metricsEnvelope = createMetricEnvelope(
       metricBucketItems,
       this._dsn,
diff --git a/packages/core/src/metrics/aggregator.ts b/packages/core/src/metrics/aggregator.ts
new file mode 100644
index 000000000000..6a49fda5918b
--- /dev/null
+++ b/packages/core/src/metrics/aggregator.ts
@@ -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);
+    }
+  }
+}
diff --git a/packages/core/src/metrics/browser-aggregator.ts b/packages/core/src/metrics/browser-aggregator.ts
new file mode 100644
index 000000000000..5b5c81353024
--- /dev/null
+++ b/packages/core/src/metrics/browser-aggregator.ts
@@ -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();
+  }
+}
diff --git a/packages/core/src/metrics/constants.ts b/packages/core/src/metrics/constants.ts
index f29ac323c2ee..e89e0fd1562b 100644
--- a/packages/core/src/metrics/constants.ts
+++ b/packages/core/src/metrics/constants.ts
@@ -27,4 +27,15 @@ 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;
+
+/**
+ * The maximum number of metrics that should be stored in memory.
+ */
+export const MAX_WEIGHT = 10000;
diff --git a/packages/core/src/metrics/envelope.ts b/packages/core/src/metrics/envelope.ts
index c7c65674b736..95622e109740 100644
--- a/packages/core/src/metrics/envelope.ts
+++ b/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',
diff --git a/packages/core/src/metrics/exports.ts b/packages/core/src/metrics/exports.ts
index 22a5e83ffb3d..66074a7e846c 100644
--- a/packages/core/src/metrics/exports.ts
+++ b/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 client = getClient<BaseClient<ClientOptions>>();
   const scope = getCurrentScope();
@@ -49,7 +49,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);
@@ -58,7 +58,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);
@@ -67,7 +67,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);
@@ -76,7 +76,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);
diff --git a/packages/core/src/metrics/instance.ts b/packages/core/src/metrics/instance.ts
index f071006c96ca..f7d37d8118ed 100644
--- a/packages/core/src/metrics/instance.ts
+++ b/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,6 +102,11 @@ 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);
@@ -94,14 +114,12 @@ export class SetMetric implements MetricInstance {
 
   /** @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,
diff --git a/packages/core/src/metrics/integration.ts b/packages/core/src/metrics/integration.ts
index 0c3fa626a9e5..531b0aa698b2 100644
--- a/packages/core/src/metrics/integration.ts
+++ b/packages/core/src/metrics/integration.ts
@@ -1,7 +1,7 @@
 import type { ClientOptions, IntegrationFn } from '@sentry/types';
 import type { BaseClient } from '../baseclient';
 import { convertIntegrationFnToClass } from '../integration';
-import { SimpleMetricsAggregator } from './simpleaggregator';
+import { BrowserMetricsAggregator } from './browser-aggregator';
 
 const INTEGRATION_NAME = 'MetricsAggregator';
 
@@ -9,7 +9,7 @@ const metricsAggregatorIntegration: IntegrationFn = () => {
   return {
     name: INTEGRATION_NAME,
     setup(client: BaseClient<ClientOptions>) {
-      client.metricsAggregator = new SimpleMetricsAggregator(client);
+      client.metricsAggregator = new BrowserMetricsAggregator(client);
     },
   };
 };
diff --git a/packages/core/src/metrics/simpleaggregator.ts b/packages/core/src/metrics/simpleaggregator.ts
deleted file mode 100644
index a628a3b5a406..000000000000
--- a/packages/core/src/metrics/simpleaggregator.ts
+++ /dev/null
@@ -1,91 +0,0 @@
-import type { Client, ClientOptions, MeasurementUnit, MetricsAggregator, Primitive } from '@sentry/types';
-import { timestampInSeconds } from '@sentry/utils';
-import {
-  DEFAULT_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';
-
-/**
- * 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;
-  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);
-  }
-
-  /**
-   * @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);
-    const bucketItem = this._buckets.get(bucketKey);
-    if (bucketItem) {
-      const [bucketMetric, bucketTimestamp] = bucketItem;
-      bucketMetric.add(value);
-      // TODO(abhi): Do we need this check?
-      if (bucketTimestamp < timestamp) {
-        bucketItem[1] = 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]);
-    }
-  }
-
-  /**
-   * @inheritDoc
-   */
-  public flush(): void {
-    // short circuit if buckets are empty.
-    if (this._buckets.size === 0) {
-      return;
-    }
-    if (this._client.captureAggregateMetrics) {
-      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();
-  }
-}
-
-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;
-}
diff --git a/packages/core/src/metrics/types.ts b/packages/core/src/metrics/types.ts
index de6032f811b8..000c401e7a34 100644
--- a/packages/core/src/metrics/types.ts
+++ b/packages/core/src/metrics/types.ts
@@ -7,4 +7,6 @@ export type MetricType =
   | typeof SET_METRIC_TYPE
   | typeof DISTRIBUTION_METRIC_TYPE;
 
-export type SimpleMetricBucket = Map<string, MetricBucketItem>;
+// TODO(@anonrig): Convert this to WeakMap when we support ES6 and
+// use FinalizationRegistry to flush the buckets when the aggregator is garbage collected.
+export type MetricBucket = Map<string, MetricBucketItem>;
diff --git a/packages/core/src/metrics/utils.ts b/packages/core/src/metrics/utils.ts
index 6e4d75fee5f6..a6674bcf30e1 100644
--- a/packages/core/src/metrics/utils.ts
+++ b/packages/core/src/metrics/utils.ts
@@ -1,5 +1,6 @@
-import type { MeasurementUnit, MetricBucketItem } from '@sentry/types';
+import type { MeasurementUnit, MetricBucketItem, Primitive } from '@sentry/types';
 import { dropUndefinedKeys } from '@sentry/utils';
+import { NAME_AND_TAG_KEY_NORMALIZATION_REGEX, TAG_VALUE_NORMALIZATION_REGEX } from './constants';
 import type { MetricType } from './types';
 
 /**
@@ -43,15 +44,26 @@ export function simpleHash(s: string): number {
  * tags: { a: value, b: anothervalue }
  * timestamp: 12345677
  */
-export function serializeMetricBuckets(metricBucketItems: Array<MetricBucketItem>): string {
+export function serializeMetricBuckets(metricBucketItems: MetricBucketItem[]): string {
   let out = '';
-  for (const [metric, timestamp, metricType, name, unit, tags] of metricBucketItems) {
-    const maybeTags = Object.keys(tags).length
-      ? `|#${Object.entries(tags)
-          .map(([key, value]) => `${key}:${String(value)}`)
-          .join(',')}`
-      : '';
-    out += `${name}@${unit}:${metric}|${metricType}${maybeTags}|T${timestamp}\n`;
+  for (const item of metricBucketItems) {
+    const tagEntries = Object.entries(item.tags);
+    const maybeTags = tagEntries.length > 0 ? `|#${tagEntries.map(([key, value]) => `${key}:${value}`).join(',')}` : '';
+    out += `${item.name}@${item.unit}:${item.metric}|${item.metricType}${maybeTags}|T${item.timestamp}\n`;
   }
   return out;
 }
+
+/**
+ * Sanitizes tags.
+ */
+export 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;
+}
diff --git a/packages/core/src/server-runtime-client.ts b/packages/core/src/server-runtime-client.ts
index f4abd134223f..66d846c23911 100644
--- a/packages/core/src/server-runtime-client.ts
+++ b/packages/core/src/server-runtime-client.ts
@@ -17,6 +17,7 @@ import { BaseClient } from './baseclient';
 import { createCheckInEnvelope } from './checkin';
 import { DEBUG_BUILD } from './debug-build';
 import { getClient } from './exports';
+import { MetricsAggregator } from './metrics/aggregator';
 import type { Scope } from './scope';
 import { SessionFlusher } from './sessionflusher';
 import { addTracingExtensions, getDynamicSamplingContextFromClient } from './tracing';
@@ -44,6 +45,10 @@ export class ServerRuntimeClient<
     addTracingExtensions();
 
     super(options);
+
+    if (options._experiments && options._experiments['metricsAggregator']) {
+      this.metricsAggregator = new MetricsAggregator(this);
+    }
   }
 
   /**
diff --git a/packages/core/test/lib/metrics/aggregator.test.ts b/packages/core/test/lib/metrics/aggregator.test.ts
new file mode 100644
index 000000000000..32396cfedcc2
--- /dev/null
+++ b/packages/core/test/lib/metrics/aggregator.test.ts
@@ -0,0 +1,157 @@
+import { MetricsAggregator } from '../../../src/metrics/aggregator';
+import { MAX_WEIGHT } from '../../../src/metrics/constants';
+import { CounterMetric } from '../../../src/metrics/instance';
+import { serializeMetricBuckets } from '../../../src/metrics/utils';
+import { TestClient, getDefaultTestClientOptions } from '../../mocks/client';
+
+let testClient: TestClient;
+
+describe('MetricsAggregator', () => {
+  const options = getDefaultTestClientOptions({ tracesSampleRate: 0.0 });
+
+  beforeEach(() => {
+    jest.useFakeTimers('legacy');
+    testClient = new TestClient(options);
+  });
+
+  it('adds items to buckets', () => {
+    const aggregator = new MetricsAggregator(testClient);
+    aggregator.add('c', 'requests', 1);
+    expect(aggregator['_buckets'].size).toEqual(1);
+
+    const firstValue = aggregator['_buckets'].values().next().value;
+    expect(firstValue).toEqual({
+      metric: expect.any(CounterMetric),
+      metricType: 'c',
+      name: 'requests',
+      tags: {},
+      timestamp: expect.any(Number),
+      unit: 'none',
+    });
+  });
+
+  it('groups same items together', () => {
+    const aggregator = new MetricsAggregator(testClient);
+    aggregator.add('c', 'requests', 1);
+    expect(aggregator['_buckets'].size).toEqual(1);
+    aggregator.add('c', 'requests', 1);
+    expect(aggregator['_buckets'].size).toEqual(1);
+
+    const firstValue = aggregator['_buckets'].values().next().value;
+    expect(firstValue).toEqual({
+      metric: expect.any(CounterMetric),
+      metricType: 'c',
+      name: 'requests',
+      tags: {},
+      timestamp: expect.any(Number),
+      unit: 'none',
+    });
+    expect(firstValue.metric._value).toEqual(2);
+  });
+
+  it('differentiates based on tag value', () => {
+    const aggregator = new MetricsAggregator(testClient);
+    aggregator.add('g', 'cpu', 50);
+    expect(aggregator['_buckets'].size).toEqual(1);
+    aggregator.add('g', 'cpu', 55, undefined, { a: 'value' });
+    expect(aggregator['_buckets'].size).toEqual(2);
+  });
+
+  describe('serializeBuckets', () => {
+    it('serializes ', () => {
+      const aggregator = new MetricsAggregator(testClient);
+      aggregator.add('c', 'requests', 8);
+      aggregator.add('g', 'cpu', 50);
+      aggregator.add('g', 'cpu', 55);
+      aggregator.add('g', 'cpu', 52);
+      aggregator.add('d', 'lcp', 1, 'second', { a: 'value', b: 'anothervalue' });
+      aggregator.add('d', 'lcp', 1.2, 'second', { a: 'value', b: 'anothervalue' });
+      aggregator.add('s', 'important_people', 'a', 'none', { numericKey: 2 });
+      aggregator.add('s', 'important_people', 'b', 'none', { numericKey: 2 });
+
+      const metricBuckets = Array.from(aggregator['_buckets']).map(([, bucketItem]) => bucketItem);
+      const serializedBuckets = serializeMetricBuckets(metricBuckets);
+
+      expect(serializedBuckets).toContain('requests@none:8|c|T');
+      expect(serializedBuckets).toContain('cpu@none:52:50:55:157:3|g|T');
+      expect(serializedBuckets).toContain('lcp@second:1:1.2|d|#a:value,b:anothervalue|T');
+      expect(serializedBuckets).toContain('important_people@none:97:98|s|#numericKey:2|T');
+    });
+  });
+
+  describe('close', () => {
+    test('should flush immediately', () => {
+      const capture = jest.spyOn(testClient, 'captureAggregateMetrics');
+      const aggregator = new MetricsAggregator(testClient);
+      aggregator.add('c', 'requests', 1);
+      aggregator.close();
+      // It should clear the interval.
+      expect(clearInterval).toHaveBeenCalled();
+      expect(capture).toBeCalled();
+      expect(capture).toBeCalledTimes(1);
+      expect(capture).toBeCalledWith([
+        {
+          metric: { _value: 1 },
+          metricType: 'c',
+          name: 'requests',
+          tags: {},
+          timestamp: expect.any(Number),
+          unit: 'none',
+        },
+      ]);
+    });
+  });
+
+  describe('flush', () => {
+    test('should flush immediately', () => {
+      const capture = jest.spyOn(testClient, 'captureAggregateMetrics');
+      const aggregator = new MetricsAggregator(testClient);
+      aggregator.add('c', 'requests', 1);
+      aggregator.flush();
+      expect(capture).toBeCalled();
+      expect(capture).toBeCalledTimes(1);
+      expect(capture).toBeCalledWith([
+        {
+          metric: { _value: 1 },
+          metricType: 'c',
+          name: 'requests',
+          tags: {},
+          timestamp: expect.any(Number),
+          unit: 'none',
+        },
+      ]);
+      capture.mockReset();
+      aggregator.close();
+      // It should clear the interval.
+      expect(clearInterval).toHaveBeenCalled();
+
+      // It shouldn't be called since it's been already flushed.
+      expect(capture).toBeCalledTimes(0);
+    });
+
+    test('should not capture if empty', () => {
+      const capture = jest.spyOn(testClient, 'captureAggregateMetrics');
+      const aggregator = new MetricsAggregator(testClient);
+      aggregator.add('c', 'requests', 1);
+      aggregator.flush();
+      expect(capture).toBeCalledTimes(1);
+      capture.mockReset();
+      aggregator.close();
+      expect(capture).toBeCalledTimes(0);
+    });
+  });
+
+  describe('add', () => {
+    test('it should respect the max weight and flush if exceeded', () => {
+      const capture = jest.spyOn(testClient, 'captureAggregateMetrics');
+      const aggregator = new MetricsAggregator(testClient);
+
+      for (let i = 0; i < MAX_WEIGHT; i++) {
+        aggregator.add('c', 'requests', 1);
+      }
+
+      expect(capture).toBeCalledTimes(1);
+      aggregator.close();
+    });
+  });
+});
diff --git a/packages/core/test/lib/metrics/simpleaggregator.test.ts b/packages/core/test/lib/metrics/browser-aggregator.test.ts
similarity index 71%
rename from packages/core/test/lib/metrics/simpleaggregator.test.ts
rename to packages/core/test/lib/metrics/browser-aggregator.test.ts
index cafc78d1e018..669959a03e05 100644
--- a/packages/core/test/lib/metrics/simpleaggregator.test.ts
+++ b/packages/core/test/lib/metrics/browser-aggregator.test.ts
@@ -1,36 +1,49 @@
+import { BrowserMetricsAggregator } from '../../../src/metrics/browser-aggregator';
 import { CounterMetric } from '../../../src/metrics/instance';
-import { SimpleMetricsAggregator } from '../../../src/metrics/simpleaggregator';
 import { serializeMetricBuckets } from '../../../src/metrics/utils';
 import { TestClient, getDefaultTestClientOptions } from '../../mocks/client';
 
-describe('SimpleMetricsAggregator', () => {
+describe('BrowserMetricsAggregator', () => {
   const options = getDefaultTestClientOptions({ tracesSampleRate: 0.0 });
   const testClient = new TestClient(options);
 
   it('adds items to buckets', () => {
-    const aggregator = new SimpleMetricsAggregator(testClient);
+    const aggregator = new BrowserMetricsAggregator(testClient);
     aggregator.add('c', 'requests', 1);
     expect(aggregator['_buckets'].size).toEqual(1);
 
     const firstValue = aggregator['_buckets'].values().next().value;
-    expect(firstValue).toEqual([expect.any(CounterMetric), expect.any(Number), 'c', 'requests', 'none', {}]);
+    expect(firstValue).toEqual({
+      metric: expect.any(CounterMetric),
+      metricType: 'c',
+      name: 'requests',
+      tags: {},
+      timestamp: expect.any(Number),
+      unit: 'none',
+    });
   });
 
   it('groups same items together', () => {
-    const aggregator = new SimpleMetricsAggregator(testClient);
+    const aggregator = new BrowserMetricsAggregator(testClient);
     aggregator.add('c', 'requests', 1);
     expect(aggregator['_buckets'].size).toEqual(1);
     aggregator.add('c', 'requests', 1);
     expect(aggregator['_buckets'].size).toEqual(1);
 
     const firstValue = aggregator['_buckets'].values().next().value;
-    expect(firstValue).toEqual([expect.any(CounterMetric), expect.any(Number), 'c', 'requests', 'none', {}]);
-
-    expect(firstValue[0]._value).toEqual(2);
+    expect(firstValue).toEqual({
+      metric: expect.any(CounterMetric),
+      metricType: 'c',
+      name: 'requests',
+      tags: {},
+      timestamp: expect.any(Number),
+      unit: 'none',
+    });
+    expect(firstValue.metric._value).toEqual(2);
   });
 
   it('differentiates based on tag value', () => {
-    const aggregator = new SimpleMetricsAggregator(testClient);
+    const aggregator = new BrowserMetricsAggregator(testClient);
     aggregator.add('g', 'cpu', 50);
     expect(aggregator['_buckets'].size).toEqual(1);
     aggregator.add('g', 'cpu', 55, undefined, { a: 'value' });
@@ -39,7 +52,7 @@ describe('SimpleMetricsAggregator', () => {
 
   describe('serializeBuckets', () => {
     it('serializes ', () => {
-      const aggregator = new SimpleMetricsAggregator(testClient);
+      const aggregator = new BrowserMetricsAggregator(testClient);
       aggregator.add('c', 'requests', 8);
       aggregator.add('g', 'cpu', 50);
       aggregator.add('g', 'cpu', 55);
diff --git a/packages/deno/src/index.ts b/packages/deno/src/index.ts
index 4a57bb6f2cfd..fe0b5ee4b620 100644
--- a/packages/deno/src/index.ts
+++ b/packages/deno/src/index.ts
@@ -68,6 +68,7 @@ export {
   startSpan,
   startInactiveSpan,
   startSpanManual,
+  metrics,
 } from '@sentry/core';
 export type { SpanStatusType } from '@sentry/core';
 
diff --git a/packages/node/src/index.ts b/packages/node/src/index.ts
index 46b1d6d742d4..36d2d8beac53 100644
--- a/packages/node/src/index.ts
+++ b/packages/node/src/index.ts
@@ -71,6 +71,7 @@ export {
   startInactiveSpan,
   startSpanManual,
   continueTrace,
+  metrics,
 } from '@sentry/core';
 export type { SpanStatusType } from '@sentry/core';
 export { autoDiscoverNodePerformanceMonitoringIntegrations } from './tracing';
diff --git a/packages/types/src/metrics.ts b/packages/types/src/metrics.ts
index 18943ee3997e..9bfb990461eb 100644
--- a/packages/types/src/metrics.ts
+++ b/packages/types/src/metrics.ts
@@ -1,25 +1,41 @@
 import type { MeasurementUnit } from './measurement';
 import type { Primitive } from './misc';
 
-export interface MetricInstance {
+/**
+ * An abstract definition of the minimum required API
+ * for a metric instance.
+ */
+export abstract class MetricInstance {
+  /**
+   * Returns the weight of the metric.
+   */
+  public get weight(): number {
+    return 1;
+  }
+
   /**
    * Adds a value to a metric.
    */
-  add(value: number | string): void;
+  public add(value: number | string): void {
+    // Override this.
+  }
+
   /**
    * Serializes the metric into a statsd format string.
    */
-  toString(): string;
+  public toString(): string {
+    return '';
+  }
 }
 
-export type MetricBucketItem = [
-  metric: MetricInstance,
-  timestamp: number,
-  metricType: 'c' | 'g' | 's' | 'd',
-  name: string,
-  unit: MeasurementUnit,
-  tags: { [key: string]: string },
-];
+export interface MetricBucketItem {
+  metric: MetricInstance;
+  timestamp: number;
+  metricType: 'c' | 'g' | 's' | 'd';
+  name: string;
+  unit: MeasurementUnit;
+  tags: Record<string, string>;
+}
 
 /**
  * A metrics aggregator that aggregates metrics in memory and flushes them periodically.
diff --git a/packages/vercel-edge/src/index.ts b/packages/vercel-edge/src/index.ts
index cbc7f6a89d7c..2ef1217ab117 100644
--- a/packages/vercel-edge/src/index.ts
+++ b/packages/vercel-edge/src/index.ts
@@ -69,6 +69,7 @@ export {
   startInactiveSpan,
   startSpanManual,
   continueTrace,
+  metrics,
 } from '@sentry/core';
 export type { SpanStatusType } from '@sentry/core';