Skip to content

Commit b722a0e

Browse files
committed
fix(metrics): dedup metrics v3 tags (#1817)
## Summary <!-- Please provide a brief summary about what this PR does. This should help the reviewers give feedback faster and with higher quality. --> Fixes logical mismatches found when testing the v3 pipeline in validation mode where v2 tag hashes differed from v3 the v3 tag hash. ## Change Type - [x] Bug fix - [ ] New feature - [ ] Non-functional (chore, refactoring, docs) - [ ] Performance ## How did you test this PR? <!-- Please how you tested these changes here --> ## References <!-- Please list any issues closed by this PR. --> <!-- - Closes: <issue link> --> <!-- Any other issues or PRs relevant to this PR? Feel free to list them here. -->
1 parent a5c01cf commit b722a0e

1 file changed

Lines changed: 14 additions & 7 deletions

File tree

  • lib/saluki-components/src/encoders/datadog/metrics

lib/saluki-components/src/encoders/datadog/metrics/mod.rs

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ use protobuf::{rt::WireType, CodedOutputStream};
88
use resource_accounting::{MemoryBounds, MemoryBoundsBuilder};
99
use saluki_common::{
1010
buf::{ChunkedBytesBuffer, FrozenChunkedBytesBuffer},
11+
iter::ReusableDeduplicator,
1112
task::HandleExt as _,
1213
};
1314
use saluki_config::GenericConfiguration;
@@ -1278,9 +1279,10 @@ async fn flush_payload(
12781279
// Encodes a batch of metrics to V3 columnar format.
12791280
fn encode_v3_metrics_batch(metrics: &[Metric], additional_tags: &SharedTagSet) -> Result<Vec<u8>, GenericError> {
12801281
let mut writer = v3::V3Writer::new();
1282+
let mut tags_deduplicator = ReusableDeduplicator::new();
12811283

12821284
for metric in metrics {
1283-
write_metric_to_v3(&mut writer, metric, additional_tags);
1285+
write_metric_to_v3(&mut writer, metric, additional_tags, &mut tags_deduplicator);
12841286
}
12851287

12861288
let mut output = Vec::new();
@@ -1292,7 +1294,10 @@ fn encode_v3_metrics_batch(metrics: &[Metric], additional_tags: &SharedTagSet) -
12921294
}
12931295

12941296
/// Writes a single metric to the V3 writer.
1295-
fn write_metric_to_v3(writer: &mut v3::V3Writer, metric: &Metric, additional_tags: &SharedTagSet) {
1297+
fn write_metric_to_v3(
1298+
writer: &mut v3::V3Writer, metric: &Metric, additional_tags: &SharedTagSet,
1299+
tags_deduplicator: &mut ReusableDeduplicator<Tag>,
1300+
) {
12961301
let metric_type = match metric.values() {
12971302
MetricValues::Counter(..) => v3::V3MetricType::Count,
12981303
MetricValues::Rate(..) => v3::V3MetricType::Rate,
@@ -1304,12 +1309,14 @@ fn write_metric_to_v3(writer: &mut v3::V3Writer, metric: &Metric, additional_tag
13041309
let mut builder = writer.write(metric_type, metric.context().name());
13051310

13061311
// Tags - chain instrumented + additional + origin tags
1307-
let all_tags = metric
1312+
let chained_tags = metric
13081313
.context()
13091314
.tags()
13101315
.into_iter()
13111316
.chain(additional_tags)
1312-
.chain(metric.context().origin_tags())
1317+
.chain(metric.context().origin_tags());
1318+
let all_tags = tags_deduplicator
1319+
.deduplicated(chained_tags)
13131320
.filter(|t| is_sketch || !is_v3_series_resource_tag(t) && !is_v3_series_device_tag(t))
13141321
.map(|t| t.as_str());
13151322
builder.set_tags(all_tags);
@@ -1321,13 +1328,13 @@ fn write_metric_to_v3(writer: &mut v3::V3Writer, metric: &Metric, additional_tag
13211328
}
13221329
if !is_sketch {
13231330
let mut device_resource = None;
1324-
for tag in metric
1331+
let chained_tags = metric
13251332
.context()
13261333
.origin_tags()
13271334
.into_iter()
13281335
.chain(metric.context().tags())
1329-
.chain(additional_tags)
1330-
{
1336+
.chain(additional_tags);
1337+
for tag in tags_deduplicator.deduplicated(chained_tags) {
13311338
if is_v3_series_device_tag(tag) {
13321339
device_resource = tag.value().filter(|device| !device.is_empty());
13331340
} else if is_v3_series_resource_tag(tag) {

0 commit comments

Comments
 (0)