Crate spectra

Expand description

Spectra is a Rust observability library for typed metrics and counters, structured event logs, and pluggable storage.

Wire a backend once with Spectra::builder(), declare schemas with spectra_metric! and spectra_schema!, then emit and query through the typed helpers those macros expand. Storage engines stay behind thin async ports — swap mem, sqlite, ClickHouse, or TensorBase without changing emit code.

§Features

Typed schema and metric DSL — declare counters and event tables with classification metadata (PII, console safety) via spectra_metric! and spectra_schema!. Each macro expands inventory registration, typed *Recorder / *Logger helpers, and transport *Payload / *_TOPIC DTOs.
Schema DSL controls — set emit defaults on the schema: level, default_sample_rate, coalesce_ms (gauges), and field classification (pii, safe_for_console). See spectra_schema! and spectra_metric!. Runtime env/TOML overrides live on spectra_core::SpectraConfig.
Composable storage — inject backends on the builder: in-memory (mem), durable embedded (sqlite), or remote (clickhouse, tensorbase).
Emit controls — global and per-name level gates, sampling, optional request/job buffers, and async batched persist (overrides schema defaults).
Query API — read metrics and events through SpectraRouter with label and time-range filters.
Direct or distributed wiring — one process can write storage, or many publishers can fan out to a consumer process that owns the database (see Mode 2).

Declarative schemas and typed logging APIs — same surface from embedded to multi-service.

§Getting started

You always emit the same way (CacheHitsRecorder::record(...), etc.). What changes is which process writes the database.

§Choose how data reaches storage

Mode 1 — Direct persist — one binary emits and stores. Start here.
Mode 2 — Distributed — many app processes publish onto a bus; a separate consumer binary writes storage.
Mode 3 — Dual-path — one process both publishes and stores (mirroring). Optional.

After you pick a mode, continue with schemas (shared by every mode).

§Mode 1 — Direct persist (one binary)

This process emits metrics/events and writes them to storage. There is no second binary and no message bus.

Your app ──emit──► Spectra ──async persist──► mem / SQLite / ClickHouse / TensorBase

Backend	Types	Feature	When to use
In-memory	`MemMetricsBackend` / `MemEventsBackend`	`mem` (default)	Local experiments
SQLite	`SqliteMetricsBackend` / `SqliteEventsBackend`	`sqlite`	Durable single host
ClickHouse	`ClickHouseMetricsBackend` / `ClickHouseEventsBackend`	`clickhouse`	Remote analytics
TensorBase	`TensorBaseMetricsBackend` / `TensorBaseEventsBackend`	`tensorbase`	ClickHouse-compatible scale-out

In-memory (both backends are required):

use std::sync::Arc;
use spectra::{MemEventsBackend, MemMetricsBackend, Spectra};

let spectra = Spectra::builder()
    .metrics_backend(Arc::new(MemMetricsBackend::new()))
    .events_backend(Arc::new(MemEventsBackend::new()))
    .embedded()
    .build()?;

High-throughput DW ingest — raise L2 batch size on the builder (not env vars):

use std::sync::Arc;
use std::time::Duration;
use spectra::{MemEventsBackend, MemMetricsBackend, PersistConfig, Spectra};

let spectra = Spectra::builder()
    .metrics_backend(Arc::new(MemMetricsBackend::new()))
    .events_backend(Arc::new(MemEventsBackend::new()))
    .persist(PersistConfig {
        batch_max: 2048,
        batch_wait: Duration::from_millis(5),
        ..PersistConfig::default()
    })
    .build()?;
// Web / scripts: use try_record_*_now / helpers (enqueue L2).
// Scripts that must exit only after durable writes:
spectra.flush_persist().await?;

Canonical path for web and Write Now: *_now → L2 batched persist. Prefer that over request_scope, which drops undrained emits on panic or early exit.

ClickHouse (omit .embedded(); constructors are async):

use std::sync::Arc;
use spectra::{ClickHouseEventsBackend, ClickHouseMetricsBackend, Spectra};

let url = "http://127.0.0.1:8123"; // or SPECTRA_CLICKHOUSE_URL
let metrics = ClickHouseMetricsBackend::connect(url).await?;
let events = ClickHouseEventsBackend::connect(url).await?;
let spectra = Spectra::builder()
    .metrics_backend(Arc::new(metrics))
    .events_backend(Arc::new(events))
    .build()?;

Runnable: quickstart, quickstart_sqlite, quickstart_clickhouse_emit. Then jump to schemas.

§Mode 2 — Distributed publish and consume (two binaries)

Use this when many services emit telemetry but you do not want each of them to open ClickHouse (or SQLite) itself. Instead:

