Skip to content

Commit fe37e71

Browse files
cijothomascijothomas
authored
Merge branch 'main' into ban/context-suppression
2 parents 3416528 + e9ae9f9 commit fe37e71

20 files changed

+365
-94
lines changed

‎opentelemetry-sdk/CHANGELOG.md

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,15 @@ the suppression flag in their dedicated thread, so that telemetry generated from
1212
those threads will not be fed back into OTel. Similarly, `SimpleLogProcessor`
1313
also modified to suppress telemetry before invoking exporters.
1414

15+
- **Feature**: Implemented and enabled cardinality capping for Metrics by
16+
default.
17+
- The default cardinality limit is 2000 and can be customized using Views.
18+
- This feature was previously removed in version 0.28 due to the lack of
19+
configurability but has now been reintroduced with the ability to configure
20+
the limit.
21+
- TODO/Placeholder: Add ability to configure cardinality limits via Instrument
22+
advisory.
23+
1524
## 0.29.0
1625

1726
Released 2025-Mar-21
@@ -69,6 +78,7 @@ Released 2025-Mar-21
6978
Custom exporters will need to internally synchronize any mutable state, if applicable.
7079

7180
- **Breaking** The `shutdown_with_timeout` method is added to MetricExporter trait. This is breaking change for custom `MetricExporter` authors.
81+
- **Breaking** The `shutdown_with_timeout` method is added to MetricReader trait. This is breaking change for custom `MetricReader` authors.
7282
- Bug Fix: `BatchLogProcessor` now correctly calls `shutdown` on the exporter
7383
when its `shutdown` is invoked.
7484

‎opentelemetry-sdk/benches/metric.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,3 @@
1-
use rand::Rng;
2-
use std::sync::{Arc, Weak};
3-
41
use criterion::{criterion_group, criterion_main, Bencher, Criterion};
52
use opentelemetry::{
63
metrics::{Counter, Histogram, MeterProvider as _},
@@ -15,6 +12,9 @@ use opentelemetry_sdk::{
1512
},
1613
Resource,
1714
};
15+
use rand::Rng;
16+
use std::sync::{Arc, Weak};
17+
use std::time::Duration;
1818

1919
#[derive(Clone, Debug)]
2020
struct SharedReader(Arc<dyn MetricReader>);
@@ -32,7 +32,7 @@ impl MetricReader for SharedReader {
3232
self.0.force_flush()
3333
}
3434

35-
fn shutdown(&self) -> OTelSdkResult {
35+
fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
3636
self.0.shutdown()
3737
}
3838

‎opentelemetry-sdk/src/metrics/instrument.rs

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,9 @@ pub struct Stream {
203203
/// dropped. If the set is empty, all attributes will be dropped, if `None` all
204204
/// attributes will be kept.
205205
pub allowed_attribute_keys: Option<Arc<HashSet<Key>>>,
206+
207+
/// Cardinality limit for the stream.
208+
pub cardinality_limit: Option<usize>,
206209
}
207210

208211
#[cfg(feature = "spec_unstable_metrics_views")]
@@ -245,6 +248,12 @@ impl Stream {
245248

246249
self
247250
}
251+
252+
/// Set the stream cardinality limit.
253+
pub fn cardinality_limit(mut self, limit: usize) -> Self {
254+
self.cardinality_limit = Some(limit);
255+
self
256+
}
248257
}
249258

250259
/// The identifying properties of an instrument.

‎opentelemetry-sdk/src/metrics/internal/aggregate.rs

Lines changed: 40 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,6 @@ use super::{
1515
precomputed_sum::PrecomputedSum, sum::Sum, Number,
1616
};
1717

