Server Architecture

Server Architecture (chronixd)

The chronixd crate provides a standalone server daemon that wraps the embedded Chronix database with multi-protocol network access.

Server Components

┌───────────────────────────── chronixd ──────────────────────────────┐
│                                                                     │
│  ┌──────────┐  ┌──────────┐  ┌─────────────┐                       │
│  │ HTTP/REST │  │   gRPC   │  │ Flight SQL  │                       │
│  │  (axum)   │  │ (tonic)  │  │(arrow-flight│                       │
│  │  :8086    │  │  :8087   │  │  :8817      │                       │
│  └─────┬─────┘  └─────┬────┘  └──────┬──────┘                      │
│        │               │              │                              │
│        └───────────────┼──────────────┘                              │
│                        │                                             │
│               ┌────────▼────────┐                                    │
│               │  Arc<Chronix>   │  (shared embedded database)        │
│               └────────┬────────┘                                    │
│                        │                                             │
│  ┌─────────────────────┼────────────────────────┐                    │
│  │ chronix maintenance │ Prometheus Metrics      │                   │
│  │ thread (built in)   │ (metrics-exporter-prom) │                   │
│  └─────────────────────┴────────────────────────┘                    │
│                                                                     │
│  ┌──────────────── ConnectorManager ────────────────┐               │
│  │ ┌──────────────┐  ┌───────────────┐              │               │
│  │ │ KafkaConsumer │  │ MqttSubscriber│  (feature-   │               │
│  │ │ (krafka)     │  │ (rumqttc)     │   gated)     │               │
│  │ └──────────────┘  └───────────────┘              │               │
│  └──────────────────────────────────────────────────┘               │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Key Design Decisions

• Shared Arc<Chronix> — All three servers share a single database instance. Writes via any protocol are immediately visible from all others.
• spawn_blocking — All Chronix methods are synchronous. The server wraps them with tokio::task::spawn_blocking() to avoid blocking the async runtime.
• Signal-based shutdown — SIGTERM/SIGINT triggers graceful shutdown: stop accepting connections, drain in-flight requests, flush memtables, close WAL, and exit. Server errors (both I/O and task panics) are logged.
• TLS optional — When a TLS section is present in the TOML config, all three servers use rustls for encryption. Otherwise they run in plaintext.
• Zero unsafe code — Connectors use an OnceLock<Weak<Self>> pattern for safe self-referencing instead of raw pointer reconstruction. new_arc() creates the Arc and stores a Weak self-reference; start() upgrades through Weak::upgrade() with a clear error if called outside an Arc.
• Panic-free production paths — now_nanos() and setup_prometheus() return Result instead of panicking. All error paths propagate cleanly.

Flight SQL DoPut Pipeline

DoPut enables high-throughput bulk writes via Arrow RecordBatches. The implementation follows this pipeline:

1. Schema extraction — Schema::try_from(&FlightData) decodes the Arrow schema from the first FlightData message in the stream
2. Batch decoding — Subsequent messages are decoded via arrow_flight::utils::flight_data_to_arrow_batch() using the extracted schema and an empty dictionary map
3. Column classification — Utf8 and dictionary-encoded string columns (Dictionary<Int32, Utf8>) are classified as tags (unless column metadata contains role=field), numeric/boolean columns as fields
4. Point conversion — arrow_batch_to_points() uses pre-downcast columnar access via TypedField and TagCol enums, performing O(columns) type downcasts instead of O(rows × columns). Tags support both plain StringArray and DictionaryArray<Int32, Utf8>. Fields are extracted with full type support (F64, I64, U64, Bool, String, Decimal128). Null values are skipped gracefully; a Decimal128 value outside what Chronix stores is an error rather than a skipped field
5. Database write — Converted points are written via db.insert_batch() inside spawn_blocking for non-blocking async execution

The measurement name is derived from CommandStatementUpdate.query.

Ingestion Connectors

chronixd includes built-in ingestion connectors for consuming time-series data from external streaming systems. Connectors are feature-gated — the real consumer/subscriber implementations are compiled only when the corresponding Cargo feature flag is enabled.

Feature Flags

rumqttc is taken with default-features = false. Its use-rustls feature is not a TLS option the subscriber offers — it never builds a Transport — so enabling it only added rustls-webpki 0.102 (two advisories, and rumqttc's own ^0.102.8 caps it below the fixed 0.103.10), the unmaintained rustls-pemfile, and a second crypto provider. MQTT connections are plaintext; TLS returns when rumqttc moves to a provider-agnostic rustls.

Without the feature flag, the connector struct is still available (for configuration parsing and testing), but start() logs a notice and returns immediately. The #[cfg_attr(not(feature = "..."), allow(dead_code))] pattern suppresses unused-field warnings for feature-gated fields.

Connector Trait

#[async_trait]
pub trait IngestionConnector: Send + Sync {
    fn name(&self) -> &str;
    fn connector_type(&self) -> &str;
    async fn start(&self) -> Result<(), ServerError>;
    async fn stop(&self) -> Result<(), ServerError>;
    async fn status(&self) -> ConnectorStatus;
    async fn metrics(&self) -> ConnectorMetrics;
    async fn is_healthy(&self) -> bool;
}

status() reports what the connector's task is doing: Running (receiving), Idle (connected, nothing arriving), Reconnecting (retrying with backoff), Failed(reason) (retrying, and naming the error), Stopped. is_healthy() is Running | Idle. Connection setup retries rather than being attempted once, so a broker that is still coming up is not a permanent failure.

Connector Format (RQ-05)

The payload format is configured via the ConnectorFormat enum:

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConnectorFormat {
    LineProtocol,
    Json,
}

Invalid format values are rejected at deserialization time (serde) rather than at runtime. Match statements in the consumer/subscriber are exhaustive — no fallthrough "unsupported format" branches.

Topic-to-Measurement Mapping (ENT-06)

Both Kafka and MQTT connectors support explicit topic_measurement_map for controlling which topic maps to which measurement name:

[mqtt]
topic_measurement_map = { "sensors/floor1/temp" = "temperature", "sensors/floor2/temp" = "temperature" }

When a topic is not in the map, the default behavior applies:

• Kafka — uses the topic name directly
• MQTT — uses the last segment of the topic path (e.g., sensors/room1/temperature → temperature)

This prevents measurement collisions when topics with different paths share the same last segment.

Kafka Consumer (kafka feature)

When compiled with --features kafka, KafkaConsumer spawns a krafka Consumer loop:

1. Builds a Consumer with brokers, group ID, offset reset policy, and the authentication implied by security_protocol / sasl_mechanism
2. Subscribes to configured topics
3. Polls a batch per second in a Tokio task
4. Parses payloads (JSON via util::parse_json_point(), or InfluxDB Line Protocol via influx::parse_line_protocol())
5. Writes the whole batch with one insert_batch()
6. Commits offsets only after the write succeeds — a failed write leaves the batch uncommitted, so it is redelivered rather than lost (at-least-once)

krafka rather than rdkafka: rdkafka builds librdkafka from C via cmake, which imposes a C toolchain on every build — including the aarch64 cross-compile — and adds an FFI surface to a tree that denies unsafe_code.

Transport security

security_protocol accepts PLAINTEXT, SSL, SASL_PLAINTEXT and SASL_SSL; sasl_mechanism accepts PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512. Credentials come from credential_file (preferred, for rotation) or from inline sasl_username/sasl_password. ca_cert points at the PEM bundle for a private CA; without it the bundled WebPKI roots apply, which is what a managed broker needs. An unrecognised value for either field is rejected by name rather than ignored.

[kafka]
brokers = "broker:9093"
group_id = "chronix-ingest"
topics = ["metrics"]
security_protocol = "SASL_SSL"
sasl_mechanism = "SCRAM-SHA-512"
credential_file = "/run/secrets/kafka.json"
ca_cert = "/etc/ssl/private-ca.pem"

7. On stop, sets the running flag and yields to allow the consumer task to drain

MQTT Subscriber (mqtt feature)

When compiled with --features mqtt, MqttSubscriber spawns a rumqttc AsyncClient loop:

1. Creates MqttOptions with broker, port, client ID, and optional TLS
2. Subscribes to configured topics with the specified QoS level
3. Handles Event::Incoming(Packet::Publish) events
4. Extracts topic-level tags from the MQTT topic path
5. Parses payloads via util::parse_json_point() with extra tags
6. Writes to the database with insert_batch()
7. Reconnects automatically with 5-second backoff on connection errors
8. Tracks reconnection count in metrics

Connector Manager

ConnectorManager orchestrates connector lifecycle:

• register() / start_all() / stop_all() — basic lifecycle
• reload(config) — diff-and-reconcile: builds the desired connector set from the new config, stops removed connectors, registers + starts new ones, preserves unchanged running connectors
• list_connectors() — returns ConnectorInfo (name, type, status, metrics) for the REST status API at GET /api/v1/connectors
• refresh_gauges() — publishes chronix_connector_up{connector,type}, called from the metrics scrape. Connector health is not part of /ready, which asks only whether the database is writable: a connector that cannot reach its broker does not stop this node answering queries. Alert on the gauge

Connectors are wired into server.rs: on startup, KafkaConsumer and MqttSubscriber instances are created from ServerConfig.kafka / ServerConfig.mqtt, registered with the manager, and started. On shutdown, stop_all() drains in-flight data before the database is closed.

Shared Utilities (util.rs)

Common parsing and conversion utilities shared across HTTP, Kafka, and MQTT:

• parse_json_point(measurement, text, extra_tags) — Parses a JSON payload into a Point. Extracts tags, fields, and optional timestamp keys. Supports optional extra_tags for MQTT topic-derived tags.
• json_value_to_field(key, value) — Converts a serde_json::Value to a FieldValue. Integers are preserved as I64 (not coerced to F64).
• now_nanos() — Current wall clock as nanosecond Unix epoch. Returns Result to gracefully handle pre-epoch clocks and i64 overflow (year 2262+).
• column_type_to_str() / column_type_to_arrow() — Schema introspection helpers for gRPC and Flight SQL responses.

Result encoding (wire::value)

One Arrow → JSON encoding, shared by every surface that answers a query in JSON or protobuf. Its match over DataType has no catch-all arm, so an Arrow release that adds a variant fails the build.

• to_json(col, idx) — a single cell, per the table in API reference.
• gRPC keeps native proto scalars where one exists — double carries NaN, which JSON cannot — and routes the rest through to_json into SqlValue.json, distinct from SqlValue.string.
• Arrow Flight SQL streams the RecordBatch unmodified, preserving the Arrow types themselves.

chronix_query::arrow_cell_to_field_value is the inverse, used by the region-snapshot and distributed-read paths: total over the six storage column types, and an error for any other Arrow type.

Server Metrics

Wire Protocol Compatibility

Prometheus Remote Write/Read

• POST /api/v1/prom/write — Accepts Prometheus remote write protocol (protobuf + snappy). Maps __name__ → measurement, labels → tags, samples → fields.
• POST /api/v1/prom/read — Serves Prometheus remote read requests. Translates ReadRequest matchers to Chronix tag filters.
• GET /metrics — Prometheus text exposition format. The built-in parser supports all four metric types: gauge, counter, histogram (bucket aggregation with le bounds, _sum, _count), and summary (quantile values with _sum, _count). # TYPE directives route lines to type-specific handlers. Histogram buckets are auto-sorted by bound.

OpenTelemetry OTLP

• POST /v1/metrics — the path the OpenTelemetry Collector's otlphttp exporter derives from its endpoint. Accepts OTLP metrics (protobuf and JSON), gzipped or not. POST /api/v1/otlp/metrics is an alias. Maps gauge, sum, histogram, summary → Chronix measurements + fields. Resource attributes → tags.

Metrics / Observability

Chronix uses the metrics facade crate (v0.24) to emit runtime telemetry. The library emits metrics via counter!, gauge!, and histogram! macros — the embedding application is responsible for installing a recorder (e.g., metrics-exporter-prometheus).

Tracing instrumentation – All forecast fit()/predict() and anomaly fit()/detect() operations are annotated with #[tracing::instrument] spans at debug level, enabling timing analysis via any tracing-subscriber backend.

Distributed Tracing (chronixd::otel)

OpenTelemetry Integration

Feature-gated otlp (default) enables OTLP trace export:

TracingConfig
  ├── service_name
  ├── log_format: Pretty | Json | Compact
  ├── log_filter: EnvFilter string
  └── otlp: Option<OtlpConfig>
        ├── endpoint (default: http://localhost:4317)
        ├── sampling: AlwaysOn | AlwaysOff | Ratio(f64)
        └── always_sample_errors

W3C Trace Context Propagation

• extract_trace_context() — reads traceparent/tracestate from gRPC metadata
• inject_trace_context() — writes trace context into outgoing gRPC metadata
• make_trace_interceptor() — tonic interceptor that injects trace context into outgoing requests
  • With otlp feature: uses OTel span context (includes current span ID)
  • Without otlp feature: uses task-local traceparent (set via with_trace_context())
• with_trace_context(parent, state, future) — scopes a task-local W3C trace context, enabling trace propagation in downstream gRPC calls even without the OTLP exporter

TracingGuard

RAII guard stores SdkTracerProvider; on drop, calls provider.shutdown() for clean OTLP flush.