Each app process is a publisher — it emits into Spectra, which hands the emit to your SpectraSink, which publishes onto a message bus.
A separate consumer process (or fleet) subscribes on that bus and writes storage.

Publisher binary(ies) ──emit──► SpectraSink ──publish──► bus (e.g. Photon)
Consumer binary       ──subscribe──► decode ──► try_*_now / storage backends

Spectra does not ship a bus. Your host owns that piece. A common choice is Photon (uf-photon on crates.io).

§What you create

Piece	Purpose
Shared schema crate	Same `spectra_*!` modules (`mod`-linked) on both sides
Publisher binary	`[[bin]]` or crate that emits; no DB writes
Consumer binary	Another `[[bin]]` or crate that owns storage
Bus	Photon, NATS, Kafka, … — host-provided

§Shared setup (both binaries)

Put spectra_schema! / spectra_metric! declarations in a shared crate both binaries depend on. mod each schema file so macros expand helpers, topic DTOs, and inventory registration (see § 4).

Helpers — typed emit API (publisher uses these)
Topic payloads — *Payload / *_TOPIC beside each schema (publisher publishes these)
Consumer persist — after decode, call try_record_counter_at / try_log_event_at with the envelope timestamp (or a storage backend)

§Publisher binary

Add a second binary in your workspace (for example src/bin/telemetry_publisher.rs, or a dedicated crate). In that process:

Implement SpectraSink: for each emit, build a topic *Payload and publish it on your bus. Keep the methods non-blocking (spawn a task, enqueue, or use Photon buffering).
Wire Spectra with .sink(...).persist_disabled().build() so this process does not write the analytics database. (The builder still requires dummy or unused backends.)
Emit with typed helpers exactly as in Mode 1.

use std::sync::Arc;
use spectra::{
    MemEventsBackend, MemMetricsBackend, Spectra, SpectraSink,
};

/// Your bus adapter — replace the body with Photon (or another) publish calls.
struct BusPublishSink;

impl SpectraSink for BusPublishSink {
    fn record_counter(&self, name: &str, labels: &[(&str, &str)], delta: i64) {
        // 1) Map name → schema module *Payload (see topics beside each schema).
        // 2) Publish asynchronously — do not block the emit thread.
        let _ = (name, labels, delta);
    }
    fn record_gauge(&self, name: &str, labels: &[(&str, &str)], value: f64) {
        let _ = (name, labels, value);
    }
    fn log_event(&self, table: &str, fields: &serde_json::Value) {
        let _ = (table, fields);
    }
}

let sink = Arc::new(BusPublishSink);
let _spectra = Spectra::builder()
    .metrics_backend(Arc::new(MemMetricsBackend::new()))
    .events_backend(Arc::new(MemEventsBackend::new()))
    .sink(sink as Arc<dyn SpectraSink>)
    .persist_disabled() // publisher does not open ClickHouse
    .build()?;

// Same emit API as Mode 1:
// CacheHitsRecorder::record(1, serde_json::json!({"region":"us"}));

Runnable sketch (in-memory stand-in for the bus): quickstart_publish_only. API detail: SpectraSink, SpectraBuilder::sink, SpectraBuilder::persist_disabled.

§Consumer binary

Create a different binary (for example src/bin/telemetry_consumer.rs). This process owns the database:

