How to Define Schemas¶

Schemas provide type safety, documentation, and validation for your trace events.

Dynamic vs Structured¶

Dynamic Events (Python only)¶

Python allows logging without pre-defined schemas:

import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("dynamic")

# Types are inferred from values
source.log("sensor", {
    "temperature": 25.5,    # → Float64
    "pressure": 1013,       # → Int64
    "status": "OK",         # → String
    "valid": True          # → Boolean
})

Use for: Quick prototyping, exploration Avoid for: Production code, APIs, shared data

Structured Events (recommended)¶

Define schemas for type safety and documentation:

PythonRustGo

import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("structured")

# Define schema with types and units
sensor_event = source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "°C"),
    zelos_sdk.TraceEventFieldMetadata("pressure", zelos_sdk.DataType.Int32, "Pa"),
    zelos_sdk.TraceEventFieldMetadata("status", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("valid", zelos_sdk.DataType.Boolean)
])

# Type-safe logging
sensor_event.log(
    temperature=25.5,
    pressure=1013,
    status="OK",
    valid=True
)

// Rust requires schema definition
let sensor_event = source
    .build_event("sensor")
    .add_f64_field("temperature", Some("°C".to_string()))
    .add_i32_field("pressure", Some("Pa".to_string()))
    .add_string_field("status", None)
    .add_bool_field("valid", None)
    .build()?;

// Type-safe logging
sensor_event.build()
    .try_insert_f64("temperature", 25.5)?
    .try_insert_i32("pressure", 1013)?
    .try_insert_string("status", "OK".to_string())?
    .try_insert_bool("valid", true)?
    .emit()?;

// Go requires schema definition
unitC := "°C"
unitPa := "Pa"
sensorEvent, _ := source.BuildEvent("sensor").
    AddFloat64Field("temperature", &unitC).
    AddInt32Field("pressure", &unitPa).
    AddStringField("status", nil).
    AddBooleanField("valid", nil).
    Build()

// Type-safe logging
builder, _ := sensorEvent.Build()
builder, _ = builder.TryInsertFloat64("temperature", 25.5)
builder, _ = builder.TryInsertInt32("pressure", 1013)
builder, _ = builder.TryInsertString("status", "OK")
builder, _ = builder.TryInsertBoolean("valid", true)
builder.Emit()

Well-Known Event Schemas¶

Use a well-known schema when an event has standard Zelos semantics. In addition to defining the fields, the schema stamps a stable event-type identifier into the trace. Zelos uses that identifier to recognize logs, spans, check results, and annotations without relying on your source or event names.

import zelos_sdk
from zelos_sdk import schemas

zelos_sdk.init()
source = zelos_sdk.TraceSource("web_service")
application_log = source.add_event("application", schemas.Log)

application_log.log(
    level="info",
    message="health check passed",
    name="web_service",
    file="server.py",
    line=42,
)

application is the event name chosen by this producer; zelos.log.v1 is the schema identity supplied by schemas.Log. The class exposes the identifier as EVENT_TYPE and its field list as FIELDS.

The agent recognizes these well-known event types, and built-in workflows produce them. A ? marks a nullable field:

Event type	Purpose and fields
`zelos.event.v1`	Point event: `name`, `status`, `data`
`zelos.span.v1`	Duration: `name`, `status`, `start_ns`, `end_ns`, `data`
`zelos.log.v1`	Log record: `level`, `message`, `name`, `file`, `line`
`zelos.check.result.v1`	Check result: `name`, `status`, `reason`, `op`, `temporal`, `lhs?`, `rhs?`, `message?`, `wall_ms`, `query_ms`, `count?`, `evidence_signal?`, `evidence_time_ns?`, `evidence_value?`, `range_start_ns?`, `range_end_ns?`, `duration_ns?`
`zelos.annotation.comment.v1`	Annotation: `name`, `status?`, `author?`, `text?`, `target_path?`, `range_start_ns?`, `range_end_ns?`, `tags?`

Use the check and annotation APIs to create their event types. If you implement another compatible producer, register the exact canonical fields and pass the matching event_type= to add_event().

Well-known schemas are versioned contracts:

The event name is application-defined and can differ between producers.
time_ns is implicit and is not part of the declared fields.
data and tags are strings. Encode structured content as JSON when needed.
Do not change the fields behind an existing event-type identifier. Use a new version such as acme.motor.state.v2 for an incompatible schema change.

You can give a custom schema the same stable identity behavior:

class MotorState:
    EVENT_TYPE = "acme.motor.state.v1"
    FIELDS = [
        zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.String),
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64, "rpm"),
    ]


