Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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")
})
.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");
}
3 changes: 3 additions & 0 deletions crates/ourios-semconv/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,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 Down
33 changes: 33 additions & 0 deletions semconv/registry/attributes.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,39 @@ groups:
brief: >-
The kind of template change an `ourios.miner.merges` event
records.
- id: ourios.miner.parse_failure.reason
type:
members:
- id: below_floor
value: "below_floor"
stability: development
brief: Best candidate similarity fell below the RFC 0001 §6.3 floor.
- id: empty_line
value: "empty_line"
stability: development
brief: The line was empty or whitespace-only after tokenization.
- id: tokenizer_failure
value: "tokenizer_failure"
stability: development
brief: The tokenizer rejected the line (today, an embedded NUL byte).
- id: degenerate_widening
value: "degenerate_widening"
stability: development
brief: Attaching would have widened the template to a degenerate form (RFC 0001 §6.4).
- id: line_too_long
value: "line_too_long"
stability: development
brief: The line tokenized past `max_line_tokens` (RFC 0023 §3.1).
- id: template_ceiling
value: "template_ceiling"
stability: development
brief: The tenant is at `max_templates`; the would-mint line was diverted (RFC 0023 §3.1).
stability: development
brief: >-
Why a line took the parse-failure path (the
`ourios.miner.parse_failures` counter's cause dimension,
following the OTel error.type convention of an attribute on
one instrument rather than per-cause counters).
- id: ourios.io.direction
type:
members:
Expand Down
2 changes: 2 additions & 0 deletions semconv/registry/metrics.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,8 @@ groups:
requirement_level: required
- ref: ourios.service
requirement_level: recommended
- ref: ourios.miner.parse_failure.reason
requirement_level: required

- id: metric.ourios.miner.params.overflow
type: metric
Expand Down