18-
pub(crate) const STREAM_CARDINALITY_LIMIT: usize = 2000;
19-
20-
/// Checks whether aggregator has hit cardinality limit for metric streams
21-
pub(crate) fn is_under_cardinality_limit(_size: usize) -> bool {
22-
true
23-
24-
// TODO: Implement this feature, after allowing the ability to customize the cardinality limit.
25-
// size < STREAM_CARDINALITY_LIMIT
26-
}
27-
2818
/// Receives measurements to be aggregated.
2919
pub(crate) trait Measure<T>: Send + Sync + 'static {
3020
fn call(&self, measurement: T, attrs: &[KeyValue]);
@@ -133,14 +123,22 @@ pub(crate) struct AggregateBuilder<T> {
133123
/// measurements.
134124
filter: AttributeSetFilter,
135125

126+
/// Cardinality limit for the metric stream
127+
cardinality_limit: usize,
128+
136129
_marker: marker::PhantomData<T>,
137130
}
138131

139132
impl<T: Number> AggregateBuilder<T> {
140-
pub(crate) fn new(temporality: Temporality, filter: Option<Filter>) -> Self {
133+
pub(crate) fn new(
134+
temporality: Temporality,
135+
filter: Option<Filter>,
136+
cardinality_limit: usize,
137+
) -> Self {
141138
AggregateBuilder {
142139
temporality,
143140
filter: AttributeSetFilter::new(filter),
141+
cardinality_limit,
144142
_marker: marker::PhantomData,
145143
}
146144
}
@@ -150,18 +148,31 @@ impl<T: Number> AggregateBuilder<T> {
150148
LastValue::new(
151149
overwrite_temporality.unwrap_or(self.temporality),
152150
self.filter.clone(),
151+
self.cardinality_limit,
153152
)
154153
.into()
155154
}
156155

157156
/// Builds a precomputed sum aggregate function input and output.
158157
pub(crate) fn precomputed_sum(&self, monotonic: bool) -> AggregateFns<T> {
159-
PrecomputedSum::new(self.temporality, self.filter.clone(), monotonic).into()
158+
PrecomputedSum::new(
159+
self.temporality,
160+
self.filter.clone(),
161+
monotonic,
162+
self.cardinality_limit,
163+
)
164+
.into()
160165
}
161166

162167
/// Builds a sum aggregate function input and output.
163168
pub(crate) fn sum(&self, monotonic: bool) -> AggregateFns<T> {
164-
Sum::new(self.temporality, self.filter.clone(), monotonic).into()
169+
Sum::new(
170+
self.temporality,
171+
self.filter.clone(),
172+
monotonic,
173+
self.cardinality_limit,
174+
)
175+
.into()
165176
}
166177

167178
/// Builds a histogram aggregate function input and output.
@@ -177,6 +188,7 @@ impl<T: Number> AggregateBuilder<T> {
177188
boundaries,
178189
record_min_max,
179190
record_sum,
191+
self.cardinality_limit,
180192
)
181193
.into()
182194
}
@@ -196,6 +208,7 @@ impl<T: Number> AggregateBuilder<T> {
196208
max_scale,
197209
record_min_max,
198210
record_sum,
211+
self.cardinality_limit,
199212
)
200213
.into()
201214
}
@@ -211,10 +224,13 @@ mod tests {
211224

212225
use super::*;
213226

227+
const CARDINALITY_LIMIT_DEFAULT: usize = 2000;
228+
214229
#[test]
215230
fn last_value_aggregation() {
216231
let AggregateFns { measure, collect } =
217-
AggregateBuilder::<u64>::new(Temporality::Cumulative, None).last_value(None);
232+
AggregateBuilder::<u64>::new(Temporality::Cumulative, None, CARDINALITY_LIMIT_DEFAULT)
233+
.last_value(None);
218234
let mut a = MetricData::Gauge(Gauge {
219235
data_points: vec![GaugeDataPoint {
220236
attributes: vec![KeyValue::new("a", 1)],
@@ -244,7 +260,8 @@ mod tests {
244260
fn precomputed_sum_aggregation() {
245261
for temporality in [Temporality::Delta, Temporality::Cumulative] {
246262
let AggregateFns { measure, collect } =
247-
AggregateBuilder::<u64>::new(temporality, None).precomputed_sum(true);
263+
AggregateBuilder::<u64>::new(temporality, None, CARDINALITY_LIMIT_DEFAULT)
264+
.precomputed_sum(true);
248265
let mut a = MetricData::Sum(Sum {
249266
data_points: vec![
250267
SumDataPoint {
@@ -290,7 +307,8 @@ mod tests {
290307
fn sum_aggregation() {
291308
for temporality in [Temporality::Delta, Temporality::Cumulative] {
292309
let AggregateFns { measure, collect } =
293-
AggregateBuilder::<u64>::new(temporality, None).sum(true);
310+
AggregateBuilder::<u64>::new(temporality, None, CARDINALITY_LIMIT_DEFAULT)
311+
.sum(true);
294312
let mut a = MetricData::Sum(Sum {
295313
data_points: vec![
296314
SumDataPoint {
@@ -335,8 +353,9 @@ mod tests {
335353
#[test]
336354
fn explicit_bucket_histogram_aggregation() {
337355
for temporality in [Temporality::Delta, Temporality::Cumulative] {
338-
let AggregateFns { measure, collect } = AggregateBuilder::<u64>::new(temporality, None)
339-
.explicit_bucket_histogram(vec![1.0], true, true);
356+
let AggregateFns { measure, collect } =
357+
AggregateBuilder::<u64>::new(temporality, None, CARDINALITY_LIMIT_DEFAULT)
358+
.explicit_bucket_histogram(vec![1.0], true, true);
340359
let mut a = MetricData::Histogram(Histogram {
341360
data_points: vec![HistogramDataPoint {
342361
attributes: vec![KeyValue::new("a1", 1)],
@@ -382,8 +401,9 @@ mod tests {
382401
#[test]
383402
fn exponential_histogram_aggregation() {
384403
for temporality in [Temporality::Delta, Temporality::Cumulative] {
385-
let AggregateFns { measure, collect } = AggregateBuilder::<u64>::new(temporality, None)
386-
.exponential_bucket_histogram(4, 20, true, true);
404+
let AggregateFns { measure, collect } =
405+
AggregateBuilder::<u64>::new(temporality, None, CARDINALITY_LIMIT_DEFAULT)
406+
.exponential_bucket_histogram(4, 20, true, true);
387407
let mut a = MetricData::ExponentialHistogram(ExponentialHistogram {
388408
data_points: vec![ExponentialHistogramDataPoint {
389409
attributes: vec![KeyValue::new("a1", 1)],

‎opentelemetry-sdk/src/metrics/internal/exponential_histogram.rs

Lines changed: 52 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -369,12 +369,16 @@ impl<T: Number> ExpoHistogram<T> {
369369
max_scale: i8,
370370
record_min_max: bool,
371371
record_sum: bool,
372+
cardinality_limit: usize,
372373
) -> Self {
373374
ExpoHistogram {
374-
value_map: ValueMap::new(BucketConfig {
375-
max_size: max_size as i32,
376-
max_scale,
377-
}),
375+
value_map: ValueMap::new(
376+
BucketConfig {
377+
max_size: max_size as i32,
378+
max_scale,
379+
},
380+
cardinality_limit,
381+
),
378382
init_time: AggregateTimeInitiator::default(),
379383
temporality,
380384
filter,
@@ -546,6 +550,8 @@ mod tests {
546550

547551
use super::*;
548552

553+
const CARDINALITY_LIMIT_DEFAULT: usize = 2000;
554+
549555
#[test]
550556
fn test_expo_histogram_data_point_record() {
551557
run_data_point_record::<f64>();
@@ -710,6 +716,7 @@ mod tests {
710716
20,
711717
true,
712718
true,
719+
CARDINALITY_LIMIT_DEFAULT,
713720
);
714721
for v in test.values {
715722
Measure::call(&h, v, &[]);
@@ -766,6 +773,7 @@ mod tests {
766773
20,
767774
true,
768775
true,
776+
CARDINALITY_LIMIT_DEFAULT,
769777
);
770778
for v in test.values {
771779
Measure::call(&h, v, &[]);
@@ -1278,12 +1286,13 @@ mod tests {
12781286
TestCase {
12791287
name: "Delta Single",
12801288
build: Box::new(move || {
1281-
AggregateBuilder::new(Temporality::Delta, None).exponential_bucket_histogram(
1282-
max_size,
1283-
max_scale,
1284-
record_min_max,
1285-
record_sum,
1286-
)
1289+
AggregateBuilder::new(Temporality::Delta, None, CARDINALITY_LIMIT_DEFAULT)
1290+
.exponential_bucket_histogram(
1291+
max_size,
1292+
max_scale,
1293+
record_min_max,
1294+
record_sum,
1295+
)
12871296
}),
12881297
input: vec![vec![4, 4, 4, 2, 16, 1]
12891298
.into_iter()
@@ -1318,13 +1327,17 @@ mod tests {
13181327
TestCase {
13191328
name: "Cumulative Single",
13201329
build: Box::new(move || {
1321-
internal::AggregateBuilder::new(Temporality::Cumulative, None)
1322-
.exponential_bucket_histogram(
1323-
max_size,
1324-
max_scale,
1325-
record_min_max,
1326-
record_sum,
1327-
)
1330+
internal::AggregateBuilder::new(
1331+
Temporality::Cumulative,
1332+
None,
1333+
CARDINALITY_LIMIT_DEFAULT,
1334+
)
1335+
.exponential_bucket_histogram(
1336+
max_size,
1337+
max_scale,
1338+
record_min_max,
1339+
record_sum,
1340+
)
13281341
}),
13291342
input: vec![vec![4, 4, 4, 2, 16, 1]
13301343
.into_iter()
@@ -1359,13 +1372,17 @@ mod tests {
13591372
TestCase {
13601373
name: "Delta Multiple",
13611374
build: Box::new(move || {
1362-
internal::AggregateBuilder::new(Temporality::Delta, None)
1363-
.exponential_bucket_histogram(
1364-
max_size,
1365-
max_scale,
1366-
record_min_max,
1367-
record_sum,
1368-
)
1375+
internal::AggregateBuilder::new(
1376+
Temporality::Delta,
1377+
None,
1378+
CARDINALITY_LIMIT_DEFAULT,
1379+
)
1380+
.exponential_bucket_histogram(
1381+
max_size,
1382+
max_scale,
1383+
record_min_max,
1384+
record_sum,
1385+
)
13691386
}),
13701387
input: vec![
13711388
vec![2, 3, 8].into_iter().map(Into::into).collect(),
@@ -1403,13 +1420,17 @@ mod tests {
14031420
TestCase {
14041421
name: "Cumulative Multiple ",
14051422
build: Box::new(move || {
1406-
internal::AggregateBuilder::new(Temporality::Cumulative, None)
1407-
.exponential_bucket_histogram(
1408-
max_size,
1409-
max_scale,
1410-
record_min_max,
1411-
record_sum,
1412-
)
1423+
internal::AggregateBuilder::new(
1424+
Temporality::Cumulative,
1425+
None,
1426+
CARDINALITY_LIMIT_DEFAULT,
1427+
)
1428+
.exponential_bucket_histogram(
1429+
max_size,
1430+
max_scale,
1431+
record_min_max,
1432+
record_sum,
1433+
)
14131434
}),
14141435
input: vec![
14151436
vec![2, 3, 8].into_iter().map(Into::into).collect(),

‎opentelemetry-sdk/src/metrics/internal/histogram.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@ impl<T: Number> Histogram<T> {
8787
mut bounds: Vec<f64>,
8888
record_min_max: bool,
8989
record_sum: bool,
90+
cardinality_limit: usize,
9091
) -> Self {
9192
#[cfg(feature = "spec_unstable_metrics_views")]
9293
{
@@ -97,7 +98,7 @@ impl<T: Number> Histogram<T> {
9798

9899
let buckets_count = bounds.len() + 1;
99100
Histogram {
100-
value_map: ValueMap::new(buckets_count),
101+
value_map: ValueMap::new(buckets_count, cardinality_limit),
101102
init_time: AggregateTimeInitiator::default(),
102103
temporality,
103104
filter,
@@ -262,6 +263,7 @@ mod tests {
262263
vec![1.0, 3.0, 6.0],
263264
false,
264265
false,
266+
2000,
265267
);
266268
for v in 1..11 {
267269
Measure::call(&hist, v, &[]);

0 commit comments

Comments
 (0)