ottyl/src/engine/snapshot/mod.rs

//! `EngineSnapshot` is the actor's live state: ingested log lines, declared params/metrics,
//! computed metric buckets, and the sort-helper indexes built on demand — all stored in one
//! flat `BTreeMap<IndexKey, DataValue>` and read/written through the left-right index in
//! `super::handle`. Its impl is split across the submodules below by concern; this file holds
//! the struct itself plus the accessors every submodule builds on.

pub(crate) mod filter;
pub(crate) mod metrics;
pub(crate) mod params;
pub(crate) mod query;
pub(crate) mod sort;

pub(crate) use filter::eval_json_comparison;

use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
use crate::engine::index_key::IndexKey;
use crate::engine::message::{EngineQuery, EngineResponse};
use crate::engine::timestamp::Timestamp;
use query::handle_query_inner;
use regex::Regex;
use sort::{SortHelperAccess, SortHelperIndexEntry};
use std::collections::{BTreeMap, HashMap};

#[derive(Debug, Clone, Default)]
pub(crate) struct EngineSnapshot {
    pub(crate) index: BTreeMap<IndexKey, DataValue>,
    pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
    /// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a
    /// regex is far more expensive than matching with one, and `apply_all_metrics`/
    /// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or
    /// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather
    /// than recompiling from the stored pattern string on every line.
    pub(crate) param_regex_cache: HashMap<String, Regex>,
    pub(crate) metric_regex_cache: HashMap<String, Regex>,
    pub(crate) log_count: usize,
    pub(crate) approx_bytes_total: u64,
    pub(crate) ingest_lag_high_watermark_ms: i64,
}

impl EngineSnapshot {
    pub(crate) fn metric_declarations_range(
        &self,
    ) -> impl Iterator<
        Item = (
            &str,
            &str,
            &[String],
            Option<&str>,
            Option<&BTreeMap<String, String>>,
            Option<&BTreeMap<String, String>>,
        ),
    > {
        self.index
            .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit)
            .filter_map(|(key, value)| match (key, value) {
                (
                    IndexKey::MetricDeclaration { name },
                    DataValue::MetricDeclaration {
                        message_match_regex,
                        label_params,
                        value_param,
                        matched_labels,
                        unmatched_labels,
                    },
                ) => Some((
                    name.as_str(),
                    message_match_regex.as_str(),
                    label_params.as_slice(),
                    value_param.as_deref(),
                    matched_labels.as_ref(),
                    unmatched_labels.as_ref(),
                )),
                _ => None,
            })
    }

    pub(crate) fn param_declarations_range(
        &self,
    ) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool, bool, bool)> {
        self.index
            .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit)
            .filter_map(|(key, value)| match (key, value) {
                (
                    IndexKey::ParamDeclaration { name },
                    DataValue::ParamDeclaration(ParamDeclaration {
                        name: _,
                        message_match_regex,
                        capture_group,
                        json_field_path,
                        kind,
                        elapsed_since_last,
                        auto_derived,
                        log_stream_gated,
                    }),
                ) => Some((
                    name.as_str(),
                    message_match_regex.as_str(),
                    capture_group.as_str(),
                    json_field_path.as_deref(),
                    *kind,
                    *elapsed_since_last,
                    *auto_derived,
                    *log_stream_gated,
                )),
                _ => None,
            })
    }
}

pub(crate) fn approx_log_bytes(text: &str) -> u64 {
    text.len() as u64
}

impl EngineSnapshot {
    pub(crate) fn from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
        let mut snapshot = EngineSnapshot { index, ..Default::default() };
        snapshot.rebuild_regex_caches();
        snapshot
    }

    /// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in
    /// `index`. Needed whenever a snapshot's declarations are populated by inserting straight
    /// into `index` rather than through the normal declare/insert methods that keep the caches
    /// in sync incrementally (e.g. `from_index` building a snapshot from a bare index).
    pub(crate) fn rebuild_regex_caches(&mut self) {
        self.param_regex_cache = self
            .param_declarations_range()
            .filter_map(|(name, message_match_regex, ..)| {
                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
            })
            .collect();
        self.metric_regex_cache = self
            .metric_declarations_range()
            .filter_map(|(name, message_match_regex, ..)| {
                regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re))
            })
            .collect();
    }

    pub(crate) fn evict_excess_logs(&mut self, max_logs: Option<usize>) {
        let Some(max_logs) = max_logs else { return };
        while self.log_count > max_logs {
            let Some(oldest_key) = self
                .index
                .range(IndexKey::Start..IndexKey::LogParamSplit)
                .next()
                .map(|(key, _)| key.clone())
            else {
                break;
            };
            if let Some(DataValue::LogLiteral { text }) = self.index.get(&oldest_key) {
                self.approx_bytes_total = self.approx_bytes_total.saturating_sub(approx_log_bytes(text));
            }
            self.index.remove(&oldest_key);
            self.log_count -= 1;
        }
    }

    pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
        self.index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
            match (key, value) {
                (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
                    Some((*sequence, *timestamp, text.as_str()))
                }
                _ => None,
            }
        })
    }

    pub(crate) fn logs_since(
        &self,
        before_sequence: u64,
        since_ms: i64,
        now_ms: i64,
    ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
        self.logs_range().filter(move |(sequence, timestamp, _)| {
            *sequence < before_sequence
                && timestamp.as_millis() > since_ms
                && timestamp.as_millis() <= now_ms
        })
    }
}

impl EngineSnapshot {
    pub(crate) fn handle_query_ro(&self, query: EngineQuery) -> EngineResponse {
        handle_query_inner(self, query, SortHelperAccess::Immutable(&self.sort_helper_indexes))
    }

    pub(crate) fn has_sort_helper_index(&self, sort_by: &str) -> bool {
        self.sort_helper_indexes.contains_key(sort_by)
    }
}