Build Spectra with real backends and persist left on (Mode 1-style .build()).
Start your bus subscriber (Photon start_executor / #[subscribe], etc.).
On each message: deserialize to a *Payload or MetricEmit / SpectraEvent, then call try_record_counter_at / try_log_event_at with the envelope ts (or write the storage backend directly).

use std::sync::Arc;
use spectra::{
    try_record_counter_at, ClickHouseEventsBackend, ClickHouseMetricsBackend, Spectra,
};

let url = std::env::var("SPECTRA_CLICKHOUSE_URL")?;
let spectra = Spectra::builder()
    .metrics_backend(Arc::new(ClickHouseMetricsBackend::connect(&url).await?))
    .events_backend(Arc::new(ClickHouseEventsBackend::connect(&url).await?))
    .build()?; // persist ON — this process writes storage

// Pseudocode for your bus callback (Photon subscriber, etc.):
// let emit = decode_metric_emit(bytes)?;
// let labels = label_pairs_from_json(&emit.labels);
// let ts = emit.ts.unwrap_or_else(chrono::Utc::now);
// try_record_counter_at(&emit.name, &labels, emit.delta.unwrap_or(1), ts);

let _ = spectra; // keep handle alive; run subscriber loop until shutdown

Runnable sketch (decode → try_record_counter_at without a live broker): quickstart_consume_forward.

§Run both

Start the consumer (and your bus / Photon) so subscriptions are ready.
Start one or more publishers.
Query storage from the consumer process (or any process that shares the same DB).

For Photon you typically set PHOTON_TRANSPORT_KEY (base64 of 32 bytes) and a broker URL such as PHOTON_NATS_URL. See Photon’s own README for adapter wiring.

§Mode 3 — Dual-path (optional)

Same process both publishes through a SpectraSink and persists to storage. Use when you want a bus mirror without moving writes to another binary.

Wire with .sink(transport).build() — omit persist_disabled so storage still receives emits.

Runnable: quickstart_transport.

§4. Declare and link schemas

Typed emit helpers expand from the same macros that register schema metadata. Required for every mode. No build.rs or OUT_DIR includes.

[dependencies]
spectra = { package = "uf-spectra", git = "https://github.com/unified-field-dev/spectra.git", tag = "v0.1.0", features = ["mem"] }
chrono = { version = "0.4", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

Typical application layout:

my-app/
├── Cargo.toml
└── src/
    ├── schemas/
    │   ├── mod.rs
    │   ├── cache_hits.rs
    │   └── request_log.rs
    └── main.rs

// src/schemas/mod.rs — one `mod` line per schema file (links inventory + helpers + topics)
pub mod cache_hits;
pub mod request_log;

pub use cache_hits::CacheHitsRecorder;
pub use request_log::RequestLogLogger;

§5. Declare metric and event schemas

Schemas are the typed contracts Spectra stores and queries. Share them across publisher and consumer binaries in Mode 2 (depend on the same schema crate).

Metric schema (spectra_metric!) — a named measurement family; optional level, default_sample_rate, and coalesce_ms (gauges).
Event schema (spectra_schema!) — a typed structured log table with classified columns; optional level and default_sample_rate (coalesce_ms is not valid on events).

Full field tables: spectra_metric! / spectra_schema!. Runtime overrides (env/TOML): spectra_core::SpectraConfig and the emit-gate table in the crate README.

// src/schemas/cache_hits.rs
use spectra::spectra_metric;

spectra_metric! {
    CacheHits {
        store: "default",       // logical store for routing
        name: "cache_hits",     // registry key + query_metrics.metric_name
        version: "0.1.0",
        description: "Counter for cache hit events",
        level: Debug,
        default_sample_rate: 0.1,
    }
}

// src/schemas/request_log.rs
use spectra::spectra_schema;

spectra_schema! {
    RequestLog {
        store: "default",
        table: "request_log",   // registry key + query_events.table
        version: "0.1.0",
        description: "Structured request debug events",
        level: Debug,
        default_sample_rate: 0.25,
        fields: [
            message: {
                r#type: String, // String | i64 | f64 | bool
                classification: { pii: false, safe_for_console: true },
            },
        ],
    }
}

Helper names come from the schema identifier: CacheHits → CacheHitsRecorder, RequestLog → RequestLogLogger. Each expansion also emits *Payload and *_TOPIC for Mode 2 transport.

§6. Emit metrics and events

Mode 1 / Mode 3: call helpers in the same process that owns (or mirrors) storage.

Mode 2: call helpers only in the publisher. The consumer persists with try_record_counter_at / try_log_event_at after decode (pass envelope ts).

use crate::schemas::{CacheHitsRecorder, RequestLogLogger};

CacheHitsRecorder::record(1, serde_json::json!({ "region": "us-west" }));
RequestLogLogger::log("request handled".to_string());

This crate’s CI demo helpers (helpers) come from platform_smoke_* schemas only — define product schemas in your application.

§7. Query persisted data

Storage persist is asynchronous. Wait briefly (or poll) before querying.

Mode 1 / Mode 3: query on the same process that wrote storage.

Mode 2: query from the consumer (or any process connected to the same database) — not from the publisher.

use spectra_core::{current_emit_ts, EventsQueryFilter, MetricsQueryRange};

tokio::time::sleep(std::time::Duration::from_millis(80)).await;
let now = current_emit_ts();

let points = spectra
    .router()
    .query_metrics(MetricsQueryRange {
        metric_name: "cache_hits".into(),
        start: now - chrono::Duration::seconds(5),
        end: now + chrono::Duration::seconds(1),
        label_matchers: vec![],
    })
    .await?;

let rows = spectra
    .router()
    .query_events(EventsQueryFilter {
        table: "request_log".into(),
        start: Some(now - chrono::Duration::seconds(5)),
        end: Some(now + chrono::Duration::seconds(1)),
        ..Default::default()
    })
    .await?;

Prefer row queries. Event chart aggregates (query_aggregate) are stubbed on most backends.

Configuration (emit gates, sampling, buffers) is documented on spectra_core::SpectraConfig.

§Notes

Enable backend features explicitly (mem is on by default; sqlite, clickhouse, and tensorbase are optional).
Product schemas belong in your application; this crate registers CI demo platform_smoke_* schemas only.
Pin consumption via git tag (for example v0.1.0). Configure storage on Spectra::builder(), not a global mode enum.
Event chart aggregates (spectra_core::EventStorageBackend::query_aggregate) are stubbed on most backends.
Schema modules must be mod-linked into the binary or inventory will not see them.
Schema macros and helpers call through the spectra facade — pin spectra as a direct dependency (no separate spectra-core pin required for the default recipe).

Re-exports§

pub use spectra_core;
pub use spectra_core::inventory;

Modules§

helpers: Typed emit helpers from this crate’s CI smoke schemas.
topics: Transport topic constants and payload DTOs from this crate’s CI smoke schemas.

Macros§

spectra_metric: Declare a metric family schema.
spectra_schema: Declare a structured event log schema.

Structs§

ChainedSink: Forwards each emit to every sink in the chain, synchronously and in registration order.
ClickHouseEventsBackend: Remote ClickHouse structured-event storage.
ClickHouseMetricsBackend: Remote ClickHouse metrics storage.
FieldClassification: Per-field GDPR-oriented metadata for Spectra event schemas.
MemEventsBackend: Non-durable in-memory structured-event storage keyed by logical table name.
MemMetricsBackend: Non-durable in-memory metrics storage.
MetricEmit: Metric emit envelope for transport publish wrappers.
PersistConfig: Queue and batch settings for crate::StoragePersistSink.
PersistHandle: Handle to wait until the persist queue has drained prior emits.
RecordingSink: Append-only in-memory sink for assertions in unit and integration tests.
SchemaFieldMetadata: Field metadata for registry / UI (macros populate this).
SchemaMetadata: Runtime metadata for a Spectra schema (event table or metric family).
SchemaMetadataInit: Function-pointer wrapper collected by inventory.
SchemaRegistry: Registry of metric and event schema metadata linked into the current binary.
Spectra: Handle to an installed process-wide Spectra runtime.
SpectraBuilder: Configures and installs Spectra with explicit storage adapter injection.
SpectraEvent: Structured event envelope for transport publish wrappers.
SpectraRouter: Resolves event tables and metric names to storage backends.
SqliteEventsBackend: Durable SQLite structured-event storage in the spectra_events table.
SqliteMetricsBackend: Durable SQLite metrics storage in the spectra_metrics table.
StoragePersistSink: SpectraSink that queues telemetry for asynchronous storage through a SpectraRouter.
TensorBaseEventsBackend: Scale-out TensorBase structured-event storage over its ClickHouse-compatible protocol.
TensorBaseMetricsBackend: Scale-out TensorBase metrics storage over its ClickHouse-compatible protocol.

Enums§

Error: Errors returned by Spectra storage, routing, serialization, and I/O paths.
LoggingKind: Event vs metric schema registration.
PersistOverflow: Behavior when the L2 persist queue is full.
SpectraLevel: Verbosity / sampling tier for a Spectra schema (more verbose = greater ordinal).

Traits§

SpectraSink: Synchronous fan-out target for metrics and structured events.

Functions§

try_log_event_at: Log a structured event immediately with an explicit emit timestamp.
try_log_event_now: Log a structured event immediately, bypassing any active emit buffer.
try_record_counter: Record a counter when a sink is configured; no-op if unset or re-entrant.
try_record_counter_at: Record a counter immediately with an explicit emit timestamp.
try_record_counter_now: Record a counter immediately, bypassing any active emit buffer.
try_record_gauge_at: Record a gauge immediately with an explicit emit timestamp.
try_record_gauge_now: Record a gauge immediately, bypassing any active emit buffer.

Type Aliases§

Result: Result type returned by Spectra operations.

Crate spectra

Crate spectra Copy item path

§Features

§Getting started

§Choose how data reaches storage

§Mode 1 — Direct persist (one binary)

§Mode 2 — Distributed publish and consume (two binaries)

§What you create

§Shared setup (both binaries)

§Publisher binary

§Consumer binary

§Run both

§Mode 3 — Dual-path (optional)

§4. Declare and link schemas

§5. Declare metric and event schemas

§6. Emit metrics and events

§7. Query persisted data

§Notes

Re-exports§

Modules§

Macros§

Structs§

Enums§

Traits§

Functions§

Type Aliases§

Crate spectra