Skip to content

Commit 344fc64

Browse files
authored
azure monitor exporter: optimize transformer with direct JSON serialization (open-telemetry#2318)
# Change Summary - Pre-serialize resource+scope fields once per ScopeLogs as a JSON byte prefix - Write per-record fields directly to byte buffer using itoa/ryu, bypassing serde_json::Map entirely - Add write_json_string, write_json_hex, write_field_value_json for zero-allocation JSON output - Make config and metrics modules public for benchmark access - Add criterion benchmark under contrib-nodes/benches/exporters/azure_monitor_exporter/ - Added contrib bench to rust-bench workflow. Benchmark results (1000 records): Original: 1.60ms (~625K records/s) Hoisted: 1.36ms (~735K records/s) +17% Hoisted + Direct Serialization: 425us (~2.35M records/s) +275% ## What issue does this PR close? ## How are these changes tested? - Existing unit tests cover already mapping uniqueness, also added tests to make sure that overlapping fields are rejected by the config valiation across resource -> scope -> log hiearchy. - Added tests for validating against encoding issues for non ASCII characters. ## Are there any user-facing changes? None.
1 parent eb3862d commit 344fc64

8 files changed

Lines changed: 1031 additions & 128 deletions

File tree

.github/workflows/rust-bench.yml

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,3 +37,21 @@ jobs:
3737
- name: cargo bench ${{ matrix.folder }}
3838
run: cargo bench
3939
working-directory: ./rust/${{ matrix.folder }}
40+
41+
bench-contrib:
42+
# Run feature-gated contrib benchmarks under the same label
43+
if: contains(github.event.pull_request.labels.*.name, 'cargobench')
44+
runs-on: ubuntu-latest
45+
steps:
46+
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
47+
with:
48+
submodules: true
49+
- uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0
50+
with:
51+
repo-token: ${{ secrets.GITHUB_TOKEN }}
52+
- uses: dtolnay/rust-toolchain@efa25f7f19611383d5b0ccf2d1c8914531636bf9
53+
with:
54+
toolchain: stable
55+
- name: cargo bench contrib-nodes (all features)
56+
run: cargo bench -p otap-df-contrib-nodes --all-features
57+
working-directory: ./rust/otap-dataflow

rust/otap-dataflow/Cargo.toml

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ async-channel = { version = "2.5.0" }
9898
http = "1.4"
9999
humantime = "2.3.0"
100100
humantime-serde = "1.1.1"
101+
itoa = "1"
101102
itertools = "0.14.0"
102103
linkme = "0.3.35"
103104
local-sync = "0.1.1"
@@ -141,6 +142,7 @@ quote = "1.0"
141142
rand = "0.10.0"
142143
rcgen = "0.14"
143144
roaring = "0.11.2"
145+
ryu = "1"
144146
rustls = { version = "0.23.22", default-features = false, features = ["std", "tls12"] }
145147
rustls-pki-types = { version = "1.11", features = ["std"] }
146148
tokio-rustls = { version = "0.26", default-features = false, features = ["tls12"] }

rust/otap-dataflow/crates/contrib-nodes/Cargo.toml

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,9 @@ azure_core = { workspace = true, optional = true, features = ["reqwest"] }
4545
azure_identity = { workspace = true, optional = true }
4646
flate2 = { workspace = true, optional = true }
4747
http = { workspace = true, optional = true }
48+
itoa = { workspace = true, optional = true }
4849
rand = { workspace = true, optional = true }
50+
ryu = { workspace = true, optional = true }
4951
reqwest = { workspace = true, optional = true, features = ["rustls-no-provider"] }
5052
sysinfo = { workspace = true, optional = true }
5153

@@ -68,7 +70,9 @@ azure-monitor-exporter = [
6870
"dep:chrono",
6971
"dep:flate2",
7072
"dep:http",
73+
"dep:itoa",
7174
"dep:rand",
75+
"dep:ryu",
7276
]
7377

7478
contrib-processors = [
@@ -85,7 +89,14 @@ recordset-kql-processor = [
8589
resource-validator-processor = []
8690

8791
[dev-dependencies]
92+
criterion = { workspace = true, features = ["html_reports"] }
8893
otap-df-engine = { path = "../engine", features = ["test-utils"] }
8994
otap-df-otap = { path = "../otap", features = ["test-utils"] }
9095
pretty_assertions.workspace = true
9196
wiremock.workspace = true
97+
98+
[[bench]]
99+
name = "azure_monitor_transformer"
100+
path = "benches/exporters/azure_monitor_exporter/transformer.rs"
101+
harness = false
102+
required-features = ["azure-monitor-exporter"]
Lines changed: 242 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,242 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
//! Benchmark for Azure Monitor Transformer: OTLP logs -> JSON conversion.
5+
//!
6+
//! MacOS M4 Pro baseline results (1000 records):
7+
//! Original (per-record resource/scope mapping): 1.60ms (~625K records/s)
8+
//! Hoisted (resource/scope computed once): 1.36ms (~735K records/s) +17%
9+
//! Hoisted + Direct Serialization (current): 425µs (~2.35M records/s) +275%
10+
11+
#![allow(unused_results)]
12+
13+
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
14+
use otap_df_contrib_nodes::exporters::azure_monitor_exporter::{Config, Transformer};
15+
use otap_df_pdata::views::otlp::bytes::logs::RawLogsData;
16+
use prost::Message;
17+
use serde_json::json;
18+
use std::collections::HashMap;
19+
20+
use otap_df_pdata::proto::opentelemetry::common::v1::{
21+
AnyValue, InstrumentationScope, KeyValue, any_value::Value as OtelAnyValueEnum,
22+
};
23+
use otap_df_pdata::proto::opentelemetry::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
24+
use otap_df_pdata::proto::opentelemetry::resource::v1::Resource;
25+
26+
use otap_df_pdata::proto::opentelemetry::collector::logs::v1::ExportLogsServiceRequest;
27+
28+
fn create_config() -> Config {
29+
use otap_df_contrib_nodes::exporters::azure_monitor_exporter::config::{
30+
ApiConfig, AuthConfig, SchemaConfig,
31+
};
32+
33+
Config {
34+
api: ApiConfig {
35+
dcr_endpoint: "https://test.ingest.monitor.azure.com".into(),
36+
stream_name: "Custom-TestTable".into(),
37+
dcr: "dcr-test-rule-id".into(),
38+
schema: SchemaConfig {
39+
resource_mapping: HashMap::from([
40+
("service.name".into(), "ServiceName".into()),
41+
("service.version".into(), "ServiceVersion".into()),
42+
("host.name".into(), "HostName".into()),
43+
]),
44+
scope_mapping: HashMap::from([
45+
("scope.name".into(), "ScopeName".into()),
46+
("scope.version".into(), "ScopeVersion".into()),
47+
]),
48+
log_record_mapping: HashMap::from([
49+
("time_unix_nano".into(), json!("TimeGenerated")),
50+
("severity_text".into(), json!("SeverityText")),
51+
("severity_number".into(), json!("SeverityNumber")),
52+
("body".into(), json!("Body")),
53+
("trace_id".into(), json!("TraceId")),
54+
("span_id".into(), json!("SpanId")),
55+
(
56+
"attributes".into(),
57+
json!({
58+
"env": "Environment",
59+
"request.id": "RequestId",
60+
"user.id": "UserId"
61+
}),
62+
),
63+
]),
64+
},
65+
},
66+
auth: AuthConfig::default(),
67+
}
68+
}
69+
70+
fn make_log_record(i: usize) -> LogRecord {
71+
LogRecord {
72+
time_unix_nano: 1_700_000_000_000_000_000 + (i as u64) * 1_000_000,
73+
observed_time_unix_nano: 1_700_000_000_000_000_000 + (i as u64) * 1_000_000,
74+
severity_number: 9, // INFO
75+
severity_text: "INFO".into(),
76+
body: Some(AnyValue {
77+
value: Some(OtelAnyValueEnum::StringValue(format!(
78+
"Log message number {i}"
79+
))),
80+
}),
81+
attributes: vec![
82+
KeyValue {
83+
key: "env".into(),
84+
value: Some(AnyValue {
85+
value: Some(OtelAnyValueEnum::StringValue("production".into())),
86+
}),
87+
},
88+
KeyValue {
89+
key: "request.id".into(),
90+
value: Some(AnyValue {
91+
value: Some(OtelAnyValueEnum::StringValue(format!("req-{i:06}"))),
92+
}),
93+
},
94+
KeyValue {
95+
key: "user.id".into(),
96+
value: Some(AnyValue {
97+
value: Some(OtelAnyValueEnum::StringValue("user-42".into())),
98+
}),
99+
},
100+
],
101+
trace_id: vec![
102+
0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab,
103+
0xcd, 0xef,
104+
],
105+
span_id: vec![0xfe, 0xdc, 0xba, 0x98, 0x76, 0x54, 0x32, 0x10],
106+
flags: 1,
107+
..Default::default()
108+
}
109+
}
110+
111+
fn make_request(
112+
num_resource_logs: usize,
113+
num_scope_logs: usize,
114+
records_per_scope: usize,
115+
) -> Vec<u8> {
116+
let resource_logs: Vec<ResourceLogs> = (0..num_resource_logs)
117+
.map(|_| ResourceLogs {
118+
resource: Some(Resource {
119+
attributes: vec![
120+
KeyValue {
121+
key: "service.name".into(),
122+
value: Some(AnyValue {
123+
value: Some(OtelAnyValueEnum::StringValue("my-service".into())),
124+
}),
125+
},
126+
KeyValue {
127+
key: "service.version".into(),
128+
value: Some(AnyValue {
129+
value: Some(OtelAnyValueEnum::StringValue("1.2.3".into())),
130+
}),
131+
},
132+
KeyValue {
133+
key: "host.name".into(),
134+
value: Some(AnyValue {
135+
value: Some(OtelAnyValueEnum::StringValue("host-01.prod".into())),
136+
}),
137+
},
138+
],
139+
dropped_attributes_count: 0,
140+
entity_refs: vec![],
141+
}),
142+
scope_logs: (0..num_scope_logs)
143+
.map(|_| ScopeLogs {
144+
scope: Some(InstrumentationScope {
145+
name: "my-library".into(),
146+
version: "0.1.0".into(),
147+
attributes: vec![
148+
KeyValue {
149+
key: "scope.name".into(),
150+
value: Some(AnyValue {
151+
value: Some(OtelAnyValueEnum::StringValue("my-library".into())),
152+
}),
153+
},
154+
KeyValue {
155+
key: "scope.version".into(),
156+
value: Some(AnyValue {
157+
value: Some(OtelAnyValueEnum::StringValue("0.1.0".into())),
158+
}),
159+
},
160+
],
161+
dropped_attributes_count: 0,
162+
}),
163+
log_records: (0..records_per_scope).map(make_log_record).collect(),
164+
schema_url: String::new(),
165+
})
166+
.collect(),
167+
schema_url: String::new(),
168+
})
169+
.collect();
170+
171+
let request = ExportLogsServiceRequest { resource_logs };
172+
request.encode_to_vec()
173+
}
174+
175+
fn bench_transform(c: &mut Criterion) {
176+
let mut group = c.benchmark_group("transformer");
177+
178+
let config = create_config();
179+
let transformer = Transformer::new(&config);
180+
181+
// Varying record counts: 1 ResourceLogs, 1 ScopeLogs, N records
182+
for num_records in [10, 100, 1000] {
183+
let bytes = make_request(1, 1, num_records);
184+
let total_records = num_records;
185+
186+
group.throughput(criterion::Throughput::Elements(total_records as u64));
187+
group.bench_with_input(
188+
BenchmarkId::new("1r_1s", total_records),
189+
&bytes,
190+
|b, bytes| {
191+
b.iter(|| {
192+
let view = RawLogsData::new(bytes);
193+
let result = transformer.convert_to_log_analytics(&view);
194+
assert_eq!(result.len(), total_records);
195+
});
196+
},
197+
);
198+
}
199+
200+
// Many scopes: 1 ResourceLogs, 10 ScopeLogs, 100 records each
201+
{
202+
let bytes = make_request(1, 10, 100);
203+
let total_records = 1000;
204+
205+
group.throughput(criterion::Throughput::Elements(total_records as u64));
206+
group.bench_with_input(
207+
BenchmarkId::new("1r_10s", total_records),
208+
&bytes,
209+
|b, bytes| {
210+
b.iter(|| {
211+
let view = RawLogsData::new(bytes);
212+
let result = transformer.convert_to_log_analytics(&view);
213+
assert_eq!(result.len(), total_records);
214+
});
215+
},
216+
);
217+
}
218+
219+
// Many resources: 10 ResourceLogs, 1 ScopeLogs, 100 records each
220+
{
221+
let bytes = make_request(10, 1, 100);
222+
let total_records = 1000;
223+
224+
group.throughput(criterion::Throughput::Elements(total_records as u64));
225+
group.bench_with_input(
226+
BenchmarkId::new("10r_1s", total_records),
227+
&bytes,
228+
|b, bytes| {
229+
b.iter(|| {
230+
let view = RawLogsData::new(bytes);
231+
let result = transformer.convert_to_log_analytics(&view);
232+
assert_eq!(result.len(), total_records);
233+
});
234+
},
235+
);
236+
}
237+
238+
group.finish();
239+
}
240+
241+
criterion_group!(benches, bench_transform);
242+
criterion_main!(benches);

0 commit comments

Comments
 (0)