saddle-observability 0.3.25

//! Authority-free output of an existing admission decision. Runtime owns and
//! consumes the original receipt; this module neither admits nor creates roots.
use crate::{CapacityDimension, DiagnosticSubmission, EmergencyDiagnosticHandle, Observer};
use saddle_admission::{
    ProfuseGwAdmissionObservationReceipt, ProfuseGwCapacityBottleneck, ProfuseGwCapacityDecision,
    ProfuseGwCapacityRejectReason, ProfuseGwCapacitySnapshot,
};
use saddle_core::{ContextFact, ContextLabel, RequestExecutionView, RequestViewPhase};
use serde::{Serialize, Serializer, ser::SerializeStruct};

/// Fixed observation facts, not a permit, account or second admission receipt.
///
/// ```compile_fail
/// let facts = saddle_observability::AdmissionCapacityFacts {};
/// ```
pub struct AdmissionCapacityFacts {
    snapshot: ProfuseGwCapacitySnapshot,
    used: usize,
    decision: ProfuseGwCapacityDecision,
    reason: Option<ProfuseGwCapacityRejectReason>,
    elapsed_ms: u64,
}
impl AdmissionCapacityFacts {
    pub fn from_receipt(receipt: &ProfuseGwAdmissionObservationReceipt) -> Self {
        Self {
            snapshot: receipt.snapshot(),
            used: receipt.used(),
            decision: receipt.decision(),
            reason: receipt.reason(),
            elapsed_ms: receipt.elapsed_ms(),
        }
    }
}

/// Borrow the original view, or report only real pre-root application/lifecycle.
/// Neither form creates or retains a request root or task.
///
/// ```compile_fail
/// fn retain(view: &saddle_core::RequestExecutionView)
///     -> saddle_observability::AdmissionEventContext<'static> {
///     saddle_observability::AdmissionEventContext::Rooted(view)
/// }
/// ```
#[derive(Clone, Copy)]
pub enum AdmissionEventContext<'a> {
    Rooted(&'a RequestExecutionView),
    Unrooted {
        application: &'a ContextLabel,
        lifecycle: RequestViewPhase,
    },
}
impl Serialize for AdmissionEventContext<'_> {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        match self {
            Self::Rooted(view) => view.serialize(serializer),
            Self::Unrooted {
                application,
                lifecycle,
            } => {
                let mut out = serializer.serialize_struct("UnrootedRequestContext", 20)?;
                out.serialize_field("schema_version", &2u8)?;
                out.serialize_field("application", &ContextFact::Present(application))?;
                out.serialize_field("lifecycle", &ContextFact::Present(lifecycle))?;
                for field in [
                    "local_request",
                    "publication",
                    "call_application",
                    "module",
                    "service",
                    "operation",
                    "trace_id",
                    "request",
                    "span_id",
                    "route",
                    "attempt",
                    "rpc_id",
                    "zone",
                    "db_operation",
                    "scope",
                    "task",
                    "target",
                ] {
                    out.serialize_field(field, &ContextFact::<()>::NotEstablished)?;
                }
                out.end()
            }
        }
    }
}

/// Construction outcome is independent of the original admission decision.
/// Concrete error details continue through the original diagnostic mechanism.
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(tag = "status", content = "failure", rename_all = "snake_case")]
pub enum AdmissionConstruction {
    NotAttempted,
    Ready,
    Cancelled,
    Failed(AdmissionConstructionFailure),
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AdmissionConstructionFailure {
    Storage,
    Deadline,
    RequestIdentity,
    Context,
    RejectionTaskBytes,
    RejectionTaskSlot,
}

/// Four independent synchronous submissions, in CPU/memory/DB/outbound order.
/// Enqueued is not written. Partial acceptance is deliberately not collapsed.
#[must_use]
pub struct AdmissionCapacitySubmission {
    pub cpu: DiagnosticSubmission,
    pub memory: DiagnosticSubmission,
    pub database: DiagnosticSubmission,
    pub profuse_contract: DiagnosticSubmission,
}

#[derive(Serialize)]
struct CapacityRecord<'a> {
    schema_version: u8,
    timestamp_unix_ms: u128,
    level: &'static str,
    event: &'static str,
    stage: &'static str,
    context: AdmissionEventContext<'a>,
    capacity_dimension: &'static str,
    budget: usize,
    limit: usize,
    used: usize,
    bottleneck: &'static str,
    decision: &'static str,
    outcome: &'static str,
    reject_reason: Option<&'static str>,
    elapsed_ms: u64,
    construction: AdmissionConstruction,
}

impl Observer {
    /// Emit original four-dimensional facts without synthetic Call/Event/root.
    /// Metrics use the existing fixed dimensions even when output is absent.
    pub fn record_admission_capacity(
        &self,
        output: Option<&EmergencyDiagnosticHandle>,
        context: AdmissionEventContext<'_>,
        facts: &AdmissionCapacityFacts,
        construction: AdmissionConstruction,
    ) -> AdmissionCapacitySubmission {
        let snapshot = facts.snapshot;
        let dimensions = [
            (CapacityDimension::Cpu, "cpu", snapshot.cpu_budget()),
            (
                CapacityDimension::Memory,
                "memory",
                snapshot.memory_budget_bytes(),
            ),
            (
                CapacityDimension::Database,
                "database",
                snapshot.database_budget(),
            ),
            (
                CapacityDimension::ProfuseContract,
                "profuse_contract",
                snapshot.profusecontract_budget(),
            ),
        ];
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis();
        let rejected = facts.decision == ProfuseGwCapacityDecision::CapacityRejected;
        let submissions = dimensions.map(|(dimension, name, budget)| {
            self.inner.metrics.capacity(
                Some(dimension),
                u64::try_from(facts.used).unwrap_or(u64::MAX),
                rejected,
            );
            let Some(output) = output else {
                return DiagnosticSubmission::OutputUnavailable;
            };
            output.submit_fixed_record(&CapacityRecord {
                schema_version: 1,
                timestamp_unix_ms: timestamp,
                level: "info",
                event: "framework.capacity",
                stage: "admission",
                context,
                capacity_dimension: name,
                budget,
                limit: snapshot.active_limit(),
                used: facts.used,
                bottleneck: match snapshot.bottleneck() {
                    ProfuseGwCapacityBottleneck::Cpu => "cpu",
                    ProfuseGwCapacityBottleneck::Memory => "memory",
                    ProfuseGwCapacityBottleneck::Database => "database",
                    ProfuseGwCapacityBottleneck::ProfuseContract => "profuse_contract",
                },
                decision: if rejected {
                    "capacity_rejected"
                } else {
                    "accepted"
                },
                outcome: if rejected { "rejected" } else { "accepted" },
                reject_reason: facts.reason.map(|reason| match reason {
                    ProfuseGwCapacityRejectReason::AtLimit => "at_limit",
                }),
                elapsed_ms: facts.elapsed_ms,
                construction,
            })
        });
        let [cpu, memory, database, profuse_contract] = submissions;
        AdmissionCapacitySubmission {
            cpu,
            memory,
            database,
            profuse_contract,
        }
    }
}

/// Source temporaries only; fixed frame/queued packet are reported by the
/// existing request_logging_layouts. Neither context form adds an allocation.
pub fn admission_capacity_layouts() -> [std::alloc::Layout; 4] {
    [
        std::alloc::Layout::new::<AdmissionCapacityFacts>(),
        std::alloc::Layout::new::<AdmissionEventContext<'static>>(),
        std::alloc::Layout::new::<CapacityRecord<'static>>(),
        std::alloc::Layout::new::<AdmissionCapacitySubmission>(),
    ]
}