Ingestion

Records enter an HD instance through the Message Bus Ingester: one engine role that reads a source, maps each record onto the dataset schema, groups records into batches and streams them to a Storage Cluster Controller over the FermiHDI HD Wire Protocol (FHDWP). Everything downstream — the fan-out to every Storage Node, the hash chain, the filter node — starts from what the ingester sends. This page is what an operator and a data owner need to know about it: where it reads from, what it accepts, and how it decides where one batch ends and the next begins.

Sources

One ingester reads one source, chosen by ingression_type:

Source	Setting	Reads
Kafka	`kafka.brokers`, `kafka.topic`, `kafka.group_id`	a topic, as a consumer group member
RabbitMQ	`rabbitmq.url`, `rabbitmq.queue`	a queue
MQTT	`mqtt.address`, `mqtt.topic`	a topic
Local directory	`file.directory`, `file.sort_by`	every file in the directory, one at a time, deleting each once its records are acknowledged
S3 or a compatible object store	`s3.bucket`, `s3.prefix`, `s3.region`, `s3.endpoint`	every object under the prefix, one at a time

Kafka is what the installer wires. The others are configured through the ingester's own settings, which reach it as environment variables, a JSON file or command-line flags in that order of precedence; the full table is in the configuration reference of the ingester.

A message from a broker is one record, in JSON or in the RowBinary form (input_format), or a CSV row. A file or object is many records, and its extension says how they are encoded.

File and object formats

Every format decodes to one JSON record per row and then takes the same path a broker message takes, so the schema mapping, address parsing, batch rotation and time compression exist once. The extension is the format:

Extension	Format	Notes
`.json`	one JSON object, or an array of them
`.jsonl`, `.ndjson`	one JSON object per line
`.csv`	a header row, then one record per row	cells that read as numbers become numbers
`.parquet`, `.pq`	Apache Parquet	a timestamp column is its integer in the column's own unit
`.orc`	Apache ORC
`.avro`	Avro object container file	the writer's schema travels in the file; a union is its branch, so a nullable `long` is `5` or `null`
`.pb`, `.protobuf`	Protocol Buffers	needs a descriptor set (`formats.protobuf_descriptor_file`, written by `protoc --descriptor_set_out --include_imports`) and the message type (`formats.protobuf_message_type`). Messages are length-delimited unless `formats.protobuf_delimited` is false. 64-bit integers stay numbers
`.xml`	XML	each child of the root is a record, or every element named `formats.xml_record_element` wherever it sits; attributes become `@name`, repeated children become arrays, text that reads as a number or boolean becomes one

Files are taken in sort_by order — by name, or by modification time — and a file with any other extension is left alone. A file is removed only after the storage cluster has acknowledged its records.

Mapping records onto the schema

The dataset schema names every field, its type and its byte position in the fixed-width record. A JSON key that matches a field name is written there; keys the schema does not name are ignored, and fields no key supplies are zero. Two types deserve a word:

Addresses. A field the schema types ipv4 is a 32-bit unsigned integer and one typed ipv6 a 128-bit one, both big-endian on the wire. The ingester accepts either the integer or the address written as text — 10.0.0.1, 2001:db8::1 — and stores the integer. An IPv4 address given for an ipv6 field is stored IPv4-mapped (::ffff:10.0.0.1). Every query service renders the integer back as an address; the SDK does neither and sees only the integer.
Time. A timestamp field is the integer it is, in the unit the dataset declares (time_index_interval_unit: seconds by default, or milliseconds, nanoseconds, minutes). It is not parsed from a date string.

How batches are formed

A batch is the unit the storage cluster stores, hashes, indexes and serves. It carries a 64-byte id, and the ingester decides the id of every record it sends. When the id changes between one record and the next the open batch is closed with an end-of-batch marker and the next one opened — and a closed batch is never reopened. There are three ways the id is chosen, and the dataset schema picks among them.

A time field, rounded to the time index. The schema's batch_id_field names a time field and sets round_batch_id_field; every record's time is rounded down to time_index_interval from the Unix epoch. With an interval of 60 seconds every batch is the top of a minute, whichever second of the minute the first record arrived in, and the batch rotates each time the minute changes. This is how a time-series dataset is stored, and it is what lets a query name its batches from a time window alone. The time field and the interval are in the same unit. The schema's flag is enough on its own; the ingester's own round_batch_id_field setting only forces rounding on.

Any field's value, as it is. Without round_batch_id_field the field's value is the batch id verbatim: a string field is the id text itself, up to 64 bytes; an integer field is its decimal text; a uint64 field is written as eight big-endian bytes. Tenant, customer, region and device ids all work this way. Two things follow. Records have to arrive grouped by that key, or the ingester closes and reopens the same batch over and over. And the query side has to be able to name the ids — a time range enumerates itself, a key set does not, so a client passes each key it wants through the SDK's string or number batch-id generator.

The ingester's clock. With no batch_id_field, or for a record that lacks the field, the id is the ingest time rounded to the interval, as a uint64. The ingester's batch_interval_seconds overrides the interval. This is what a Kafka ingester does by default.

Files that span several batches

A file's rows are rarely in time order, and a closed batch is never reopened, so a file or S3 source orders each file's records by the batch-id field before handing them over (formats.sort_records_by_batch_field, on by default) when the batch id is a rounded time field. A Parquet file holding five minutes of records yields five batches, one per minute, in order, however its rows were written. The ordering applies to numeric keys under time rounding; a file keyed by an unrounded string or integer arrives in file order and should be written grouped by that key.

A custom ingester

The ingester is one program built on the SDK's ingest client, and a customer can build another: FermiHDIIngestClient registers with the DMS as an HD_CLIENT node (a registration token, a certificate from the DMS-CA, the SCC targets in return), opens the FHDWP data plane on the interface it is given, and sends its batches straight to the Storage Cluster Controllers — the same path the ingester uses, without any of its sources. It draws on the HD_CLIENT element of the licence, or the HD_ANY pool. The ingest client is available from C++ and Python; the SDK overview has the modes and the bindings that support each.

What to watch

The ingester registers with the DMS like every node and serves /healthz and /metrics on its health port (8081 by default). The figures that matter are records in, batches closed, and the acknowledgement lag from the storage cluster; a file source that stops consuming files has usually stopped receiving acknowledgements, and the SCC's log says why. The engine sizing — two cores at the floor, four allocated, 300 × 2 MB hugepages — is in the Administrator's Guide.

docs.fermihdi.io — ingestion.md