Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 89 additions & 28 deletions crates/ourios-miner/src/cluster.rs
Original file line number Diff line number Diff line change
Expand Up @@ -507,11 +507,16 @@ impl MinerCluster {
/// `body_retentions_total` metric doc explicitly excludes,
/// not the §6.3 lossy-zone retention the gauge is meant to
/// surface. That path uses [`Self::record_tokenizer_failure`].
fn record_parse_failure(&self, record: &OtlpLogRecord, service: Option<&str>) {
fn record_parse_failure(
&self,
record: &OtlpLogRecord,
service: Option<&str>,
reason: &'static str,
) {
self.parse_failures_total.fetch_add(1, Ordering::Relaxed);
self.body_retentions_total.fetch_add(1, Ordering::Relaxed);
self.metrics
.record_parse_failure(&record.tenant_id, service);
.record_parse_failure(&record.tenant_id, service, reason);
self.metrics.record_body_retention(&record.tenant_id);
}

Expand All @@ -528,7 +533,7 @@ impl MinerCluster {
fn record_tokenizer_failure(&self, record: &OtlpLogRecord, service: Option<&str>) {
self.parse_failures_total.fetch_add(1, Ordering::Relaxed);
self.metrics
.record_parse_failure(&record.tenant_id, service);
.record_parse_failure(&record.tenant_id, service, "tokenizer_failure");
}

/// Build the OTLP-envelope half of a `MinedRecord` from the
Expand Down Expand Up @@ -1230,7 +1235,7 @@ impl MinerCluster {
rec.body = Some(raw.to_string());
rec.lossy_flag = true;
self.emit_record(rec, service);
self.record_parse_failure(record, service);
self.record_parse_failure(record, service, "empty_line");
return NO_TEMPLATE;
}

Expand All @@ -1242,7 +1247,14 @@ impl MinerCluster {
// Vec<u16>`), whose violation would otherwise be the
// silent-merge bug `[CLAUDE.md §3.1]` exists to prevent.
if masked_strs.len() > usize::from(effective_config.max_line_tokens) {
return self.emit_string_parse_failure(record, service, raw, separators, params);
return self.emit_string_parse_failure(
record,
service,
raw,
separators,
params,
"line_too_long",
);
}

// Phase 1 — read-only candidate selection. RFC §6.2 step
Expand All @@ -1267,8 +1279,14 @@ impl MinerCluster {
// parse-failure path — body retained, counted,
// never force-merged.
if self.at_template_ceiling(&record.tenant_id, effective_config.max_templates) {
return self
.emit_string_parse_failure(record, service, raw, separators, params);
return self.emit_string_parse_failure(
record,
service,
raw,
separators,
params,
"template_ceiling",
);
}
let new_id = self.create_new_leaf(
record,
Expand Down Expand Up @@ -1328,7 +1346,12 @@ impl MinerCluster {
.at_template_ceiling(&record.tenant_id, effective_config.max_templates)
{
return self.emit_string_parse_failure(
record, service, raw, separators, params,
record,
service,
raw,
separators,
params,
"template_ceiling",
);
}
self.body_retentions_total.fetch_add(1, Ordering::Relaxed);
Expand Down Expand Up @@ -1359,9 +1382,14 @@ impl MinerCluster {
}
// Parse failure: no template allocated.
// Both counters bump via the shared helper.
ConfidenceZone::ParseFailure => {
self.emit_string_parse_failure(record, service, raw, separators, params)
}
ConfidenceZone::ParseFailure => self.emit_string_parse_failure(
record,
service,
raw,
separators,
params,
"below_floor",
),
}
}
}
Expand All @@ -1379,6 +1407,7 @@ impl MinerCluster {
raw: &str,
separators: Vec<String>,
params: Vec<Param>,
reason: &'static str,
) -> u64 {
let mut rec = Self::record_envelope(record, BodyKind::String);
rec.separators = separators;
Expand All @@ -1389,10 +1418,41 @@ impl MinerCluster {
// (body is already retained for the parse-failure reason).
self.apply_overflow_retention(record, service, &mut rec, raw);
self.emit_record(rec, service);
self.record_parse_failure(record, service);
self.record_parse_failure(record, service, reason);
NO_TEMPLATE
}

/// Emit the §6.4 degenerate-widening rejection audit event —
/// the record of *why* the attach refused to widen (RFC 0017
/// §3.1 keeps the audit stream the template history of record).
#[allow(clippy::too_many_arguments)] // mirrors the audit payload's fields 1:1
fn emit_rejected_degenerate_audit(
&mut self,
record: &OtlpLogRecord,
raw: &str,
template_id: u64,
version: u32,
current_template: String,
would_be_template: String,
would_be_positions: Vec<u16>,
) {
self.audit_sink.emit(AuditEvent {
tenant_id: record.tenant_id.clone(),
timestamp: self.clock.now(),
payload: AuditPayload::Template {
template_id,
triggering_line_hash: hash_triggering_line(raw.as_bytes()),
triggering_line_sample: Some(sample_first_256_bytes(raw)),
change: TemplateChange::RejectedDegenerate {
version,
current_template,
would_be_template,
would_be_positions,
},
},
});
}

/// RFC 0023 §3.1 bound 2 — whether the tenant's Drain-tree
/// leaf count has reached the configured ceiling. An unseen
/// tenant is trivially below it.
Expand Down Expand Up @@ -1671,26 +1731,27 @@ impl MinerCluster {
would_be_template,
would_be_positions,
} => {
self.audit_sink.emit(AuditEvent {
tenant_id: record.tenant_id.clone(),
timestamp: self.clock.now(),
payload: AuditPayload::Template {
template_id,
triggering_line_hash: hash_triggering_line(raw.as_bytes()),
triggering_line_sample: Some(sample_first_256_bytes(raw)),
change: TemplateChange::RejectedDegenerate {
version,
current_template,
would_be_template,
would_be_positions,
},
},
});
self.emit_rejected_degenerate_audit(
record,
raw,
template_id,
version,
current_template,
would_be_template,
would_be_positions,
);
// §6.4 treats degenerate widening as a parse
// failure that retains body (the line-ordered
// params fallback is fine — reconstruct ignores
// `params` on the lossy path).
self.emit_string_parse_failure(record, service, raw, separators, params)
self.emit_string_parse_failure(
record,
service,
raw,
separators,
params,
"degenerate_widening",
)
}
AttachPlan::Mutated {
template_id,
Expand Down
28 changes: 25 additions & 3 deletions crates/ourios-miner/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -561,9 +561,31 @@ impl MinerMetrics {
}

/// Record one parse-failure line (§6.8 `ourios.miner.parse_failures`).
pub(crate) fn record_parse_failure(&self, tenant: &TenantId, service: Option<&str>) {
self.parse_failures_total
.add(1, &service_attrs(tenant, service));
pub(crate) fn record_parse_failure(
&self,
tenant: &TenantId,
service: Option<&str>,
reason: &'static str,
) {
// RFC 0023 §3.4 — the cause dimension rides the one counter
// as an attribute (the OTel error.type convention), never
// per-cause counters. Values are the
// `ourios.miner.parse_failure.reason` enum members. Built
// with exact capacity — this is a per-line hot path on
// saturated tenants.
let mut attrs = Vec::with_capacity(2 + usize::from(service.is_some()));
attrs.push(KeyValue::new(
semconv::OURIOS_TENANT,
tenant.as_str().to_owned(),
));
if let Some(name) = service {
attrs.push(KeyValue::new(semconv::OURIOS_SERVICE, name.to_owned()));
}
attrs.push(KeyValue::new(
semconv::OURIOS_MINER_PARSE_FAILURE_REASON,
reason,
));
self.parse_failures_total.add(1, &attrs);
}

/// Record one body-retention event for the
Expand Down
85 changes: 77 additions & 8 deletions crates/ourios-miner/tests/rfc0023_bounded_memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -259,13 +259,82 @@ fn rfc0023_5_default_bounds_are_invisible_on_healthy_corpora() {

/// Scenario RFC0023.6 — saturation is observable.
/// See `docs/rfcs/0023-bounded-template-memory.md` §5.
#[test]
#[ignore = "RFC0023.6 stub — implemented in the telemetry green slice"]
fn rfc0023_6_ceiling_saturation_is_observable() {
todo!(
"RFC0023.6 — a ceiling-saturated tenant shows \
ourios.miner.parse_failures increments with \
reason = template_ceiling and ourios.miner.template.count at the \
ceiling value"
///
/// A ceiling-saturated tenant is diagnosable from telemetry alone:
/// `ourios.miner.parse_failures` carries
/// `ourios.miner.parse_failure.reason = template_ceiling` increments
/// and `ourios.miner.template.count` reads the ceiling value.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn rfc0023_6_ceiling_saturation_is_observable() {
use opentelemetry_sdk::metrics::data::{
AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics,
};

fn has_attr<'a>(
mut attrs: impl Iterator<Item = &'a opentelemetry::KeyValue>,
key: &str,
value: &str,
) -> bool {
attrs.any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
}

let (guard, exporter) = ourios_telemetry::init_in_memory("ourios-test");
let ceiling = 1u32;
let config = MinerConfig::default()
.with_max_templates(ceiling)
.expect("non-zero ceiling");
let mut cluster = MinerCluster::new(config);
let tenant = TenantId::new("saturated");

// One mint fills the ceiling; two structurally distinct lines
// divert with reason = template_ceiling.
cluster.ingest(&record(&tenant, "alpha started"));
cluster.ingest(&record(&tenant, "beta accepted request now"));
cluster.ingest(&record(&tenant, "gamma rejected request from peer five"));
guard.force_flush().expect("force_flush succeeds");

let rms = exporter.get_finished_metrics().expect("metrics exported");
let metric = |name: &str| {
rms.iter()
.flat_map(ResourceMetrics::scope_metrics)
.flat_map(ScopeMetrics::metrics)
.find(|m| m.name() == name)
.unwrap_or_else(|| panic!("{name} missing from exported stream"))
.data()
};

// The reason-attributed failure counter.
let AggregatedMetrics::U64(MetricData::Sum(sum)) =
metric(ourios_semconv::OURIOS_MINER_PARSE_FAILURES)
else {
panic!("parse_failures should be a u64 sum");
};
let ceiling_failures: u64 = sum
.data_points()
.filter(|dp| {
has_attr(
dp.attributes(),
ourios_semconv::OURIOS_MINER_PARSE_FAILURE_REASON,
"template_ceiling",
) && has_attr(dp.attributes(), ourios_semconv::OURIOS_TENANT, "saturated")
})
Comment thread
jensholdgaard marked this conversation as resolved.
.map(opentelemetry_sdk::metrics::data::SumDataPoint::value)
.sum();
assert_eq!(
ceiling_failures, 2,
"both diverted lines counted under reason = template_ceiling",
);

// The gauge reads the ceiling.
let AggregatedMetrics::U64(MetricData::Gauge(gauge)) =
metric(ourios_semconv::OURIOS_MINER_TEMPLATE_COUNT)
else {
panic!("template.count should be a u64 gauge");
};
let count: u64 = gauge
.data_points()
.find(|dp| has_attr(dp.attributes(), ourios_semconv::OURIOS_TENANT, "saturated"))
.map(opentelemetry_sdk::metrics::data::GaugeDataPoint::value)
.expect("the saturated tenant's data point");
assert_eq!(u64::from(ceiling), count, "the gauge reads the ceiling");
}
10 changes: 8 additions & 2 deletions crates/ourios-semconv/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,7 @@ pub const OURIOS_MINER_MERGES: &str = "ourios.miner.merges";
pub const OURIOS_MINER_PARAMS_OVERFLOW: &str = "ourios.miner.params.overflow";

/// `ourios.miner.params.overflow.utilization` (gauge, unit `1`).
pub const OURIOS_MINER_PARAMS_OVERFLOW_UTILIZATION: &str =
"ourios.miner.params.overflow.utilization";
pub const OURIOS_MINER_PARAMS_OVERFLOW_UTILIZATION: &str = "ourios.miner.params.overflow.utilization";

/// `ourios.miner.parse_failures` (counter, unit `{failure}`).
pub const OURIOS_MINER_PARSE_FAILURES: &str = "ourios.miner.parse_failures";
Expand Down Expand Up @@ -139,8 +138,10 @@ pub const OURIOS_STORAGE_PARQUET_FILE_SIZE: &str = "ourios.storage.parquet.file.
/// `ourios.wal.append.duration` (histogram, unit `s`).
pub const OURIOS_WAL_APPEND_DURATION: &str = "ourios.wal.append.duration";


// Attribute keys.


/// `ourios.audit_sink.flush.outcome` attribute key.
pub const OURIOS_AUDIT_SINK_FLUSH_OUTCOME: &str = "ourios.audit_sink.flush.outcome";

Expand All @@ -150,6 +151,9 @@ pub const OURIOS_COMPACTION_RESULT: &str = "ourios.compaction.result";
/// `ourios.io.direction` attribute key.
pub const OURIOS_IO_DIRECTION: &str = "ourios.io.direction";

/// `ourios.miner.parse_failure.reason` attribute key.
pub const OURIOS_MINER_PARSE_FAILURE_REASON: &str = "ourios.miner.parse_failure.reason";

/// `ourios.miner.template_change` attribute key.
pub const OURIOS_MINER_TEMPLATE_CHANGE: &str = "ourios.miner.template_change";

Expand All @@ -168,6 +172,8 @@ pub const OURIOS_SINK_FLUSH_TRIGGER: &str = "ourios.sink.flush.trigger";
/// `ourios.tenant` attribute key.
pub const OURIOS_TENANT: &str = "ourios.tenant";



// Log event names (the server's own dogfooded logs; every `tracing` call
// site names its event with one of these — `weaver registry live-check`
// enforces it at emission time).
Expand Down
Loading
Loading