Menu
Public repository / a5b1e159b6b764bd248f442cf3ae484a147dbc62
akurai-log / src/lib.rs
Latest change 2577f04e8d3b30f5b729c10273a73b0d54351113 - feat: enforce tenant-safe log contract by Ólafur Búi Ólafsson
//! `akurai-log` — a [`tracing`] layer that ships structured log events to
//! AkurAI-Monitor's `POST /api/ingest` endpoint, in addition to normal stdout.
//!
//! Events carry one privacy-safe, optional correlation context. Correlation IDs
//! are opaque strings and are emitted as top-level fields so Monitor queries
//! cannot accidentally mix tenant scope with arbitrary structured fields.
use std::sync::mpsc::{sync_channel, SyncSender};
use std::sync::OnceLock;
use std::time::{Duration, Instant};
use tracing::field::{Field, Visit};
use tracing_subscriber::layer::{Context, Layer};
use tracing_subscriber::prelude::*;
use tracing_subscriber::{fmt, EnvFilter};
pub const CONTRACT_VERSION: u8 = 1;
const CHANNEL_CAPACITY: usize = 8192;
const BATCH_MAX: usize = 128;
pub const MAX_LINE_LENGTH: usize = 8192;
const MAX_FIELD_LENGTH: usize = 1024;
const MAX_FIELDS: usize = 64;
pub const MAX_ID_LENGTH: usize = 128;
const FLUSH_INTERVAL: Duration = Duration::from_secs(2);
const CORRELATION_FIELDS: [&str; 5] = [
"organization_id",
"workspace_id",
"installation_id",
"deployment_id",
"request_id",
];
/// Privacy policy applied before a field can reach Monitor.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PrivacyClass {
Correlation,
Metadata,
Sensitive,
Prohibited,
}
/// Classify a contract field by name. Correlation fields are kept out of the
/// free-form `fields` object; prohibited fields are dropped completely.
pub fn classify_field(name: &str) -> PrivacyClass {
let Some(key) = normalize_field_name(name) else {
return PrivacyClass::Prohibited;
};
if CORRELATION_FIELDS.contains(&key.as_str()) {
return PrivacyClass::Correlation;
}
if ["user_id", "session_id", "ip_address"]
.iter()
.any(|part| key.contains(part))
{
return PrivacyClass::Sensitive;
}
if [
"authorization",
"api_key",
"apikey",
"bearer",
"password",
"passwd",
"secret",
"token",
"cookie",
"email",
"note",
"content",
"file",
"body",
"prompt",
]
.iter()
.any(|part| key.contains(part))
{
return PrivacyClass::Prohibited;
}
PrivacyClass::Metadata
}
fn normalize_field_name(name: &str) -> Option<String> {
if name.is_empty() || !name.is_ascii() {
return None;
}
let chars: Vec<char> = name.chars().collect();
let mut normalized = String::with_capacity(name.len());
for (index, ch) in chars.iter().copied().enumerate() {
if !ch.is_ascii_alphanumeric() && ch != '_' {
return None;
}
if ch.is_ascii_uppercase() {
let boundary = index > 0
&& (chars[index - 1].is_ascii_lowercase()
|| chars[index - 1].is_ascii_digit()
|| (chars[index - 1].is_ascii_uppercase()
&& chars
.get(index + 1)
.is_some_and(|next| next.is_ascii_lowercase())));
if boundary {
normalized.push('_');
}
}
normalized.push(ch.to_ascii_lowercase());
}
Some(normalized)
}
/// A validated, optional tenant/request correlation context.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CorrelationContext {
organization_id: Option<String>,
workspace_id: Option<String>,
installation_id: Option<String>,
deployment_id: Option<String>,
request_id: Option<String>,
}
impl CorrelationContext {
pub fn new() -> Self {
Self::default()
}
pub fn from_env() -> Self {
let mut context = Self::new();
context.set_from_env("organization_id", "MONITOR_ORGANIZATION_ID");
context.set_from_env("workspace_id", "MONITOR_WORKSPACE_ID");
context.set_from_env("installation_id", "MONITOR_INSTALLATION_ID");
context.set_from_env("deployment_id", "MONITOR_DEPLOYMENT_ID");
context.set_from_env("request_id", "MONITOR_REQUEST_ID");
context
}
pub fn with_organization_id(mut self, value: impl Into<String>) -> Self {
self.organization_id = sanitize_id(&value.into());
self
}
pub fn with_workspace_id(mut self, value: impl Into<String>) -> Self {
self.workspace_id = sanitize_id(&value.into());
self
}
pub fn with_installation_id(mut self, value: impl Into<String>) -> Self {
self.installation_id = sanitize_id(&value.into());
self
}
pub fn with_deployment_id(mut self, value: impl Into<String>) -> Self {
self.deployment_id = sanitize_id(&value.into());
self
}
pub fn with_request_id(mut self, value: impl Into<String>) -> Self {
self.request_id = sanitize_id(&value.into());
self
}
pub fn organization_id(&self) -> Option<&str> {
self.organization_id.as_deref()
}
pub fn workspace_id(&self) -> Option<&str> {
self.workspace_id.as_deref()
}
pub fn installation_id(&self) -> Option<&str> {
self.installation_id.as_deref()
}
pub fn deployment_id(&self) -> Option<&str> {
self.deployment_id.as_deref()
}
pub fn request_id(&self) -> Option<&str> {
self.request_id.as_deref()
}
/// Merge event scope into configured scope. A conflicting value is
/// rejected instead of allowing one event to carry mixed tenant scope.
pub fn merge(&self, event: &Self) -> Option<Self> {
fn merge_value(base: &Option<String>, event: &Option<String>) -> Option<Option<String>> {
match (base, event) {
(Some(left), Some(right)) if left != right => None,
(Some(left), _) => Some(Some(left.clone())),
(None, Some(right)) => Some(Some(right.clone())),
(None, None) => Some(None),
}
}
Some(Self {
organization_id: merge_value(&self.organization_id, &event.organization_id)?,
workspace_id: merge_value(&self.workspace_id, &event.workspace_id)?,
installation_id: merge_value(&self.installation_id, &event.installation_id)?,
deployment_id: merge_value(&self.deployment_id, &event.deployment_id)?,
request_id: merge_value(&self.request_id, &event.request_id)?,
})
}
fn set_from_env(&mut self, field: &str, variable: &str) {
if let Ok(value) = std::env::var(variable) {
set_correlation_value(self, field, &value);
}
}
}
fn is_correlation_field(name: &str) -> bool {
normalize_field_name(name)
.map(|name| CORRELATION_FIELDS.contains(&name.as_str()))
.unwrap_or(false)
}
fn set_correlation_value(context: &mut CorrelationContext, field: &str, value: &str) {
let value = sanitize_id(value);
let Some(field) = normalize_field_name(field) else {
return;
};
match field.as_str() {
"organization_id" => context.organization_id = value,
"workspace_id" => context.workspace_id = value,
"installation_id" => context.installation_id = value,
"deployment_id" => context.deployment_id = value,
"request_id" => context.request_id = value,
_ => {}
}
}
fn sanitize_id(value: &str) -> Option<String> {
let value = value.trim();
if value.is_empty() || value.len() > MAX_ID_LENGTH {
return None;
}
if value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | ':' | '-'))
{
Some(value.to_string())
} else {
None
}
}
/// Redact values that could contain credentials or direct identifiers.
pub fn redact_text(value: &str) -> String {
let mut decoded = value.to_string();
let max_rounds = decoded.len().saturating_add(1);
for _ in 0..max_rounds {
let next = percent_decode_once(&decoded);
if next == decoded {
break;
}
decoded = next;
}
let normalized = decoded.replace(['\r', '\n'], " ");
let lower = normalized.to_ascii_lowercase();
if has_secret_pattern(&lower) || lower.contains("-----begin ") {
return "[REDACTED_SECRET]".to_string();
}
if looks_like_email(&normalized) {
return "[REDACTED_EMAIL]".to_string();
}
normalized
}
fn has_secret_pattern(lower: &str) -> bool {
for marker in [
"authorization",
"api_key",
"apikey",
"api-key",
"password",
"passwd",
"secret",
"token",
"cookie",
"bearer",
] {
let mut offset = 0;
while let Some(found) = lower[offset..].find(marker) {
let start = offset + found;
let after = &lower[start + marker.len()..];
let after_quotes = after.trim_start_matches(['"', '\'']);
let value = if marker == "bearer"
&& !after_quotes.starts_with(':')
&& !after_quotes.starts_with('=')
{
after_quotes.trim_start_matches(|ch: char| ch.is_ascii_whitespace())
} else {
let separator =
after_quotes.trim_start_matches(|ch: char| ch.is_ascii_whitespace());
if !separator.starts_with(':') && !separator.starts_with('=') {
offset = start + marker.len();
continue;
}
separator[1..]
.trim_matches(|ch: char| ch.is_ascii_whitespace() || ch == '"' || ch == '\'')
};
if !value.is_empty() {
return true;
}
offset = start + marker.len();
}
}
false
}
fn percent_decode_once(value: &str) -> String {
fn hex(ch: u8) -> Option<u8> {
match ch {
b'0'..=b'9' => Some(ch - b'0'),
b'a'..=b'f' => Some(ch - b'a' + 10),
b'A'..=b'F' => Some(ch - b'A' + 10),
_ => None,
}
}
let bytes = value.as_bytes();
let mut decoded = Vec::with_capacity(bytes.len());
let mut index = 0;
while index < bytes.len() {
if bytes[index] == b'%' && index + 2 < bytes.len() {
if let (Some(high), Some(low)) = (hex(bytes[index + 1]), hex(bytes[index + 2])) {
decoded.push((high << 4) | low);
index += 3;
continue;
}
}
decoded.push(bytes[index]);
index += 1;
}
String::from_utf8_lossy(&decoded).into_owned()
}
fn truncate_text(value: &str, max: usize) -> String {
redact_text(value).chars().take(max).collect()
}
fn looks_like_email(value: &str) -> bool {
value.split_whitespace().any(|part| {
let part = part.trim_matches(|ch: char| ",.;:()[]{}<>\"'".contains(ch));
let Some(at) = part.find('@') else {
return false;
};
let (left, right) = part.split_at(at);
let right = &right[1..];
!left.is_empty() && !right.is_empty() && right.contains('.')
})
}
fn sanitize_field_value(name: &str, value: serde_json::Value) -> Option<serde_json::Value> {
if classify_field(name) != PrivacyClass::Metadata {
return None;
}
match value {
serde_json::Value::String(value) => Some(serde_json::Value::String(truncate_text(
&value,
MAX_FIELD_LENGTH,
))),
serde_json::Value::Number(_) | serde_json::Value::Bool(_) => Some(value),
_ => None,
}
}
/// Keep only bounded scalar metadata and remove prohibited content.
pub fn sanitize_fields(
fields: serde_json::Map<String, serde_json::Value>,
) -> serde_json::Map<String, serde_json::Value> {
let mut fields: Vec<_> = fields.into_iter().collect();
fields.sort_by(|(left, _), (right, _)| left.cmp(right));
fields
.into_iter()
.take(MAX_FIELDS)
.filter_map(|(name, value)| sanitize_field_value(&name, value).map(|value| (name, value)))
.collect()
}
/// One structured log record queued for shipment.
#[derive(Clone)]
struct Record {
source: String,
level: &'static str,
target: String,
message: String,
fields: serde_json::Map<String, serde_json::Value>,
correlation: CorrelationContext,
pid: i64,
ts: i64,
}
/// Sender is populated at `init()` from runtime env — kept as a `OnceLock`
/// because the initializer needs runtime input (the configured channel).
static SENDER: OnceLock<SyncSender<Record>> = OnceLock::new();
/// Ingestion targets that must be ignored to avoid a log→ship→log loop.
fn is_noise(target: &str) -> bool {
target.starts_with("reqwest")
|| target.starts_with("hyper")
|| target.starts_with("rustls")
|| target.starts_with("h2")
|| target.starts_with("want")
|| target.starts_with("mio")
}
fn level_str(level: &tracing::Level) -> &'static str {
match *level {
tracing::Level::ERROR => "error",
tracing::Level::WARN => "warn",
tracing::Level::INFO => "info",
tracing::Level::DEBUG => "debug",
tracing::Level::TRACE => "trace",
}
}
/// Collects the event's `message`, correlation IDs, and safe metadata.
#[derive(Default)]
struct FieldVisitor {
message: String,
fields: serde_json::Map<String, serde_json::Value>,
correlation: CorrelationContext,
correlation_fields: serde_json::Map<String, serde_json::Value>,
}
impl FieldVisitor {
fn record_value(&mut self, field: &Field, value: serde_json::Value) {
let name = field.name();
if is_correlation_field(name) {
let canonical = normalize_field_name(name).expect("correlation field is ASCII");
if name == canonical || !self.correlation_fields.contains_key(&canonical) {
self.correlation_fields.insert(canonical, value);
}
} else if let Some(value) = sanitize_field_value(name, value) {
self.fields.insert(name.to_string(), value);
}
}
fn finalize_correlation(&mut self) {
for (name, value) in std::mem::take(&mut self.correlation_fields) {
if let serde_json::Value::String(value) = value {
set_correlation_value(&mut self.correlation, &name, &value);
}
}
}
}
impl Visit for FieldVisitor {
fn record_str(&mut self, field: &Field, value: &str) {
if field.name() == "message" {
self.message = value.to_string();
} else {
self.record_value(field, serde_json::Value::from(value));
}
}
fn record_i64(&mut self, field: &Field, value: i64) {
self.record_value(field, serde_json::Value::from(value));
}
fn record_u64(&mut self, field: &Field, value: u64) {
self.record_value(field, serde_json::Value::from(value));
}
fn record_bool(&mut self, field: &Field, value: bool) {
self.record_value(field, serde_json::Value::from(value));
}
fn record_f64(&mut self, field: &Field, value: f64) {
if let Some(value) = serde_json::Number::from_f64(value) {
self.record_value(field, serde_json::Value::Number(value));
}
}
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
let rendered = format!("{value:?}");
if field.name() == "message" {
self.message = rendered;
} else {
self.record_value(field, serde_json::Value::from(rendered));
}
}
}
/// The tracing layer that forwards events into the shipment channel.
struct MonitorLayer {
source: String,
pid: i64,
correlation: CorrelationContext,
}
impl<S: tracing::Subscriber> Layer<S> for MonitorLayer {
fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
let target = event.metadata().target();
if is_noise(target) {
return;
}
let Some(sender) = SENDER.get() else {
return;
};
let mut visitor = FieldVisitor::default();
event.record(&mut visitor);
visitor.finalize_correlation();
let Some(correlation) = self.correlation.merge(&visitor.correlation) else {
return;
};
let record = Record {
source: truncate_text(&self.source, MAX_LINE_LENGTH),
level: level_str(event.metadata().level()),
target: truncate_text(target, MAX_LINE_LENGTH),
message: truncate_text(&visitor.message, MAX_LINE_LENGTH),
fields: visitor.fields,
correlation,
pid: self.pid,
ts: now_secs(),
};
// Drop on backpressure — shipping logs must never block the app.
let _ = sender.try_send(record);
}
}
fn now_secs() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
/// Initialize stdout logging plus shipment to AkurAI-Monitor using context
/// from `MONITOR_*_ID` environment variables.
pub fn init(app_name: &str) {
init_with_context(app_name, CorrelationContext::from_env());
}
/// Initialize logging with an explicit, privacy-validated correlation context.
pub fn init_with_context(app_name: &str, correlation: CorrelationContext) {
let stdout_filter = EnvFilter::try_from_default_env()
.unwrap_or_else(|_| EnvFilter::new(format!("{}=info", app_name.replace('-', "_"))));
let url = std::env::var("MONITOR_INGEST_URL")
.ok()
.filter(|s| !s.is_empty());
let token = std::env::var("MONITOR_INGEST_TOKEN")
.ok()
.filter(|s| !s.is_empty());
let monitor_layer = match (url, token) {
(Some(url), Some(token)) => {
let host_id = std::env::var("MONITOR_HOST_ID")
.ok()
.and_then(|s| s.parse::<i64>().ok());
let app_id = std::env::var("MONITOR_APP_ID")
.ok()
.and_then(|s| s.parse::<i64>().ok());
let ship_level =
std::env::var("MONITOR_LOG_LEVEL").unwrap_or_else(|_| "debug".to_string());
let (tx, rx) = sync_channel::<Record>(CHANNEL_CAPACITY);
let _ = SENDER.set(tx);
spawn_shipper(rx, url, token, host_id, app_id);
let filter = EnvFilter::new(&ship_level);
Some(
MonitorLayer {
source: app_name.to_string(),
pid: std::process::id() as i64,
correlation,
}
.with_filter(filter),
)
}
_ => None,
};
tracing_subscriber::registry()
.with(
fmt::layer()
.with_writer(std::io::stdout)
.with_filter(stdout_filter),
)
.with(monitor_layer)
.init();
}
/// Background thread: batch records and POST them to `/api/ingest`.
fn spawn_shipper(
rx: std::sync::mpsc::Receiver<Record>,
base_url: String,
token: String,
host_id: Option<i64>,
app_id: Option<i64>,
) {
let endpoint = format!("{}/api/ingest", base_url.trim_end_matches('/'));
std::thread::Builder::new()
.name("akurai-log-shipper".into())
.spawn(move || {
let client = match reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(5))
.build()
{
Ok(c) => c,
Err(_) => return,
};
let mut batch: Vec<Record> = Vec::with_capacity(BATCH_MAX);
let mut last_flush = Instant::now();
loop {
match rx.recv_timeout(FLUSH_INTERVAL) {
Ok(record) => {
batch.push(record);
while batch.len() < BATCH_MAX {
match rx.try_recv() {
Ok(r) => batch.push(r),
Err(_) => break,
}
}
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
flush(&client, &endpoint, &token, host_id, app_id, &mut batch);
return;
}
}
if !batch.is_empty()
&& (batch.len() >= BATCH_MAX || last_flush.elapsed() >= FLUSH_INTERVAL)
{
flush(&client, &endpoint, &token, host_id, app_id, &mut batch);
last_flush = Instant::now();
}
}
})
.ok();
}
fn flush(
client: &reqwest::blocking::Client,
endpoint: &str,
token: &str,
host_id: Option<i64>,
app_id: Option<i64>,
batch: &mut Vec<Record>,
) {
if batch.is_empty() {
return;
}
let logs: Vec<serde_json::Value> = batch
.iter()
.map(|record| {
serde_json::json!({
"source": record.source,
"line": record.message,
"level": record.level,
"target": record.target,
"pid": record.pid,
"origin": "server",
"fields": if record.fields.is_empty() { serde_json::Value::Null }
else { serde_json::Value::Object(record.fields.clone()) },
"host_id": host_id,
"application_id": app_id,
"organization_id": record.correlation.organization_id(),
"workspace_id": record.correlation.workspace_id(),
"installation_id": record.correlation.installation_id(),
"deployment_id": record.correlation.deployment_id(),
"request_id": record.correlation.request_id(),
"ts": record.ts,
})
})
.collect();
let body = serde_json::json!({ "logs": logs });
// Best-effort; a failed ship is dropped, never retried into unbounded memory.
let _ = client.post(endpoint).bearer_auth(token).json(&body).send();
batch.clear();
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn noise_targets_are_filtered() {
assert!(is_noise("reqwest::connect"));
assert!(is_noise("hyper::proto::h1"));
assert!(is_noise("rustls::client"));
assert!(!is_noise("akurai_dns::server"));
assert!(!is_noise("notes_api::auth"));
}
#[test]
fn levels_map_to_lowercase_strings() {
assert_eq!(level_str(&tracing::Level::ERROR), "error");
assert_eq!(level_str(&tracing::Level::WARN), "warn");
assert_eq!(level_str(&tracing::Level::INFO), "info");
assert_eq!(level_str(&tracing::Level::DEBUG), "debug");
assert_eq!(level_str(&tracing::Level::TRACE), "trace");
}
#[test]
fn visitor_default_is_empty_and_json_safe() {
let v = FieldVisitor::default();
assert!(v.message.is_empty());
assert!(v.fields.is_empty());
assert_eq!(v.correlation, CorrelationContext::default());
let value = serde_json::to_value(&v.fields).expect("serializable");
assert!(value.is_object());
}
#[test]
fn fixture_privacy_and_correlation_contract_is_stable() {
let fixture: serde_json::Value =
serde_json::from_str(include_str!("../fixtures/monitor-contract.json"))
.expect("fixture JSON");
assert_eq!(fixture["version"], CONTRACT_VERSION);
for field in CORRELATION_FIELDS {
assert!(fixture["correlation_fields"]
.as_array()
.expect("correlation fields")
.iter()
.any(|value| value == field));
}
for value in fixture["invalid_correlation"]
.as_array()
.expect("invalid IDs")
{
assert!(sanitize_id(value.as_str().expect("string")).is_none());
}
for case in fixture["classification_cases"]
.as_array()
.expect("classification cases")
{
let name = case["name"].as_str().expect("field name");
let expected = match case["class"].as_str().expect("privacy class") {
"correlation" => PrivacyClass::Correlation,
"metadata" => PrivacyClass::Metadata,
"sensitive" => PrivacyClass::Sensitive,
"prohibited" => PrivacyClass::Prohibited,
other => panic!("unknown privacy class {other}"),
};
assert_eq!(
classify_field(name),
expected,
"classification mismatch for {name}"
);
}
for case in fixture["normalization_cases"]
.as_array()
.expect("normalization cases")
{
let input = case["input"].as_str().expect("normalization input");
let expected = case["output"].as_str().expect("normalization output");
assert_eq!(normalize_field_name(input).as_deref(), Some(expected));
}
let mut fields = serde_json::Map::new();
for (name, value) in fixture["unsafe_fields"].as_object().expect("unsafe fields") {
fields.insert(name.clone(), value.clone());
}
assert!(sanitize_fields(fields).is_empty());
let safe_fields = fixture["safe_fields"]
.as_object()
.expect("safe fields")
.clone();
assert_eq!(fixture["max_fields"].as_u64(), Some(MAX_FIELDS as u64));
let bounded_fields = (0..(MAX_FIELDS + 10))
.map(|index| {
(
format!("field_{index}"),
serde_json::Value::from(index as u64),
)
})
.collect();
assert_eq!(sanitize_fields(bounded_fields).len(), MAX_FIELDS);
let ordering_fields = fixture["ordering_keys"]
.as_array()
.expect("ordering keys")
.iter()
.map(|key| {
(
key.as_str().expect("ordering key").to_string(),
serde_json::Value::Bool(true),
)
})
.collect();
let surviving: Vec<String> = sanitize_fields(ordering_fields).keys().cloned().collect();
let expected: Vec<String> = fixture["ordering_survives"]
.as_array()
.expect("ordering survives")
.iter()
.map(|key| key.as_str().expect("surviving key").to_string())
.collect();
assert_eq!(surviving, expected);
assert_eq!(sanitize_fields(safe_fields.clone()), safe_fields);
for case in fixture["redaction_cases"]
.as_array()
.expect("redaction cases")
{
assert_eq!(
redact_text(case["input"].as_str().expect("redaction input")),
case["output"].as_str().expect("redaction output")
);
}
let mut context = CorrelationContext::new();
set_correlation_value(&mut context, "organizationId", "org-a");
assert_eq!(context.organization_id(), Some("org-a"));
}
#[test]
fn conflicting_tenant_context_is_rejected() {
let base = CorrelationContext::new().with_organization_id("org-a");
let event = CorrelationContext::new().with_organization_id("org-b");
assert!(base.merge(&event).is_none());
}
#[test]
fn no_tenant_context_remains_valid() {
assert_eq!(
CorrelationContext::default().merge(&CorrelationContext::default()),
Some(CorrelationContext::default())
);
}
}