1use std::fmt;
5
6use tracing::field::{Field, Visit};
7
8use crate::{
9 record::{MAX_DIMENSIONS, MAX_EXTRAS},
10 spec::{DimSource, SpanSpec},
11};
12
13#[derive(Clone, Debug)]
14pub struct SpecFields {
15 spec: &'static SpanSpec,
16 dims: [String; MAX_DIMENSIONS],
17 extras: [u64; MAX_EXTRAS],
18 duration_override: Option<u64>,
19}
20
21impl SpecFields {
22 pub fn new(spec: &'static SpanSpec) -> Self {
23 Self {
24 spec,
25 dims: Default::default(),
26 extras: [0; MAX_EXTRAS],
27 duration_override: None,
28 }
29 }
30
31 pub fn dims(&self) -> &[String; MAX_DIMENSIONS] {
32 &self.dims
33 }
34
35 pub fn extras(&self) -> &[u64; MAX_EXTRAS] {
36 &self.extras
37 }
38
39 pub fn duration_override(&self) -> Option<u64> {
40 self.duration_override
41 }
42
43 fn set_text_dim(&mut self, name: &str, value: &str) {
44 for (slot, source) in self.spec.dims.iter().enumerate() {
45 if matches!(source, DimSource::Text(field) if *field == name) {
46 self.dims[slot].replace_range(.., value);
47 }
48 }
49 }
50}
51
52impl Visit for SpecFields {
53 fn record_u64(&mut self, field: &Field, value: u64) {
54 let name = field.name();
55 if self.spec.duration_override == Some(name) {
56 self.duration_override = Some(value);
57 }
58 for (slot, source) in self.spec.dims.iter().enumerate() {
59 if let DimSource::Number {
60 field: dim_field,
61 prefix,
62 } = source && *dim_field == name
63 {
64 self.dims[slot].clear();
65 self.dims[slot].push_str(prefix);
66 self.dims[slot].push_str(&value.to_string());
67 }
68 }
69 for (slot, extra) in self.spec.extras.iter().enumerate() {
70 if *extra == name {
71 self.extras[slot] = value;
72 }
73 }
74 }
75
76 fn record_i64(&mut self, field: &Field, value: i64) {
77 if value >= 0 {
78 self.record_u64(field, value as u64);
79 }
80 }
81
82 fn record_str(&mut self, field: &Field, value: &str) {
83 self.set_text_dim(field.name(), value);
84 }
85
86 fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
87 let rendered = format!("{:?}", value);
88 self.set_text_dim(field.name(), rendered.trim_matches('"'));
89 }
90}
91
92#[cfg(test)]
93mod tests {
94 use std::sync::Arc;
95
96 use reifydb_runtime::sync::mutex::Mutex;
97 use tracing::{
98 Subscriber, debug_span,
99 span::{Attributes, Id},
100 subscriber::with_default,
101 };
102 use tracing_subscriber::{
103 Layer, Registry,
104 layer::{Context, SubscriberExt},
105 registry::LookupSpan,
106 };
107
108 use super::*;
109 use crate::spec::spec_for;
110
111 struct CaptureLayer {
112 captured: Arc<Mutex<Option<SpecFields>>>,
113 }
114
115 impl<S> Layer<S> for CaptureLayer
116 where
117 S: Subscriber + for<'a> LookupSpan<'a>,
118 {
119 fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
120 let Some(name) = ctx.span(id).map(|s| s.name()) else {
121 return;
122 };
123 let Some(spec) = spec_for(name) else {
124 return;
125 };
126 let mut v = SpecFields::new(spec);
127 attrs.record(&mut v);
128 *self.captured.lock() = Some(v);
129 }
130 }
131
132 fn capture(build: impl FnOnce()) -> SpecFields {
133 let captured = Arc::new(Mutex::new(None));
134 let layer = CaptureLayer {
135 captured: captured.clone(),
136 };
137 let subscriber = Registry::default().with(layer);
138 with_default(subscriber, build);
139 let taken = captured.lock().clone();
140 taken.expect("a span matching a spec must be captured")
141 }
142
143 #[test]
144 fn apply_fields_land_in_the_slots_its_spec_declares() {
145 let captured = capture(|| {
148 let _span = debug_span!(
149 "flow::engine::apply",
150 node_type = "map",
151 operator_id = 79u64,
152 input_rows = 10u64,
153 output_rows = 7u64,
154 apply_time_us = 250u64,
155 lock_wait_us = 5u64,
156 );
157 });
158 assert_eq!(captured.dims()[0], "map");
159 assert_eq!(captured.dims()[1], "op79", "operator_id must label the second dimension");
160 assert_eq!(captured.extras(), &[10, 7, 5, 0]);
161 assert_eq!(captured.duration_override(), Some(250));
162 }
163
164 #[test]
165 fn a_span_without_a_duration_override_field_reports_none() {
166 let captured = capture(|| {
169 let _span = debug_span!(
170 "flow::state::range_limited",
171 site = "timer::hydrate_probe",
172 operator_id = 7u64,
173 rows_fetched = 12u64,
174 rows_tombstoned = 4u64,
175 );
176 });
177 assert_eq!(captured.duration_override(), None);
178 assert_eq!(captured.dims()[0], "timer::hydrate_probe");
179 assert_eq!(captured.dims()[1], "op7");
180 assert_eq!(captured.extras()[0], 12);
181 assert_eq!(captured.extras()[1], 4);
182 }
183
184 #[test]
185 fn a_debug_formatted_dimension_loses_its_quotes() {
186 let captured = capture(|| {
190 let _span =
191 debug_span!("flow::state::range_limited", site = ?"reclaim::range", operator_id = 1u64);
192 });
193 assert_eq!(captured.dims()[0], "reclaim::range");
194 }
195
196 #[test]
197 fn a_field_the_spec_does_not_declare_is_ignored() {
198 let captured = capture(|| {
201 let _span = debug_span!(
202 "flow::state::range_limited",
203 site = "reclaim::range",
204 operator_id = 1u64,
205 rows_fetched = 5u64,
206 rows_tombstoned = 0u64,
207 num_parents = 9u64,
208 );
209 });
210 assert_eq!(captured.extras()[2], 0, "an undeclared field must not reach a slot");
211 assert_eq!(captured.extras()[3], 0);
212 }
213}