use std::collections::BTreeSet;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use serde::{Deserialize, Serialize};
use crate::{IndexMode, IndexRecordModel};
pub const PROFILE_RECORD_MODEL_FORMAT_VERSION: u16 = 1;
pub const PROFILE_PRIVACY_POLICY_FORMAT_VERSION: u16 = 1;
pub const PROFILE_BATCH_FORMAT_VERSION: u16 = 1;
pub const PROFILE_SOURCE_FORMAT_VERSION: u16 = 1;
pub const MAX_PROFILE_BATCH_BYTES: usize = 1_048_576;
pub const MAX_PROFILE_SAMPLE_TYPES: usize = 16;
pub const MAX_PROFILE_SAMPLES: usize = 10_000;
pub const MAX_PROFILE_TABLE_ENTRIES: usize = 65_536;
pub const MAX_PROFILE_STACK_DEPTH: usize = 256;
pub const MAX_PROFILE_LABELS: usize = 128;
pub const MAX_PROFILE_PRIVACY_RULES: usize = 256;
pub const MAX_PROFILE_UNIQUE_STRINGS: usize = 16_384;
pub const MAX_PROFILE_IDENTIFIER_BYTES: usize = 1_024;
pub const MAX_PROFILE_STRING_BYTES: usize = 4_096;
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ProfilePrivacyActionV1 {
Allow,
Drop,
Redact,
Reject,
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum ProfilePrivacyFieldV1 {
ResourceAttribute { key: String },
ScopeAttribute { key: String },
SampleLabel { key: String },
FunctionName,
FunctionSystemName,
FunctionFilename,
MappingFilename,
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfilePrivacyRuleV1 {
pub field: ProfilePrivacyFieldV1,
pub action: ProfilePrivacyActionV1,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfilePrivacyPolicyV1 {
pub format_version: u16,
pub rules: Vec<ProfilePrivacyRuleV1>,
}
impl ProfilePrivacyPolicyV1 {
pub fn validate(&self) -> Result<(), ProfileRecordError> {
if self.format_version != PROFILE_PRIVACY_POLICY_FORMAT_VERSION {
return Err(ProfileRecordError::UnsupportedPrivacyPolicyVersion(self.format_version));
}
if self.rules.len() > MAX_PROFILE_PRIVACY_RULES
|| self.rules.windows(2).any(|pair| pair[0].field >= pair[1].field)
{
return Err(ProfileRecordError::InvalidPrivacyPolicy);
}
for rule in &self.rules {
match &rule.field {
ProfilePrivacyFieldV1::ResourceAttribute { key }
| ProfilePrivacyFieldV1::ScopeAttribute { key }
| ProfilePrivacyFieldV1::SampleLabel { key } => validate_identifier(key)?,
ProfilePrivacyFieldV1::FunctionName
| ProfilePrivacyFieldV1::FunctionSystemName
| ProfilePrivacyFieldV1::FunctionFilename
| ProfilePrivacyFieldV1::MappingFilename => {}
}
}
Ok(())
}
#[must_use]
pub fn semantic_digest_v1(&self) -> [u8; 32] {
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/profile-privacy-policy/v1\0");
hasher.update(&self.format_version.to_be_bytes());
hasher.update(&(self.rules.len() as u64).to_be_bytes());
for rule in &self.rules {
match &rule.field {
ProfilePrivacyFieldV1::ResourceAttribute { key } => hash_named(&mut hasher, b"resource", key),
ProfilePrivacyFieldV1::ScopeAttribute { key } => hash_named(&mut hasher, b"scope", key),
ProfilePrivacyFieldV1::SampleLabel { key } => hash_named(&mut hasher, b"sample", key),
ProfilePrivacyFieldV1::FunctionName => {
hasher.update(b"function_name\0");
}
ProfilePrivacyFieldV1::FunctionSystemName => {
hasher.update(b"function_system_name\0");
}
ProfilePrivacyFieldV1::FunctionFilename => {
hasher.update(b"function_filename\0");
}
ProfilePrivacyFieldV1::MappingFilename => {
hasher.update(b"mapping_filename\0");
}
};
hasher.update(&[match rule.action {
ProfilePrivacyActionV1::Allow => 0,
ProfilePrivacyActionV1::Drop => 1,
ProfilePrivacyActionV1::Redact => 2,
ProfilePrivacyActionV1::Reject => 3,
}]);
}
*hasher.finalize().as_bytes()
}
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileLabelV1 {
pub key: String,
pub value: String,
}
#[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileSampleTypeV1 {
pub sample_type: String,
pub unit: String,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum ProfileNumberV1 {
I64 { value: i64 },
F64Bits { value: u64 },
}
impl ProfileNumberV1 {
pub fn validate(self) -> Result<(), ProfileRecordError> {
if let Self::F64Bits { value } = self
&& !f64::from_bits(value).is_finite()
{
return Err(ProfileRecordError::NonFiniteValue);
}
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileMappingV1 {
pub mapping_id: String,
pub build_id: String,
pub module_id: String,
pub filename: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileFunctionV1 {
pub function_id: String,
pub name: Option<String>,
pub system_name: Option<String>,
pub filename: Option<String>,
pub start_line: Option<u64>,
pub symbol_observation_id: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileLineV1 {
pub function_id: String,
pub line: u64,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileLocationV1 {
pub location_id: String,
pub mapping_id: Option<String>,
pub address: u64,
pub lines: Vec<ProfileLineV1>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileStackV1 {
pub stack_id: String,
pub parent_stack_id: Option<String>,
pub location_id: String,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileSampleV1 {
pub sample_id: String,
pub stack_id: String,
pub values: Vec<ProfileNumberV1>,
pub labels: Vec<ProfileLabelV1>,
pub trace_id: Option<String>,
pub span_id: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileInstrumentationScopeV1 {
pub name: String,
pub version: Option<String>,
pub attributes: Vec<ProfileLabelV1>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileBatchV1 {
pub format_version: u16,
pub profile_id: String,
pub batch_id: String,
pub start_unix_nanos: u64,
pub end_unix_nanos: u64,
pub duration_nanos: u64,
pub period_nanos: u64,
pub sample_types: Vec<ProfileSampleTypeV1>,
pub resource_attributes: Vec<ProfileLabelV1>,
pub instrumentation_scope: ProfileInstrumentationScopeV1,
pub mappings: Vec<ProfileMappingV1>,
pub functions: Vec<ProfileFunctionV1>,
pub locations: Vec<ProfileLocationV1>,
pub stacks: Vec<ProfileStackV1>,
pub samples: Vec<ProfileSampleV1>,
}
impl ProfileBatchV1 {
pub fn validate(&self, model: &IndexRecordModel) -> Result<(), ProfileRecordError> {
if self.format_version != PROFILE_BATCH_FORMAT_VERSION {
return Err(ProfileRecordError::UnsupportedBatchVersion(self.format_version));
}
let IndexRecordModel::ProfileV1 { privacy, .. } = model else {
return Err(ProfileRecordError::RecordModelMismatch);
};
privacy.validate()?;
validate_identifier(&self.profile_id)?;
validate_identifier(&self.batch_id)?;
if self.start_unix_nanos >= self.end_unix_nanos
|| self.duration_nanos != self.end_unix_nanos.saturating_sub(self.start_unix_nanos)
|| self.period_nanos == 0
{
return Err(ProfileRecordError::InvalidTimeRange);
}
if self.sample_types.is_empty()
|| self.sample_types.len() > MAX_PROFILE_SAMPLE_TYPES
|| self.sample_types.windows(2).any(|pair| pair[0] >= pair[1])
{
return Err(ProfileRecordError::InvalidSampleTypes);
}
for sample_type in &self.sample_types {
validate_identifier(&sample_type.sample_type)?;
validate_identifier(&sample_type.unit)?;
}
validate_labels(&self.resource_attributes)?;
validate_identifier(&self.instrumentation_scope.name)?;
validate_optional_string(self.instrumentation_scope.version.as_deref())?;
validate_labels(&self.instrumentation_scope.attributes)?;
validate_table_order(&self.mappings, |entry| entry.mapping_id.as_str())?;
validate_table_order(&self.functions, |entry| entry.function_id.as_str())?;
validate_table_order(&self.locations, |entry| entry.location_id.as_str())?;
validate_table_order(&self.stacks, |entry| entry.stack_id.as_str())?;
validate_table_order(&self.samples, |entry| entry.sample_id.as_str())?;
if self.samples.is_empty() || self.samples.len() > MAX_PROFILE_SAMPLES {
return Err(ProfileRecordError::TableBudgetExceeded);
}
validate_tables(self)?;
validate_unique_strings(self)?;
let encoded = serde_json::to_vec(self).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
if encoded.len() > MAX_PROFILE_BATCH_BYTES {
return Err(ProfileRecordError::BatchBudgetExceeded(encoded.len()));
}
Ok(())
}
pub fn routing_identity_v1(&self, model: &IndexRecordModel) -> Result<[u8; 32], ProfileRecordError> {
self.validate(model)?;
let mut hasher = blake3::Hasher::new();
hasher.update(b"lodestone/profile-routing/v1\0");
hasher.update(&(self.profile_id.len() as u64).to_be_bytes());
hasher.update(self.profile_id.as_bytes());
Ok(*hasher.finalize().as_bytes())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileQueryableColumnsV1 {
pub resource_attributes: Vec<ProfileLabelV1>,
pub build_ids: Vec<String>,
pub module_ids: Vec<String>,
pub function_ids: Vec<String>,
pub symbol_observation_ids: Vec<String>,
pub trace_ids: Vec<String>,
pub span_ids: Vec<String>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CanonicalProfileSourceV1 {
pub format_version: u16,
pub batch: ProfileBatchV1,
pub columns: ProfileQueryableColumnsV1,
}
pub fn canonical_profile_source_v1(
mut batch: ProfileBatchV1,
model: &IndexRecordModel,
) -> Result<(Vec<u8>, [u8; 32]), ProfileRecordError> {
let IndexRecordModel::ProfileV1 { privacy, .. } = model else {
return Err(ProfileRecordError::RecordModelMismatch);
};
apply_privacy(&mut batch, privacy)?;
let routing = batch.routing_identity_v1(model)?;
let source = CanonicalProfileSourceV1 {
format_version: PROFILE_SOURCE_FORMAT_VERSION,
columns: profile_columns(&batch),
batch,
};
let encoded = serde_json::to_vec(&source).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
if encoded.len() > MAX_PROFILE_BATCH_BYTES {
return Err(ProfileRecordError::BatchBudgetExceeded(encoded.len()));
}
Ok((encoded, routing))
}
pub fn decode_canonical_profile_source_v1(
bytes: &[u8],
model: &IndexRecordModel,
) -> Result<(CanonicalProfileSourceV1, [u8; 32]), ProfileRecordError> {
if bytes.len() > MAX_PROFILE_BATCH_BYTES {
return Err(ProfileRecordError::BatchBudgetExceeded(bytes.len()));
}
let source: CanonicalProfileSourceV1 =
serde_json::from_slice(bytes).map_err(|error| ProfileRecordError::Codec(error.to_string()))?;
if source.format_version != PROFILE_SOURCE_FORMAT_VERSION {
return Err(ProfileRecordError::UnsupportedSourceVersion(source.format_version));
}
let (canonical, routing) = canonical_profile_source_v1(source.batch.clone(), model)?;
if canonical != bytes || source.columns != profile_columns(&source.batch) {
return Err(ProfileRecordError::NonCanonicalSource);
}
Ok((source, routing))
}
pub fn validate_profile_record_model_v1(
format_version: u16,
privacy: &ProfilePrivacyPolicyV1,
mode: IndexMode,
) -> Result<(), crate::MetricRecordError> {
if format_version != PROFILE_RECORD_MODEL_FORMAT_VERSION {
return Err(crate::MetricRecordError::UnsupportedRecordModelVersion(format_version));
}
if mode != IndexMode::AppendLog {
return Err(crate::MetricRecordError::ProfileModelRequiresAppendLog);
}
privacy
.validate()
.map_err(|_| crate::MetricRecordError::InvalidProfilePrivacyPolicy)
}
fn validate_tables(batch: &ProfileBatchV1) -> Result<(), ProfileRecordError> {
for mapping in &batch.mappings {
validate_identifier(&mapping.mapping_id)?;
validate_identifier(&mapping.build_id)?;
validate_identifier(&mapping.module_id)?;
validate_optional_string(mapping.filename.as_deref())?;
}
for function in &batch.functions {
validate_identifier(&function.function_id)?;
validate_optional_string(function.name.as_deref())?;
validate_optional_string(function.system_name.as_deref())?;
validate_optional_string(function.filename.as_deref())?;
if let Some(identity) = &function.symbol_observation_id {
validate_identifier(identity)?;
}
}
for location in &batch.locations {
validate_identifier(&location.location_id)?;
if location.lines.len() > MAX_PROFILE_STACK_DEPTH {
return Err(ProfileRecordError::StackDepthExceeded);
}
if let Some(mapping_id) = &location.mapping_id
&& batch
.mappings
.binary_search_by(|entry| entry.mapping_id.cmp(mapping_id))
.is_err()
{
return Err(ProfileRecordError::InvalidTableReference);
}
for line in &location.lines {
if batch
.functions
.binary_search_by(|entry| entry.function_id.cmp(&line.function_id))
.is_err()
{
return Err(ProfileRecordError::InvalidTableReference);
}
}
}
for stack in &batch.stacks {
validate_identifier(&stack.stack_id)?;
if batch
.locations
.binary_search_by(|entry| entry.location_id.cmp(&stack.location_id))
.is_err()
{
return Err(ProfileRecordError::InvalidTableReference);
}
if let Some(parent) = &stack.parent_stack_id
&& batch
.stacks
.binary_search_by(|entry| entry.stack_id.cmp(parent))
.is_err()
{
return Err(ProfileRecordError::InvalidTableReference);
}
validate_stack_chain(stack, &batch.stacks)?;
}
for sample in &batch.samples {
validate_identifier(&sample.sample_id)?;
if sample.values.len() != batch.sample_types.len()
|| batch
.stacks
.binary_search_by(|entry| entry.stack_id.cmp(&sample.stack_id))
.is_err()
{
return Err(ProfileRecordError::InvalidTableReference);
}
for value in &sample.values {
value.validate()?;
}
validate_labels(&sample.labels)?;
if let Some(trace_id) = &sample.trace_id {
validate_identifier(trace_id)?;
}
if let Some(span_id) = &sample.span_id {
validate_identifier(span_id)?;
}
}
Ok(())
}
fn validate_stack_chain(stack: &ProfileStackV1, stacks: &[ProfileStackV1]) -> Result<(), ProfileRecordError> {
let mut visited = BTreeSet::new();
let mut current = Some(stack.stack_id.as_str());
for _ in 0..=MAX_PROFILE_STACK_DEPTH {
let Some(stack_id) = current else {
return Ok(());
};
if !visited.insert(stack_id) {
return Err(ProfileRecordError::ReferenceCycle);
}
let position = stacks
.binary_search_by(|entry| entry.stack_id.as_str().cmp(stack_id))
.map_err(|_| ProfileRecordError::InvalidTableReference)?;
current = stacks[position].parent_stack_id.as_deref();
}
Err(ProfileRecordError::StackDepthExceeded)
}
fn validate_labels(labels: &[ProfileLabelV1]) -> Result<(), ProfileRecordError> {
if labels.len() > MAX_PROFILE_LABELS || labels.windows(2).any(|pair| pair[0].key >= pair[1].key) {
return Err(ProfileRecordError::InvalidLabels);
}
for label in labels {
validate_identifier(&label.key)?;
validate_string(&label.value)?;
}
Ok(())
}
fn validate_table_order<T>(values: &[T], identity: impl Fn(&T) -> &str) -> Result<(), ProfileRecordError> {
if values.len() > MAX_PROFILE_TABLE_ENTRIES
|| values.windows(2).any(|pair| identity(&pair[0]) >= identity(&pair[1]))
{
return Err(ProfileRecordError::TableBudgetExceeded);
}
Ok(())
}
fn validate_unique_strings(batch: &ProfileBatchV1) -> Result<(), ProfileRecordError> {
let mut strings = BTreeSet::new();
let mut insert = |value: &str| {
strings.insert(value.to_owned());
if strings.len() > MAX_PROFILE_UNIQUE_STRINGS {
Err(ProfileRecordError::UniqueStringBudgetExceeded)
} else {
Ok(())
}
};
insert(&batch.profile_id)?;
insert(&batch.batch_id)?;
insert(&batch.instrumentation_scope.name)?;
if let Some(version) = &batch.instrumentation_scope.version {
insert(version)?;
}
for sample_type in &batch.sample_types {
insert(&sample_type.sample_type)?;
insert(&sample_type.unit)?;
}
for label in batch
.resource_attributes
.iter()
.chain(&batch.instrumentation_scope.attributes)
.chain(batch.samples.iter().flat_map(|sample| sample.labels.iter()))
{
insert(&label.key)?;
insert(&label.value)?;
}
for mapping in &batch.mappings {
insert(&mapping.mapping_id)?;
insert(&mapping.build_id)?;
insert(&mapping.module_id)?;
if let Some(filename) = &mapping.filename {
insert(filename)?;
}
}
for function in &batch.functions {
insert(&function.function_id)?;
for value in [
function.name.as_deref(),
function.system_name.as_deref(),
function.filename.as_deref(),
function.symbol_observation_id.as_deref(),
]
.into_iter()
.flatten()
{
insert(value)?;
}
}
for location in &batch.locations {
insert(&location.location_id)?;
if let Some(mapping_id) = &location.mapping_id {
insert(mapping_id)?;
}
for line in &location.lines {
insert(&line.function_id)?;
}
}
for stack in &batch.stacks {
insert(&stack.stack_id)?;
insert(&stack.location_id)?;
if let Some(parent) = &stack.parent_stack_id {
insert(parent)?;
}
}
for sample in &batch.samples {
insert(&sample.sample_id)?;
insert(&sample.stack_id)?;
if let Some(trace_id) = &sample.trace_id {
insert(trace_id)?;
}
if let Some(span_id) = &sample.span_id {
insert(span_id)?;
}
}
Ok(())
}
fn apply_privacy(batch: &mut ProfileBatchV1, policy: &ProfilePrivacyPolicyV1) -> Result<(), ProfileRecordError> {
policy.validate()?;
for rule in &policy.rules {
match &rule.field {
ProfilePrivacyFieldV1::ResourceAttribute { key } => {
apply_label_rule(&mut batch.resource_attributes, key, rule.action)?;
}
ProfilePrivacyFieldV1::ScopeAttribute { key } => {
apply_label_rule(&mut batch.instrumentation_scope.attributes, key, rule.action)?;
}
ProfilePrivacyFieldV1::SampleLabel { key } => {
for sample in &mut batch.samples {
apply_label_rule(&mut sample.labels, key, rule.action)?;
}
}
ProfilePrivacyFieldV1::FunctionName => {
for function in &mut batch.functions {
apply_optional_rule(&mut function.name, rule.action)?;
}
}
ProfilePrivacyFieldV1::FunctionSystemName => {
for function in &mut batch.functions {
apply_optional_rule(&mut function.system_name, rule.action)?;
}
}
ProfilePrivacyFieldV1::FunctionFilename => {
for function in &mut batch.functions {
apply_optional_rule(&mut function.filename, rule.action)?;
}
}
ProfilePrivacyFieldV1::MappingFilename => {
for mapping in &mut batch.mappings {
apply_optional_rule(&mut mapping.filename, rule.action)?;
}
}
}
}
Ok(())
}
fn apply_label_rule(
labels: &mut Vec<ProfileLabelV1>,
key: &str,
action: ProfilePrivacyActionV1,
) -> Result<(), ProfileRecordError> {
let Ok(position) = labels.binary_search_by(|label| label.key.as_str().cmp(key)) else {
return Ok(());
};
match action {
ProfilePrivacyActionV1::Allow => {}
ProfilePrivacyActionV1::Drop => {
labels.remove(position);
}
ProfilePrivacyActionV1::Redact => labels[position].value = "<redacted>".to_owned(),
ProfilePrivacyActionV1::Reject => return Err(ProfileRecordError::PrivacyRejected),
}
Ok(())
}
fn apply_optional_rule(value: &mut Option<String>, action: ProfilePrivacyActionV1) -> Result<(), ProfileRecordError> {
if value.is_none() {
return Ok(());
}
match action {
ProfilePrivacyActionV1::Allow => {}
ProfilePrivacyActionV1::Drop => *value = None,
ProfilePrivacyActionV1::Redact => *value = Some("<redacted>".to_owned()),
ProfilePrivacyActionV1::Reject => return Err(ProfileRecordError::PrivacyRejected),
}
Ok(())
}
fn profile_columns(batch: &ProfileBatchV1) -> ProfileQueryableColumnsV1 {
ProfileQueryableColumnsV1 {
resource_attributes: batch.resource_attributes.clone(),
build_ids: sorted_unique(batch.mappings.iter().map(|entry| entry.build_id.clone())),
module_ids: sorted_unique(batch.mappings.iter().map(|entry| entry.module_id.clone())),
function_ids: sorted_unique(batch.functions.iter().map(|entry| entry.function_id.clone())),
symbol_observation_ids: sorted_unique(
batch
.functions
.iter()
.filter_map(|entry| entry.symbol_observation_id.clone()),
),
trace_ids: sorted_unique(batch.samples.iter().filter_map(|entry| entry.trace_id.clone())),
span_ids: sorted_unique(batch.samples.iter().filter_map(|entry| entry.span_id.clone())),
}
}
fn sorted_unique(values: impl Iterator<Item = String>) -> Vec<String> {
values.collect::<BTreeSet<_>>().into_iter().collect()
}
fn validate_identifier(value: &str) -> Result<(), ProfileRecordError> {
if value.is_empty()
|| value.len() > MAX_PROFILE_IDENTIFIER_BYTES
|| value.bytes().any(|byte| byte == 0 || byte <= 0x1f || byte == 0x7f)
{
return Err(ProfileRecordError::InvalidIdentifier);
}
Ok(())
}
fn validate_optional_string(value: Option<&str>) -> Result<(), ProfileRecordError> {
value.map_or(Ok(()), validate_string)
}
fn validate_string(value: &str) -> Result<(), ProfileRecordError> {
if value.len() > MAX_PROFILE_STRING_BYTES || value.bytes().any(|byte| byte == 0) {
return Err(ProfileRecordError::StringBudgetExceeded);
}
Ok(())
}
fn hash_named(hasher: &mut blake3::Hasher, kind: &[u8], value: &str) {
hasher.update(kind);
hasher.update(&[0]);
hasher.update(&(value.len() as u64).to_be_bytes());
hasher.update(value.as_bytes());
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ProfileRecordError {
UnsupportedRecordModelVersion(u16),
UnsupportedPrivacyPolicyVersion(u16),
UnsupportedBatchVersion(u16),
UnsupportedSourceVersion(u16),
RecordModelMismatch,
InvalidPrivacyPolicy,
PrivacyRejected,
InvalidIdentifier,
StringBudgetExceeded,
InvalidTimeRange,
InvalidSampleTypes,
InvalidLabels,
InvalidTableReference,
ReferenceCycle,
StackDepthExceeded,
NonFiniteValue,
TableBudgetExceeded,
UniqueStringBudgetExceeded,
BatchBudgetExceeded(usize),
NonCanonicalSource,
Codec(String),
}
impl Display for ProfileRecordError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
write!(formatter, "profile record: {self:?}")
}
}
impl Error for ProfileRecordError {}
#[cfg(test)]
mod tests {
use super::*;
fn model() -> IndexRecordModel {
IndexRecordModel::ProfileV1 {
format_version: PROFILE_RECORD_MODEL_FORMAT_VERSION,
privacy: ProfilePrivacyPolicyV1 {
format_version: PROFILE_PRIVACY_POLICY_FORMAT_VERSION,
rules: vec![ProfilePrivacyRuleV1 {
field: ProfilePrivacyFieldV1::ResourceAttribute {
key: "secret".to_owned(),
},
action: ProfilePrivacyActionV1::Redact,
}],
},
}
}
fn batch() -> ProfileBatchV1 {
ProfileBatchV1 {
format_version: PROFILE_BATCH_FORMAT_VERSION,
profile_id: "cpu-main".to_owned(),
batch_id: "batch-1".to_owned(),
start_unix_nanos: 10,
end_unix_nanos: 20,
duration_nanos: 10,
period_nanos: 1,
sample_types: vec![ProfileSampleTypeV1 {
sample_type: "cpu".to_owned(),
unit: "nanoseconds".to_owned(),
}],
resource_attributes: vec![ProfileLabelV1 {
key: "secret".to_owned(),
value: "private".to_owned(),
}],
instrumentation_scope: ProfileInstrumentationScopeV1 {
name: "profiler".to_owned(),
version: Some("1".to_owned()),
attributes: Vec::new(),
},
mappings: vec![ProfileMappingV1 {
mapping_id: "map-1".to_owned(),
build_id: "build-1".to_owned(),
module_id: "module-1".to_owned(),
filename: Some("service".to_owned()),
}],
functions: vec![ProfileFunctionV1 {
function_id: "fn-1".to_owned(),
name: Some("main".to_owned()),
system_name: None,
filename: None,
start_line: Some(1),
symbol_observation_id: Some("symbol-1".to_owned()),
}],
locations: vec![ProfileLocationV1 {
location_id: "loc-1".to_owned(),
mapping_id: Some("map-1".to_owned()),
address: 1,
lines: vec![ProfileLineV1 {
function_id: "fn-1".to_owned(),
line: 1,
}],
}],
stacks: vec![ProfileStackV1 {
stack_id: "stack-1".to_owned(),
parent_stack_id: None,
location_id: "loc-1".to_owned(),
}],
samples: vec![ProfileSampleV1 {
sample_id: "sample-1".to_owned(),
stack_id: "stack-1".to_owned(),
values: vec![ProfileNumberV1::I64 { value: 7 }],
labels: Vec::new(),
trace_id: Some("trace-1".to_owned()),
span_id: Some("span-1".to_owned()),
}],
}
}
#[test]
fn privacy_is_applied_before_canonical_source_and_columns() -> Result<(), Box<dyn Error>> {
let (bytes, routing) = canonical_profile_source_v1(batch(), &model())?;
let (decoded, decoded_routing) = decode_canonical_profile_source_v1(&bytes, &model())?;
assert_eq!(decoded.batch.resource_attributes[0].value, "<redacted>");
assert_eq!(decoded.columns.build_ids, vec!["build-1"]);
assert_eq!(decoded.columns.trace_ids, vec!["trace-1"]);
assert_eq!(routing, decoded_routing);
let mut next_batch = batch();
next_batch.batch_id = "batch-2".to_owned();
assert_eq!(canonical_profile_source_v1(next_batch, &model())?.1, routing);
let mut next_profile = batch();
next_profile.profile_id = "cpu-secondary".to_owned();
assert_ne!(canonical_profile_source_v1(next_profile, &model())?.1, routing);
Ok(())
}
#[test]
fn malformed_references_cycles_and_nonfinite_values_refuse() {
let mut missing = batch();
missing.samples[0].stack_id = "missing".to_owned();
assert_eq!(
missing.validate(&model()),
Err(ProfileRecordError::InvalidTableReference)
);
let mut cyclic = batch();
cyclic.stacks[0].parent_stack_id = Some("stack-1".to_owned());
assert_eq!(cyclic.validate(&model()), Err(ProfileRecordError::ReferenceCycle));
let mut nonfinite = batch();
nonfinite.samples[0].values[0] = ProfileNumberV1::F64Bits {
value: f64::NAN.to_bits(),
};
assert_eq!(nonfinite.validate(&model()), Err(ProfileRecordError::NonFiniteValue));
}
#[test]
fn reject_privacy_rule_prevents_canonicalization() {
let mut rejecting = model();
if let IndexRecordModel::ProfileV1 { privacy, .. } = &mut rejecting {
privacy.rules[0].action = ProfilePrivacyActionV1::Reject;
}
assert_eq!(
canonical_profile_source_v1(batch(), &rejecting).map(|_| ()),
Err(ProfileRecordError::PrivacyRejected)
);
let mut ambiguous = model();
if let IndexRecordModel::ProfileV1 { privacy, .. } = &mut ambiguous {
privacy.rules.push(ProfilePrivacyRuleV1 {
field: ProfilePrivacyFieldV1::ResourceAttribute {
key: "secret".to_owned(),
},
action: ProfilePrivacyActionV1::Reject,
});
}
assert_eq!(
ambiguous.validate(IndexMode::AppendLog),
Err(crate::MetricRecordError::InvalidProfilePrivacyPolicy)
);
}
}