Architecture

Rows arrive in four ways:

• Kafka. A consumer reads Avro topics through a schema registry and creates the table from the schema.
• Prometheus. Prometheus, Grafana Alloy or vmagent push samples to POST /api/v1/write.
• The agent. muscles agent runs next to a data source, tails log files and the systemd journal, keeps batches on disk and resends them until the server acknowledges them.
• Inserts. A client sends INSERT over the HTTP API or the ClickHouse HTTP interface, as SQL values or as JSONEachRow, CSV, TabSeparated, RowBinary or Native rows.

Results return over the HTTP API as JSON, Arrow IPC or any ClickHouse output format, and over the ClickHouse HTTP interface, which the ClickHouse drivers and the Grafana data source use. Polars and a JDBC driver connect as well.

Tables elsewhere take part in the same SQL. CREATE EXTERNAL TABLE declares a table in PostgreSQL, MariaDB, ClickHouse or Redis, or an Iceberg table on S3, and a SELECT reads it in place. An external table joins other external tables and stored tables, so an Iceberg table in the lake joins a PostgreSQL table in one statement.

IMPORT PARQUET and IMPORT ICEBERG load files into a table in one crash-safe statement, SELECT ... INTO OUTFILE writes Parquet, and muscles dump writes a Kafka topic to an Iceberg table on S3.

Each table has its own write-ahead log: append-only segments of records, each with its length, a sync stamp and a CRC32C checksum. Every ingest path ends in the same function, which checks the rows against the table schema, appends them to the log and returns after the log is synced. A Kafka offset is committed after the sync, and an HTTP insert gets its 200 after it. An insert token lets a client retry a batch without storing it twice, also across a restart.

After a crash, replay truncates a half-written record at the end of the log. A record that was acknowledged and is damaged is reported as corruption, and replay stops there. When an fsync fails, the log is marked failed and the table rejects writes until every row it holds is in a part: Linux reports a write-back error once, so a later fsync could pass with the data still unwritten.

After the log, a row goes into the table's memtable, one array per column, which queries read through a snapshot. A background flusher seals the memtable at a row count or an interval and writes it as a part: a read-only directory with one compressed file per column in granules of up to 8192 rows, a marks file with the position, checksum and first key of each granule, and a manifest with a checksum.

The part is written under a temporary name, every file is synced, and one atomic rename publishes it. Indexes are side files that a reader may ignore, so a damaged index costs speed and never a wrong row. A DELETE writes a deletion vector next to the part.

Background merges combine small parts into larger ones. That is where a ReplacingMergeTree table keeps one row per key, an aggregating view adds up the rows of a group, and rows past a table TTL are dropped. A merge never changes a published part: it writes a new one, swaps it in and reclaims the old ones after the last reader has finished. With tiering on, a background task uploads parts older than a set age to S3 as one object and leaves a small stub behind, so a cold read fetches only the columns and granules it needs.

The parser accepts a subset of ClickHouse SQL. The planner turns a SELECT into a scan: the columns to read, the key range the WHERE clause implies, the predicates per row and the aggregates per group. Joins are ordered smallest table first. A GROUP BY on a time bucket over a table with a rollup reads the coarsest rung that fits, adds the newest raw rows, and returns the same answer as a scan of the raw table.

A scan opens a snapshot of the parts and the memtable. Before it decompresses anything, it drops whole parts whose key range cannot match, then granules by their first key, then whatever the skip indexes rule out: minmax ranges, set lists, bloom filters, and the text and vector indexes. The granules that remain are decompressed one at a time, and the filter and aggregate kernels run over whole column arrays rather than row by row.

A filtered SELECT or a GROUP BY over several parts runs on a pool of worker threads, each over its own range of parts or granules. Partial aggregates are merged with the same algebra that merges results across shards. Every query runs under a governor with limits on time, rows and memory, and a GROUP BY or ORDER BY that passes its memory budget spills to temporary files. An external table is read by a connector that streams row batches through the same kernels, and a join reads each external side into memory before it runs.

Every node runs the same binary, and MUSCLES_NODE_MODE picks its role. all, the default, runs everything on one node. data owns one shard: it ingests, stores, merges and tiers, and it serves queries for that shard. query is a coordinator without a disk: it plans, sends a fragment of the query to every data node and merges their partial results. ingest is a worker without a disk: it consumes Kafka, computes the shard of each record from its key, forwards the record to the owner, and advances its offset only after the owner has acknowledged the write.

A shard map names the data nodes. Keyed Kafka ingest and INSERT ... SELECT place each row on the shard of its key, and a plain SQL insert stays on the node that receives it. A join across shards needs the shard key in its ON clause, or a small table that the coordinator broadcasts.

Replication is off by default. When it is on, a primary ships every published part to an S3 bucket and writes a signed commit record to a Kafka topic. A replica reads the topic, downloads the parts and installs them, and builds its catalog from a signed schema snapshot in the bucket. A replica refuses client writes. To fail over, an administrator promotes a replica by giving it a higher generation number, and an old primary that sees the higher generation stops shipping parts.

Settings come from MUSCLES_* environment variables, or from a muscles.yaml that the variables override. The data lives in two directories, one for logs and one for parts.

The container image is a static binary on a distroless base with a read-only root file system and no shell. The Helm chart deploys a StatefulSet per shard, a Deployment of query nodes behind a Service, health probes, a pod disruption budget per shard and a Prometheus ServiceMonitor.

The operator manages the same shapes from MusclesCluster, MusclesIngestCluster, MusclesTable and MusclesUser resources, and scales ingest workers on Kafka consumer lag. Before a rolling upgrade it can run the new image once and report any difference in the stored formats.

• Iceberg to muscles and muscles to Iceberg. An export that writes Parquet files and Iceberg metadata from a stopped data directory or a snapshot, and the import for the other direction.
• ClickHouse to muscles and muscles to ClickHouse. Table definitions and data through Native format files, keeping the sort key and the codecs where both systems have them.
• A standalone reader. A program that reads the published part and log formats without the server.