motor_state = source.add_event("state", MotorState)
motor_state.log(state="running", rpm=3200.0)

Passing a class to add_event() reads its FIELDS and optional EVENT_TYPE class variables. A field-list schema remains untyped unless you pass event_type= explicitly.

Discover Typed Event Tables¶

List event tables by exact event type or by a trailing-* prefix:

zelos live events --event-type 'zelos.span.v1' --latest
zelos live events --event-type 'zelos.check.*'
zelos trace events capture.trz --event-type 'zelos.annotation.*'

zelos live events searches the live catalog with a 60-second lookback by default; change it with --lookback 5m. zelos trace events always searches the entire bounded trace and has no lookback option. Only a trailing * is a wildcard, so zelos.*.v1 does not match.

For the generic agent-tool endpoint, event_list has the same distinction: scope: {"kind": "live", "durationSeconds": 60} searches a live lookback, while scope: {"kind": "trace", "tracePaths": ["capture.trz"]} lists the whole trace.

Data Types¶

Choosing the Right Type¶

Type	Range	Use For	Memory
Int8	-128 to 127	State codes, small counters	1 byte
Int16	±32K	Sensor ADC values	2 bytes
Int32	±2.1B	Timestamps (seconds), IDs	4 bytes
Int64	±9.2×10¹⁸	Nanosecond timestamps	8 bytes
UInt8	0 to 255	Percentages, byte values	1 byte
UInt16	0 to 65K	Port numbers, ADC values	2 bytes
UInt32	0 to 4.2B	Counters, CAN IDs	4 bytes
UInt64	0 to 1.8×10¹⁹	Large counters, UUIDs	8 bytes
Float32	±3.4×10³⁸	Sensor data (7 digits)	4 bytes
Float64	±1.7×10³⁰⁸	Precise measurements (15 digits)	8 bytes
Boolean	true/false	Flags, states	1 byte
String	UTF-8 text	Text, identifiers	Variable
Binary	Raw bytes	Payloads, images	Variable
TimestampNs	Nanoseconds	Time points	8 bytes

Complete Example¶

PythonRustGo

# One field per data type
event = source.add_event("all_types", [
    # Choose smallest type that fits your data
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.UInt8),        # 0-255
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),  # ±0.1°C precision
    zelos_sdk.TraceEventFieldMetadata("timestamp", zelos_sdk.DataType.TimestampNs),
    zelos_sdk.TraceEventFieldMetadata("message", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("payload", zelos_sdk.DataType.Binary),
])

import time
event.log(
    state=3,
    temperature=25.5,
    timestamp=time.time_ns(),
    message="Sensor reading",
    payload=bytes([0xDE, 0xAD, 0xBE, 0xEF])
)

let event = source
    .build_event("all_types")
    .add_u8_field("state", None)
    .add_f32_field("temperature", Some("°C".to_string()))
    .add_timestamp_ns_field("timestamp", None)
    .add_string_field("message", None)
    .add_binary_field("payload", None)
    .build()?;

use zelos_trace::time::now_time_ns;
event.build()
    .try_insert_u8("state", 3)?
    .try_insert_f32("temperature", 25.5)?
    .try_insert_timestamp_ns("timestamp", now_time_ns())?
    .try_insert_string("message", "Sensor reading".to_string())?
    .try_insert_binary("payload", vec![0xDE, 0xAD, 0xBE, 0xEF])?
    .emit()?;

unitC := "°C"
event, _ := source.BuildEvent("all_types").
    AddUint8Field("state", nil).
    AddFloat32Field("temperature", &unitC).
    AddTimestampNsField("timestamp", nil).
    AddStringField("message", nil).
    AddBinaryField("payload", nil).
    Build()

builder, _ := event.Build()
builder, _ = builder.TryInsertUint8("state", 3)
builder, _ = builder.TryInsertFloat32("temperature", 25.5)
builder, _ = builder.TryInsertTimestampNs("timestamp", time.Now().UnixNano())
builder, _ = builder.TryInsertString("message", "Sensor reading")
builder, _ = builder.TryInsertBinary("payload", []byte{0xDE, 0xAD, 0xBE, 0xEF})
builder.Emit()

Units¶

Always specify units for physical quantities:

# Common units by domain
telemetry = source.add_event("telemetry", [
    # Mechanical
    zelos_sdk.TraceEventFieldMetadata("speed", zelos_sdk.DataType.Float64, "m/s"),
    zelos_sdk.TraceEventFieldMetadata("acceleration", zelos_sdk.DataType.Float64, "m/s²"),
    zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),

    # Electrical
    zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float32, "V"),
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float32, "A"),
    zelos_sdk.TraceEventFieldMetadata("power", zelos_sdk.DataType.Float32, "W"),

    # Thermal
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),

    # Dimensionless
    zelos_sdk.TraceEventFieldMetadata("efficiency", zelos_sdk.DataType.Float32, "%"),
    zelos_sdk.TraceEventFieldMetadata("gear", zelos_sdk.DataType.Int8),  # No unit
])

Enumerations¶

Map numeric values to readable strings using value tables:

PythonRustGo

from enum import IntEnum

class State(IntEnum):
    IDLE = 0
    RUNNING = 1
    ERROR = 2

# Define event
status = source.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.UInt8),
])

# Map values to strings for visualization
source.add_value_table("status", "state", {
    0: "IDLE",
    1: "RUNNING",
    2: "ERROR"
})

# Log using enum
status.log(state=State.RUNNING)

use zelos_trace_types::Value;

#[repr(u8)]
enum State {
    Idle = 0,
    Running = 1,
    Error = 2,
}

// Define event
let status = source
    .build_event("status")
    .add_u8_field("state", None)
    .build()?;

// Add value table
source.add_value_table("status", "state", vec![
    (Value::UInt8(0), "IDLE".to_string()),
    (Value::UInt8(1), "RUNNING".to_string()),
    (Value::UInt8(2), "ERROR".to_string()),
].into_iter())?;

// Log using enum
status.build()
    .try_insert_u8("state", State::Running as u8)?
    .emit()?;

const (
    StateIdle = iota
    StateRunning
    StateError
)

// Define event
status, _ := source.BuildEvent("status").
    AddUint8Field("state", nil).
    Build()

// Add value table
source.AddValueTable("status", "state", map[*zelos.Value]string{
    zelos.NewUint8Value(0): "IDLE",
    zelos.NewUint8Value(1): "RUNNING",
    zelos.NewUint8Value(2): "ERROR",
})

// Log using enum
builder, _ := status.Build()
builder, _ = builder.TryInsertUint8("state", StateRunning)
builder.Emit()

Nested Events¶

Organize related events using path notation:

PythonRustGo

# Parent event
source.add_event("motor", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64, "rpm"),
])

# Child events use "/" separator
source.add_event("motor/thermal", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),
])

# Access via attributes (Python only)
source.motor.log(rpm=3500)
source.motor.thermal.log(temperature=85.5)

# Or via path
source.log("motor", {"rpm": 3500})
source.log("motor/thermal", {"temperature": 85.5})

// Use naming convention with "/"
let motor = source.build_event("motor")
    .add_f64_field("rpm", Some("rpm".to_string()))
    .build()?;

let motor_thermal = source.build_event("motor/thermal")
    .add_f32_field("temperature", Some("°C".to_string()))
    .build()?;

// Use naming convention with "/"
unitRpm := "rpm"
motor, _ := source.BuildEvent("motor").
    AddFloat64Field("rpm", &unitRpm).
    Build()

unitC2 := "°C"
motorThermal, _ := source.BuildEvent("motor/thermal").
    AddFloat32Field("temperature", &unitC2).
    Build()

Best Practices¶

1. Define Once¶

# Good
event = source.add_event("data", schema)
for _ in range(1000000):
    event.log(value=read_value())

# Bad - Redefining schema
for _ in range(1000000):
    source.add_event("data", schema)  # Wasteful!

2. Choose Minimal Types¶

# Use smallest type that fits
field("percentage", UInt8)     # 0-100 fits in UInt8
field("temperature", Float32)  # 0.1°C precision is enough

# Don't oversize
field("percentage", Float64)   # Wasteful for 0-100

# Single event preserves correlation
source.add_event("position", [
    field("x", Float64, "m"),
    field("y", Float64, "m"),
    field("z", Float64, "m"),
])

# Separate events lose correlation
source.add_event("pos_x", [field("value", Float64, "m")])
source.add_event("pos_y", [field("value", Float64, "m")])

How to Define Schemas¶

Dynamic vs Structured¶

Dynamic Events (Python only)¶

Structured Events (recommended)¶

Well-Known Event Schemas¶

Discover Typed Event Tables¶

Data Types¶

Choosing the Right Type¶

Complete Example¶

Units¶

Enumerations¶

Nested Events¶

Best Practices¶

1. Define Once¶

2. Choose Minimal Types¶

3. Group Related Fields¶