lodestone_types/
profile.rs

1//! Mapping-owned continuous-profile batches and canonical source projection.
2
3use std::collections::BTreeSet;
4use std::error::Error;
5use std::fmt::{self, Display, Formatter};
6
7use serde::{Deserialize, Serialize};
8
9use crate::{IndexMode, IndexRecordModel};
10
11pub const PROFILE_RECORD_MODEL_FORMAT_VERSION: u16 = 1;
12pub const PROFILE_PRIVACY_POLICY_FORMAT_VERSION: u16 = 1;
13pub const PROFILE_BATCH_FORMAT_VERSION: u16 = 1;
14pub const PROFILE_SOURCE_FORMAT_VERSION: u16 = 1;
15pub const MAX_PROFILE_BATCH_BYTES: usize = 1_048_576;
16pub const MAX_PROFILE_SAMPLE_TYPES: usize = 16;
17pub const MAX_PROFILE_SAMPLES: usize = 10_000;
18pub const MAX_PROFILE_TABLE_ENTRIES: usize = 65_536;
19pub const MAX_PROFILE_STACK_DEPTH: usize = 256;
20pub const MAX_PROFILE_LABELS: usize = 128;
21pub const MAX_PROFILE_PRIVACY_RULES: usize = 256;
22pub const MAX_PROFILE_UNIQUE_STRINGS: usize = 16_384;
23pub const MAX_PROFILE_IDENTIFIER_BYTES: usize = 1_024;
24pub const MAX_PROFILE_STRING_BYTES: usize = 4_096;
25
26#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
27#[serde(rename_all = "snake_case")]
28pub enum ProfilePrivacyActionV1 {
29    Allow,
30    Drop,
31    Redact,
32    Reject,
33}
34
35#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
36#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
37pub enum ProfilePrivacyFieldV1 {
38    ResourceAttribute { key: String },
39    ScopeAttribute { key: String },
40    SampleLabel { key: String },
41    FunctionName,
42    FunctionSystemName,
43    FunctionFilename,
44    MappingFilename,
45}
46
47#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
48#[serde(deny_unknown_fields)]
49pub struct ProfilePrivacyRuleV1 {
50    pub field: ProfilePrivacyFieldV1,
51    pub action: ProfilePrivacyActionV1,
52}
53
54#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
55#[serde(deny_unknown_fields)]
56pub struct ProfilePrivacyPolicyV1 {
57    pub format_version: u16,
58    pub rules: Vec<ProfilePrivacyRuleV1>,
59}
60
61impl ProfilePrivacyPolicyV1 {
62    pub fn validate(&self) -> Result<(), ProfileRecordError> {
63        if self.format_version != PROFILE_PRIVACY_POLICY_FORMAT_VERSION {
64            return Err(ProfileRecordError::UnsupportedPrivacyPolicyVersion(self.format_version));
65        }
66        if self.rules.len() > MAX_PROFILE_PRIVACY_RULES
67            || self.rules.windows(2).any(|pair| pair[0].field >= pair[1].field)
68        {
69            return Err(ProfileRecordError::InvalidPrivacyPolicy);
70        }
71        for rule in &self.rules {
72            match &rule.field {
73                ProfilePrivacyFieldV1::ResourceAttribute { key }
74                | ProfilePrivacyFieldV1::ScopeAttribute { key }
75                | ProfilePrivacyFieldV1::SampleLabel { key } => validate_identifier(key)?,
76                ProfilePrivacyFieldV1::FunctionName
77                | ProfilePrivacyFieldV1::FunctionSystemName
78                | ProfilePrivacyFieldV1::FunctionFilename
79                | ProfilePrivacyFieldV1::MappingFilename => {}
80            }
81        }
82        Ok(())
83    }
84
85    #[must_use]
86    pub fn semantic_digest_v1(&self) -> [u8; 32] {
87        let mut hasher = blake3::Hasher::new();
88        hasher.update(b"lodestone/profile-privacy-policy/v1\0");
89        hasher.update(&self.format_version.to_be_bytes());
90        hasher.update(&(self.rules.len() as u64).to_be_bytes());
91        for rule in &self.rules {
92            match &rule.field {
93                ProfilePrivacyFieldV1::ResourceAttribute { key } => hash_named(&mut hasher, b"resource", key),
94                ProfilePrivacyFieldV1::ScopeAttribute { key } => hash_named(&mut hasher, b"scope", key),
95                ProfilePrivacyFieldV1::SampleLabel { key } => hash_named(&mut hasher, b"sample", key),
96                ProfilePrivacyFieldV1::FunctionName => {
97                    hasher.update(b"function_name\0");
98                }
99                ProfilePrivacyFieldV1::FunctionSystemName => {
100                    hasher.update(b"function_system_name\0");
101                }
102                ProfilePrivacyFieldV1::FunctionFilename => {
103                    hasher.update(b"function_filename\0");
104                }
105                ProfilePrivacyFieldV1::MappingFilename => {
106                    hasher.update(b"mapping_filename\0");
107                }
108            };
109            hasher.update(&[match rule.action {
110                ProfilePrivacyActionV1::Allow => 0,
111                ProfilePrivacyActionV1::Drop => 1,
112                ProfilePrivacyActionV1::Redact => 2,
113                ProfilePrivacyActionV1::Reject => 3,
114            }]);
115        }
116        *hasher.finalize().as_bytes()
117    }
118}
119
120#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
121#[serde(deny_unknown_fields)]
122pub struct ProfileLabelV1 {
123    pub key: String,
124    pub value: String,
125}
126
127#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
128#[serde(deny_unknown_fields)]
129pub struct ProfileSampleTypeV1 {
130    pub sample_type: String,
131    pub unit: String,
132}
133
134#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
135#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
136pub enum ProfileNumberV1 {
137    I64 { value: i64 },
138    F64Bits { value: u64 },
139}
140
141impl ProfileNumberV1 {
142    pub fn validate(self) -> Result<(), ProfileRecordError> {
143        if let Self::F64Bits { value } = self
144            && !f64::from_bits(value).is_finite()
145        {
146            return Err(ProfileRecordError::NonFiniteValue);
147        }
148        Ok(())
149    }
150}
151
152#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
153#[serde(deny_unknown_fields)]
154pub struct ProfileMappingV1 {
155    pub mapping_id: String,
156    pub build_id: String,
157    pub module_id: String,
158    pub filename: Option<String>,
159}
160
161#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
162#[serde(deny_unknown_fields)]
163pub struct ProfileFunctionV1 {
164    pub function_id: String,
165    pub name: Option<String>,
166    pub system_name: Option<String>,
167    pub filename: Option<String>,
168    pub start_line: Option<u64>,
169    pub symbol_observation_id: Option<String>,
170}
171
172#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
173#[serde(deny_unknown_fields)]
174pub struct ProfileLineV1 {
175    pub function_id: String,
176    pub line: u64,
177}
178
179#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
180#[serde(deny_unknown_fields)]
181pub struct ProfileLocationV1 {
182    pub location_id: String,
183    pub mapping_id: Option<String>,
184    pub address: u64,
185    pub lines: Vec<ProfileLineV1>,
186}
187
188#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
189#[serde(deny_unknown_fields)]
190pub struct ProfileStackV1 {
191    pub stack_id: String,
192    pub parent_stack_id: Option<String>,
193    pub location_id: String,
194}
195
196#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
197#[serde(deny_unknown_fields)]
198pub struct ProfileSampleV1 {
199    pub sample_id: String,
200    pub stack_id: String,
201    pub values: Vec<ProfileNumberV1>,
202    pub labels: Vec<ProfileLabelV1>,
203    pub trace_id: Option<String>,
204    pub span_id: Option<String>,
205}
206
207#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
208#[serde(deny_unknown_fields)]
209pub struct ProfileInstrumentationScopeV1 {
210    pub name: String,
211    pub version: Option<String>,
212    pub attributes: Vec<ProfileLabelV1>,
213}
214
215#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
216#[serde(deny_unknown_fields)]
217pub struct ProfileBatchV1 {
218    pub format_version: u16,
219    pub profile_id: String,
220    pub batch_id: String,
221    pub start_unix_nanos: u64,
222    pub end_unix_nanos: u64,
223    pub duration_nanos: u64,
224    pub period_nanos: u64,
225    pub sample_types: Vec<ProfileSampleTypeV1>,
226    pub resource_attributes: Vec<ProfileLabelV1>,
227    pub instrumentation_scope: ProfileInstrumentationScopeV1,
228    pub mappings: Vec<ProfileMappingV1>,
229    pub functions: Vec<ProfileFunctionV1>,
230    pub locations: Vec<ProfileLocationV1>,
231    pub stacks: Vec<ProfileStackV1>,
232    pub samples: Vec<ProfileSampleV1>,
233}
234
235impl ProfileBatchV1 {
236    pub fn validate(&self, model: &IndexRecordModel) -> Result<(), ProfileRecordError> {
237        if self.format_version != PROFILE_BATCH_FORMAT_VERSION {
238            return Err(ProfileRecordError::UnsupportedBatchVersion(self.format_version));
239        }
240        let IndexRecordModel::ProfileV1 { privacy, .. } = model else {
241            return Err(ProfileRecordError::RecordModelMismatch);
242        };
243        privacy.validate()?;
244        validate_identifier(&self.profile_id)?;
245        validate_identifier(&self.batch_id)?;
246        if self.start_unix_nanos >= self.end_unix_nanos
247            || self.duration_nanos != self.end_unix_nanos.saturating_sub(self.start_unix_nanos)
248            || self.period_nanos == 0
249        {
250            return Err(ProfileRecordError::InvalidTimeRange);
251        }
252        if self.sample_types.is_empty()
253            || self.sample_types.len() > MAX_PROFILE_SAMPLE_TYPES
254            || self.sample_types.windows(2).any(|pair| pair[0] >= pair[1])
255        {
256            return Err(ProfileRecordError::InvalidSampleTypes);
257        }
258        for sample_type in &self.sample_types {
259            validate_identifier(&sample_type.sample_type)?;
260            validate_identifier(&sample_type.unit)?;
261        }
262        validate_labels(&self.resource_attributes)?;
263        validate_identifier(&self.instrumentation_scope.name)?;
264        validate_optional_string(self.instrumentation_scope.version.as_deref())?;
265        validate_labels(&self.instrumentation_scope.attributes)?;
266        validate_table_order(&self.mappings, |entry| entry.mapping_id.as_str())?;
267        validate_table_order(&self.functions, |entry| entry.function_id.as_str())?;
268        validate_table_order(&self.locations, |entry| entry.location_id.as_str())?;
269        validate_table_order(&self.stacks, |entry| entry.stack_id.as_str())?;
270        validate_table_order(&self.samples, |entry| entry.sample_id.as_str())?;
271        if self.samples.is_empty() || self.samples.len() > MAX_PROFILE_SAMPLES {
272            return Err(ProfileRecordError::TableBudgetExceeded);
273        }
274        validate_tables(self)?;
275        validate_unique_strings(self)?;
276        let encoded = serde_json::to_vec(self).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
277        if encoded.len() > MAX_PROFILE_BATCH_BYTES {
278            return Err(ProfileRecordError::BatchBudgetExceeded(encoded.len()));
279        }
280        Ok(())
281    }
282
283    pub fn routing_identity_v1(&self, model: &IndexRecordModel) -> Result<[u8; 32], ProfileRecordError> {
284        self.validate(model)?;
285        let mut hasher = blake3::Hasher::new();
286        hasher.update(b"lodestone/profile-routing/v1\0");
287        hasher.update(&(self.profile_id.len() as u64).to_be_bytes());
288        hasher.update(self.profile_id.as_bytes());
289        Ok(*hasher.finalize().as_bytes())
290    }
291}
292
293#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
294#[serde(deny_unknown_fields)]
295pub struct ProfileQueryableColumnsV1 {
296    pub resource_attributes: Vec<ProfileLabelV1>,
297    pub build_ids: Vec<String>,
298    pub module_ids: Vec<String>,
299    pub function_ids: Vec<String>,
300    pub symbol_observation_ids: Vec<String>,
301    pub trace_ids: Vec<String>,
302    pub span_ids: Vec<String>,
303}
304
305#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
306#[serde(deny_unknown_fields)]
307pub struct CanonicalProfileSourceV1 {
308    pub format_version: u16,
309    pub batch: ProfileBatchV1,
310    pub columns: ProfileQueryableColumnsV1,
311}
312
313pub fn canonical_profile_source_v1(
314    mut batch: ProfileBatchV1,
315    model: &IndexRecordModel,
316) -> Result<(Vec<u8>, [u8; 32]), ProfileRecordError> {
317    let IndexRecordModel::ProfileV1 { privacy, .. } = model else {
318        return Err(ProfileRecordError::RecordModelMismatch);
319    };
320    apply_privacy(&mut batch, privacy)?;
321    let routing = batch.routing_identity_v1(model)?;
322    let source = CanonicalProfileSourceV1 {
323        format_version: PROFILE_SOURCE_FORMAT_VERSION,
324        columns: profile_columns(&batch),
325        batch,
326    };
327    let encoded = serde_json::to_vec(&source).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
328    if encoded.len() > MAX_PROFILE_BATCH_BYTES {
329        return Err(ProfileRecordError::BatchBudgetExceeded(encoded.len()));
330    }
331    Ok((encoded, routing))
332}
333
334pub fn decode_canonical_profile_source_v1(
335    bytes: &[u8],
336    model: &IndexRecordModel,
337) -> Result<(CanonicalProfileSourceV1, [u8; 32]), ProfileRecordError> {
338    if bytes.len() > MAX_PROFILE_BATCH_BYTES {
339        return Err(ProfileRecordError::BatchBudgetExceeded(bytes.len()));
340    }
341    let source: CanonicalProfileSourceV1 =
342        serde_json::from_slice(bytes).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
343    if source.format_version != PROFILE_SOURCE_FORMAT_VERSION {
344        return Err(ProfileRecordError::UnsupportedSourceVersion(source.format_version));
345    }
346    let (canonical, routing) = canonical_profile_source_v1(source.batch.clone(), model)?;
347    if canonical != bytes || source.columns != profile_columns(&source.batch) {
348        return Err(ProfileRecordError::NonCanonicalSource);
349    }
350    Ok((source, routing))
351}
352
353pub fn validate_profile_record_model_v1(
354    format_version: u16,
355    privacy: &ProfilePrivacyPolicyV1,
356    mode: IndexMode,
357) -> Result<(), crate::MetricRecordError> {
358    if format_version != PROFILE_RECORD_MODEL_FORMAT_VERSION {
359        return Err(crate::MetricRecordError::UnsupportedRecordModelVersion(format_version));
360    }
361    if mode != IndexMode::AppendLog {
362        return Err(crate::MetricRecordError::ProfileModelRequiresAppendLog);
363    }
364    privacy
365        .validate()
366        .map_err(|_| crate::MetricRecordError::InvalidProfilePrivacyPolicy)
367}
368
369fn validate_tables(batch: &ProfileBatchV1) -> Result<(), ProfileRecordError> {
370    for mapping in &batch.mappings {
371        validate_identifier(&mapping.mapping_id)?;
372        validate_identifier(&mapping.build_id)?;
373        validate_identifier(&mapping.module_id)?;
374        validate_optional_string(mapping.filename.as_deref())?;
375    }
376    for function in &batch.functions {
377        validate_identifier(&function.function_id)?;
378        validate_optional_string(function.name.as_deref())?;
379        validate_optional_string(function.system_name.as_deref())?;
380        validate_optional_string(function.filename.as_deref())?;
381        if let Some(identity) = &function.symbol_observation_id {
382            validate_identifier(identity)?;
383        }
384    }
385    for location in &batch.locations {
386        validate_identifier(&location.location_id)?;
387        if location.lines.len() > MAX_PROFILE_STACK_DEPTH {
388            return Err(ProfileRecordError::StackDepthExceeded);
389        }
390        if let Some(mapping_id) = &location.mapping_id
391            && batch
392                .mappings
393                .binary_search_by(|entry| entry.mapping_id.cmp(mapping_id))
394                .is_err()
395        {
396            return Err(ProfileRecordError::InvalidTableReference);
397        }
398        for line in &location.lines {
399            if batch
400                .functions
401                .binary_search_by(|entry| entry.function_id.cmp(&line.function_id))
402                .is_err()
403            {
404                return Err(ProfileRecordError::InvalidTableReference);
405            }
406        }
407    }
408    for stack in &batch.stacks {
409        validate_identifier(&stack.stack_id)?;
410        if batch
411            .locations
412            .binary_search_by(|entry| entry.location_id.cmp(&stack.location_id))
413            .is_err()
414        {
415            return Err(ProfileRecordError::InvalidTableReference);
416        }
417        if let Some(parent) = &stack.parent_stack_id
418            && batch
419                .stacks
420                .binary_search_by(|entry| entry.stack_id.cmp(parent))
421                .is_err()
422        {
423            return Err(ProfileRecordError::InvalidTableReference);
424        }
425        validate_stack_chain(stack, &batch.stacks)?;
426    }
427    for sample in &batch.samples {
428        validate_identifier(&sample.sample_id)?;
429        if sample.values.len() != batch.sample_types.len()
430            || batch
431                .stacks
432                .binary_search_by(|entry| entry.stack_id.cmp(&sample.stack_id))
433                .is_err()
434        {
435            return Err(ProfileRecordError::InvalidTableReference);
436        }
437        for value in &sample.values {
438            value.validate()?;
439        }
440        validate_labels(&sample.labels)?;
441        if let Some(trace_id) = &sample.trace_id {
442            validate_identifier(trace_id)?;
443        }
444        if let Some(span_id) = &sample.span_id {
445            validate_identifier(span_id)?;
446        }
447    }
448    Ok(())
449}
450
451fn validate_stack_chain(stack: &ProfileStackV1, stacks: &[ProfileStackV1]) -> Result<(), ProfileRecordError> {
452    let mut visited = BTreeSet::new();
453    let mut current = Some(stack.stack_id.as_str());
454    for _ in 0..=MAX_PROFILE_STACK_DEPTH {
455        let Some(stack_id) = current else {
456            return Ok(());
457        };
458        if !visited.insert(stack_id) {
459            return Err(ProfileRecordError::ReferenceCycle);
460        }
461        let position = stacks
462            .binary_search_by(|entry| entry.stack_id.as_str().cmp(stack_id))
463            .map_err(|_| ProfileRecordError::InvalidTableReference)?;
464        current = stacks[position].parent_stack_id.as_deref();
465    }
466    Err(ProfileRecordError::StackDepthExceeded)
467}
468
469fn validate_labels(labels: &[ProfileLabelV1]) -> Result<(), ProfileRecordError> {
470    if labels.len() > MAX_PROFILE_LABELS || labels.windows(2).any(|pair| pair[0].key >= pair[1].key) {
471        return Err(ProfileRecordError::InvalidLabels);
472    }
473    for label in labels {
474        validate_identifier(&label.key)?;
475        validate_string(&label.value)?;
476    }
477    Ok(())
478}
479
480fn validate_table_order<T>(values: &[T], identity: impl Fn(&T) -> &str) -> Result<(), ProfileRecordError> {
481    if values.len() > MAX_PROFILE_TABLE_ENTRIES
482        || values.windows(2).any(|pair| identity(&pair[0]) >= identity(&pair[1]))
483    {
484        return Err(ProfileRecordError::TableBudgetExceeded);
485    }
486    Ok(())
487}
488
489fn validate_unique_strings(batch: &ProfileBatchV1) -> Result<(), ProfileRecordError> {
490    let mut strings = BTreeSet::new();
491    let mut insert = |value: &str| {
492        strings.insert(value.to_owned());
493        if strings.len() > MAX_PROFILE_UNIQUE_STRINGS {
494            Err(ProfileRecordError::UniqueStringBudgetExceeded)
495        } else {
496            Ok(())
497        }
498    };
499    insert(&batch.profile_id)?;
500    insert(&batch.batch_id)?;
501    insert(&batch.instrumentation_scope.name)?;
502    if let Some(version) = &batch.instrumentation_scope.version {
503        insert(version)?;
504    }
505    for sample_type in &batch.sample_types {
506        insert(&sample_type.sample_type)?;
507        insert(&sample_type.unit)?;
508    }
509    for label in batch
510        .resource_attributes
511        .iter()
512        .chain(&batch.instrumentation_scope.attributes)
513        .chain(batch.samples.iter().flat_map(|sample| sample.labels.iter()))
514    {
515        insert(&label.key)?;
516        insert(&label.value)?;
517    }
518    for mapping in &batch.mappings {
519        insert(&mapping.mapping_id)?;
520        insert(&mapping.build_id)?;
521        insert(&mapping.module_id)?;
522        if let Some(filename) = &mapping.filename {
523            insert(filename)?;
524        }
525    }
526    for function in &batch.functions {
527        insert(&function.function_id)?;
528        for value in [
529            function.name.as_deref(),
530            function.system_name.as_deref(),
531            function.filename.as_deref(),
532            function.symbol_observation_id.as_deref(),
533        ]
534        .into_iter()
535        .flatten()
536        {
537            insert(value)?;
538        }
539    }
540    for location in &batch.locations {
541        insert(&location.location_id)?;
542        if let Some(mapping_id) = &location.mapping_id {
543            insert(mapping_id)?;
544        }
545        for line in &location.lines {
546            insert(&line.function_id)?;
547        }
548    }
549    for stack in &batch.stacks {
550        insert(&stack.stack_id)?;
551        insert(&stack.location_id)?;
552        if let Some(parent) = &stack.parent_stack_id {
553            insert(parent)?;
554        }
555    }
556    for sample in &batch.samples {
557        insert(&sample.sample_id)?;
558        insert(&sample.stack_id)?;
559        if let Some(trace_id) = &sample.trace_id {
560            insert(trace_id)?;
561        }
562        if let Some(span_id) = &sample.span_id {
563            insert(span_id)?;
564        }
565    }
566    Ok(())
567}
568
569fn apply_privacy(batch: &mut ProfileBatchV1, policy: &ProfilePrivacyPolicyV1) -> Result<(), ProfileRecordError> {
570    policy.validate()?;
571    for rule in &policy.rules {
572        match &rule.field {
573            ProfilePrivacyFieldV1::ResourceAttribute { key } => {
574                apply_label_rule(&mut batch.resource_attributes, key, rule.action)?;
575            }
576            ProfilePrivacyFieldV1::ScopeAttribute { key } => {
577                apply_label_rule(&mut batch.instrumentation_scope.attributes, key, rule.action)?;
578            }
579            ProfilePrivacyFieldV1::SampleLabel { key } => {
580                for sample in &mut batch.samples {
581                    apply_label_rule(&mut sample.labels, key, rule.action)?;
582                }
583            }
584            ProfilePrivacyFieldV1::FunctionName => {
585                for function in &mut batch.functions {
586                    apply_optional_rule(&mut function.name, rule.action)?;
587                }
588            }
589            ProfilePrivacyFieldV1::FunctionSystemName => {
590                for function in &mut batch.functions {
591                    apply_optional_rule(&mut function.system_name, rule.action)?;
592                }
593            }
594            ProfilePrivacyFieldV1::FunctionFilename => {
595                for function in &mut batch.functions {
596                    apply_optional_rule(&mut function.filename, rule.action)?;
597                }
598            }
599            ProfilePrivacyFieldV1::MappingFilename => {
600                for mapping in &mut batch.mappings {
601                    apply_optional_rule(&mut mapping.filename, rule.action)?;
602                }
603            }
604        }
605    }
606    Ok(())
607}
608
609fn apply_label_rule(
610    labels: &mut Vec<ProfileLabelV1>,
611    key: &str,
612    action: ProfilePrivacyActionV1,
613) -> Result<(), ProfileRecordError> {
614    let Ok(position) = labels.binary_search_by(|label| label.key.as_str().cmp(key)) else {
615        return Ok(());
616    };
617    match action {
618        ProfilePrivacyActionV1::Allow => {}
619        ProfilePrivacyActionV1::Drop => {
620            labels.remove(position);
621        }
622        ProfilePrivacyActionV1::Redact => labels[position].value = "<redacted>".to_owned(),
623        ProfilePrivacyActionV1::Reject => return Err(ProfileRecordError::PrivacyRejected),
624    }
625    Ok(())
626}
627
628fn apply_optional_rule(value: &mut Option<String>, action: ProfilePrivacyActionV1) -> Result<(), ProfileRecordError> {
629    if value.is_none() {
630        return Ok(());
631    }
632    match action {
633        ProfilePrivacyActionV1::Allow => {}
634        ProfilePrivacyActionV1::Drop => *value = None,
635        ProfilePrivacyActionV1::Redact => *value = Some("<redacted>".to_owned()),
636        ProfilePrivacyActionV1::Reject => return Err(ProfileRecordError::PrivacyRejected),
637    }
638    Ok(())
639}
640
641fn profile_columns(batch: &ProfileBatchV1) -> ProfileQueryableColumnsV1 {
642    ProfileQueryableColumnsV1 {
643        resource_attributes: batch.resource_attributes.clone(),
644        build_ids: sorted_unique(batch.mappings.iter().map(|entry| entry.build_id.clone())),
645        module_ids: sorted_unique(batch.mappings.iter().map(|entry| entry.module_id.clone())),
646        function_ids: sorted_unique(batch.functions.iter().map(|entry| entry.function_id.clone())),
647        symbol_observation_ids: sorted_unique(
648            batch
649                .functions
650                .iter()
651                .filter_map(|entry| entry.symbol_observation_id.clone()),
652        ),
653        trace_ids: sorted_unique(batch.samples.iter().filter_map(|entry| entry.trace_id.clone())),
654        span_ids: sorted_unique(batch.samples.iter().filter_map(|entry| entry.span_id.clone())),
655    }
656}
657
658fn sorted_unique(values: impl Iterator<Item = String>) -> Vec<String> {
659    values.collect::<BTreeSet<_>>().into_iter().collect()
660}
661
662fn validate_identifier(value: &str) -> Result<(), ProfileRecordError> {
663    if value.is_empty()
664        || value.len() > MAX_PROFILE_IDENTIFIER_BYTES
665        || value.bytes().any(|byte| byte == 0 || byte <= 0x1f || byte == 0x7f)
666    {
667        return Err(ProfileRecordError::InvalidIdentifier);
668    }
669    Ok(())
670}
671
672fn validate_optional_string(value: Option<&str>) -> Result<(), ProfileRecordError> {
673    value.map_or(Ok(()), validate_string)
674}
675
676fn validate_string(value: &str) -> Result<(), ProfileRecordError> {
677    if value.len() > MAX_PROFILE_STRING_BYTES || value.bytes().any(|byte| byte == 0) {
678        return Err(ProfileRecordError::StringBudgetExceeded);
679    }
680    Ok(())
681}
682
683fn hash_named(hasher: &mut blake3::Hasher, kind: &[u8], value: &str) {
684    hasher.update(kind);
685    hasher.update(&[0]);
686    hasher.update(&(value.len() as u64).to_be_bytes());
687    hasher.update(value.as_bytes());
688}
689
690#[derive(Clone, Debug, Eq, PartialEq)]
691pub enum ProfileRecordError {
692    UnsupportedRecordModelVersion(u16),
693    UnsupportedPrivacyPolicyVersion(u16),
694    UnsupportedBatchVersion(u16),
695    UnsupportedSourceVersion(u16),
696    RecordModelMismatch,
697    InvalidPrivacyPolicy,
698    PrivacyRejected,
699    InvalidIdentifier,
700    StringBudgetExceeded,
701    InvalidTimeRange,
702    InvalidSampleTypes,
703    InvalidLabels,
704    InvalidTableReference,
705    ReferenceCycle,
706    StackDepthExceeded,
707    NonFiniteValue,
708    TableBudgetExceeded,
709    UniqueStringBudgetExceeded,
710    BatchBudgetExceeded(usize),
711    NonCanonicalSource,
712    Codec(String),
713}
714
715impl Display for ProfileRecordError {
716    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
717        write!(formatter, "profile record: {self:?}")
718    }
719}
720
721impl Error for ProfileRecordError {}
722
723#[cfg(test)]
724mod tests {
725    use super::*;
726
727    fn model() -> IndexRecordModel {
728        IndexRecordModel::ProfileV1 {
729            format_version: PROFILE_RECORD_MODEL_FORMAT_VERSION,
730            privacy: ProfilePrivacyPolicyV1 {
731                format_version: PROFILE_PRIVACY_POLICY_FORMAT_VERSION,
732                rules: vec![ProfilePrivacyRuleV1 {
733                    field: ProfilePrivacyFieldV1::ResourceAttribute {
734                        key: "secret".to_owned(),
735                    },
736                    action: ProfilePrivacyActionV1::Redact,
737                }],
738            },
739        }
740    }
741
742    fn batch() -> ProfileBatchV1 {
743        ProfileBatchV1 {
744            format_version: PROFILE_BATCH_FORMAT_VERSION,
745            profile_id: "cpu-main".to_owned(),
746            batch_id: "batch-1".to_owned(),
747            start_unix_nanos: 10,
748            end_unix_nanos: 20,
749            duration_nanos: 10,
750            period_nanos: 1,
751            sample_types: vec![ProfileSampleTypeV1 {
752                sample_type: "cpu".to_owned(),
753                unit: "nanoseconds".to_owned(),
754            }],
755            resource_attributes: vec![ProfileLabelV1 {
756                key: "secret".to_owned(),
757                value: "private".to_owned(),
758            }],
759            instrumentation_scope: ProfileInstrumentationScopeV1 {
760                name: "profiler".to_owned(),
761                version: Some("1".to_owned()),
762                attributes: Vec::new(),
763            },
764            mappings: vec![ProfileMappingV1 {
765                mapping_id: "map-1".to_owned(),
766                build_id: "build-1".to_owned(),
767                module_id: "module-1".to_owned(),
768                filename: Some("service".to_owned()),
769            }],
770            functions: vec![ProfileFunctionV1 {
771                function_id: "fn-1".to_owned(),
772                name: Some("main".to_owned()),
773                system_name: None,
774                filename: None,
775                start_line: Some(1),
776                symbol_observation_id: Some("symbol-1".to_owned()),
777            }],
778            locations: vec![ProfileLocationV1 {
779                location_id: "loc-1".to_owned(),
780                mapping_id: Some("map-1".to_owned()),
781                address: 1,
782                lines: vec![ProfileLineV1 {
783                    function_id: "fn-1".to_owned(),
784                    line: 1,
785                }],
786            }],
787            stacks: vec![ProfileStackV1 {
788                stack_id: "stack-1".to_owned(),
789                parent_stack_id: None,
790                location_id: "loc-1".to_owned(),
791            }],
792            samples: vec![ProfileSampleV1 {
793                sample_id: "sample-1".to_owned(),
794                stack_id: "stack-1".to_owned(),
795                values: vec![ProfileNumberV1::I64 { value: 7 }],
796                labels: Vec::new(),
797                trace_id: Some("trace-1".to_owned()),
798                span_id: Some("span-1".to_owned()),
799            }],
800        }
801    }
802
803    #[test]
804    fn privacy_is_applied_before_canonical_source_and_columns() -> Result<(), Box<dyn Error>> {
805        let (bytes, routing) = canonical_profile_source_v1(batch(), &model())?;
806        let (decoded, decoded_routing) = decode_canonical_profile_source_v1(&bytes, &model())?;
807        assert_eq!(decoded.batch.resource_attributes[0].value, "<redacted>");
808        assert_eq!(decoded.columns.build_ids, vec!["build-1"]);
809        assert_eq!(decoded.columns.trace_ids, vec!["trace-1"]);
810        assert_eq!(routing, decoded_routing);
811        let mut next_batch = batch();
812        next_batch.batch_id = "batch-2".to_owned();
813        assert_eq!(canonical_profile_source_v1(next_batch, &model())?.1, routing);
814        let mut next_profile = batch();
815        next_profile.profile_id = "cpu-secondary".to_owned();
816        assert_ne!(canonical_profile_source_v1(next_profile, &model())?.1, routing);
817        Ok(())
818    }
819
820    #[test]
821    fn malformed_references_cycles_and_nonfinite_values_refuse() {
822        let mut missing = batch();
823        missing.samples[0].stack_id = "missing".to_owned();
824        assert_eq!(
825            missing.validate(&model()),
826            Err(ProfileRecordError::InvalidTableReference)
827        );
828
829        let mut cyclic = batch();
830        cyclic.stacks[0].parent_stack_id = Some("stack-1".to_owned());
831        assert_eq!(cyclic.validate(&model()), Err(ProfileRecordError::ReferenceCycle));
832
833        let mut nonfinite = batch();
834        nonfinite.samples[0].values[0] = ProfileNumberV1::F64Bits {
835            value: f64::NAN.to_bits(),
836        };
837        assert_eq!(nonfinite.validate(&model()), Err(ProfileRecordError::NonFiniteValue));
838    }
839
840    #[test]
841    fn reject_privacy_rule_prevents_canonicalization() {
842        let mut rejecting = model();
843        if let IndexRecordModel::ProfileV1 { privacy, .. } = &mut rejecting {
844            privacy.rules[0].action = ProfilePrivacyActionV1::Reject;
845        }
846        assert_eq!(
847            canonical_profile_source_v1(batch(), &rejecting).map(|_| ()),
848            Err(ProfileRecordError::PrivacyRejected)
849        );
850
851        let mut ambiguous = model();
852        if let IndexRecordModel::ProfileV1 { privacy, .. } = &mut ambiguous {
853            privacy.rules.push(ProfilePrivacyRuleV1 {
854                field: ProfilePrivacyFieldV1::ResourceAttribute {
855                    key: "secret".to_owned(),
856                },
857                action: ProfilePrivacyActionV1::Reject,
858            });
859        }
860        assert_eq!(
861            ambiguous.validate(IndexMode::AppendLog),
862            Err(crate::MetricRecordError::InvalidProfilePrivacyPolicy)
863        );
864    }
865}
lodestone_types/profile.rs

lodestone_types/
profile.rs
