From 0a575a720ead839337c951294d9bb7a2bc3bae55 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Wed, 29 Jul 2026 17:00:47 -0700 Subject: [PATCH 1/2] feat(telemetry): generated metrics schema and CI compatibility gate Publishes the telemetry metric vocabulary as a generated, committed JSON Schema and enforces the compatibility rules consumers depend on. telemetry/metrics.schema.json is the source of truth for what the client collects: every one of the 65 properties carries a description of what it measures and its unit, so a consumer needs nothing from this repo's source to interpret a document. A test fails the build if any property reaches the schema without one - a metric documented only by its name is a metric nobody outside this repo can read. xet_data/src/telemetry/payload.rs remains the source of truth for the code; the schema is generated from it and must never be hand-edited: UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib telemetry::schema Without that variable the same test asserts the committed file is current, so a payload change that forgets to regenerate fails the build. schemars is a dev-dependency and the JsonSchema derives are cfg(test): `cargo tree -e normal` shows zero occurrences, so nothing ships. Scope: this publishes what the client *emits* - property names, JSON types, and meanings. It deliberately does not describe how any consumer stores or indexes those documents. That is the consumer's concern, this repo is public, and it has no way to keep a description of someone else's storage layer correct. A consumer types its own storage from each property's `type`; the schema states the compatibility rules that make that safe, and the api_changes note spells out that the numeric properties must be stored as numbers or the alerting cannot work. CI gate (scripts/check_telemetry_schema_compat.py, telemetry-schema-compat job) diffs the branch schema against the merge target: added property passes, removed or retyped fails. A missing baseline is treated as the introducing commit. Checkout uses fetch-depth 0, since the shallow default cannot read the base ref. Verified against all six cases including malformed input. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 15 + Cargo.lock | 146 ++++-- Cargo.toml | 2 + ...update_260729_telemetry_schema_contract.md | 88 ++++ scripts/check_telemetry_schema_compat.py | 170 +++++++ telemetry/metrics.schema.json | 440 ++++++++++++++++++ xet_data/Cargo.toml | 1 + xet_data/src/telemetry/mod.rs | 3 + xet_data/src/telemetry/payload.rs | 80 +++- xet_data/src/telemetry/schema.rs | 280 +++++++++++ 10 files changed, 1182 insertions(+), 43 deletions(-) create mode 100644 api_changes/update_260729_telemetry_schema_contract.md create mode 100755 scripts/check_telemetry_schema_compat.py create mode 100644 telemetry/metrics.schema.json create mode 100644 xet_data/src/telemetry/schema.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 20ea2249e..def46c392 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,21 @@ jobs: run: | cargo fmt --manifest-path ./Cargo.toml --all -- --check cargo fmt --manifest-path ./hf_xet/Cargo.toml --all -- --check + telemetry-schema-compat: + name: Telemetry schema compatibility + runs-on: + group: aws-general-8-plus + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + # The baseline is read from the merge target, so the shallow default clone is not enough. + fetch-depth: 0 + - name: Check for breaking metric changes + run: | + # Consumers of this schema type each property on first sight and cannot change it + # in place. Adding a metric is fine; removing one or changing its type is not. + python3 scripts/check_telemetry_schema_compat.py \ + --baseline-ref "origin/${{ github.base_ref || github.event.repository.default_branch }}" detect-unused-dependencies: runs-on: group: aws-general-8-plus diff --git a/Cargo.lock b/Cargo.lock index 397c3c231..6fab8976c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -357,7 +357,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -757,7 +757,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1162,7 +1162,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1197,7 +1197,7 @@ checksum = "780eb241654bf097afb00fc5f054a09b687dad862e485fdcf8399bb056565370" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1281,7 +1281,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1296,6 +1296,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ecdsa" version = "0.17.0-rc.18" @@ -1387,7 +1393,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1588,7 +1594,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1736,7 +1742,7 @@ checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2427,7 +2433,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2710,7 +2716,7 @@ dependencies = [ "cfg-if 1.0.4", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2906,7 +2912,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3111,7 +3117,7 @@ checksum = "d9b20ed30f105399776b9c883e68e536ef602a16ae6f596d2c473591d6ad64c6" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3320,7 +3326,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.117", ] [[package]] @@ -3375,7 +3381,7 @@ dependencies = [ "itertools 0.14.0", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3430,7 +3436,7 @@ dependencies = [ "proc-macro2", "pyo3-macros-backend", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3442,7 +3448,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3657,6 +3663,26 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "ref-cast" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "216e8f773d7923bcba9ceb86a86c93cabb3903a11872fc3f138c49630e50b96d" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9283685feec7d69af75fb0e858d5e7378f33fe4fc699383b2916ab9273e03c" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "regex" version = "1.12.3" @@ -4041,6 +4067,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "schemars" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d115b50f4aaeea07e79c1912f645c7513d81715d0420f8bc77a18c6260b307f" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 2.0.117", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -4129,7 +4180,18 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", +] + +[[package]] +name = "serde_derive_internals" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", ] [[package]] @@ -4174,7 +4236,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4221,7 +4283,7 @@ checksum = "94e153fc76e1c6a068703d6d29c508a0b15c061c4b7e43da59cc097bc342673c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4504,7 +4566,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4524,6 +4586,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -4541,7 +4614,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4639,7 +4712,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4650,7 +4723,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4753,7 +4826,7 @@ checksum = "5c55a2eff8b69ce66c84f85e1da1c233edc36ceb85a2058d11b0d6a3c7e7569c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4832,7 +4905,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d01145a2c788d6aae4cd653afec1e8332534d7d783d01897cefcafe4428de992" dependencies = [ "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -4956,7 +5029,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -5029,7 +5102,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ad06847b7afb65c7866a36664b75c40b895e318cea4f71299f013fb22965329d" dependencies = [ "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -5264,7 +5337,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wasm-bindgen-shared", ] @@ -5450,7 +5523,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -5461,7 +5534,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -5793,7 +5866,7 @@ dependencies = [ "heck", "indexmap", "prettyplease", - "syn", + "syn 2.0.117", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -5809,7 +5882,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -5964,6 +6037,7 @@ dependencies = [ "pyo3", "rand 0.10.1", "regex", + "schemars", "serde", "serde_json", "serial_test", @@ -6046,7 +6120,7 @@ checksum = "b659052874eb698efe5b9e8cf382204678a0086ebf46982b79d6ca3182927e5d" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -6067,7 +6141,7 @@ checksum = "0e8bc7269b54418e7aeeef514aa68f8690b8c0489a06b0136e5f57c4c5ccab89" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -6087,7 +6161,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -6127,7 +6201,7 @@ checksum = "eadce39539ca5cb3985590102671f2567e659fca9666581ad3411d59207951f3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 1b60c2260..7b24dc7d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -94,6 +94,8 @@ reqwest = { version = "0.13.1", features = [ reqwest-middleware = "0.5" rust-netrc = "0.1" safe-transmute = "0.11" +# Telemetry schema generation only; a dev-dependency, never shipped. +schemars = "1.2" serde = { version = "1", features = ["derive"] } serde_json = "1" serde_repr = "0.1" diff --git a/api_changes/update_260729_telemetry_schema_contract.md b/api_changes/update_260729_telemetry_schema_contract.md new file mode 100644 index 000000000..9b2ee0ae9 --- /dev/null +++ b/api_changes/update_260729_telemetry_schema_contract.md @@ -0,0 +1,88 @@ +# Telemetry metric schema is now a generated, published contract + +**Date**: 2026-07-29 +**Crate**: `xet-data` (`xet_data::telemetry`) +**Artifact**: `telemetry/metrics.schema.json` + +Follows [`update_260728_client_transfer_telemetry.md`](./update_260728_client_transfer_telemetry.md), +which introduced the telemetry payload itself. + +## What changed + +The telemetry metric vocabulary is now exported as a generated JSON Schema at +`telemetry/metrics.schema.json`, committed alongside the code that produces it. + +**That file is the source of truth for what the client collects.** Every property carries a +description of what it measures and its unit, so a consumer needs nothing from this repo's source +to interpret a document. `xet_data/src/telemetry/payload.rs` remains the source of truth for the +*code*; the schema is generated from it and must never be hand-edited. + +**The path is a published contract and must not move or be renamed.** Consumers pin it by tag: + +``` +https://raw.githubusercontent.com/huggingface/xet-core//telemetry/metrics.schema.json +``` + +## Scope: what this repo does and does not publish + +This publishes **what the client emits** — property names, JSON types, and meanings. + +It deliberately does **not** describe how any consumer stores, indexes, or aggregates those +documents. That is the consumer's concern, this repo is public, and it has no way to keep a +description of someone else's storage layer correct. A consumer that needs to type its own storage +derives that from each property's `type`: + +| JSON Schema | Meaning | +|---|---| +| `"integer"` | fits an unsigned 64-bit integer | +| `"number"` | finite double; never NaN or infinity | +| `"boolean"` | — | +| `"string"` | short, low-cardinality except `transfer_id` | + +Every value is a scalar: never null, never nested, never an array. + +## Regenerating + +```bash +UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib telemetry::schema +``` + +Without that variable, the same test asserts the committed file is current — so changing a payload +struct without committing the regenerated schema fails `cargo test`. + +`schemars` was added as a **dev-dependency**, and the `JsonSchema` derives are `cfg(test)`. +`cargo tree -p xet-data -e normal` shows zero occurrences of it or its transitive dependencies, so +nothing new ships in a release build. + +## Every metric must be documented + +A test fails the build if any property reaches the schema without a `description`. Add a `///` doc +comment to the field in `payload.rs` and regenerate. A metric documented only by its name is a +metric nobody outside this repo can interpret. + +## Compatibility rules, now enforced in CI + +`scripts/check_telemetry_schema_compat.py` diffs the branch's schema against the merge target and +fails the `telemetry-schema-compat` job on a breaking change: + +| Change | Verdict | Why | +|---|---|---| +| Property added | allowed | a consumer that has not seen it yet simply ignores it | +| Property removed | **rejected** | dashboards and alerts built on it silently go empty | +| Property type changed | **rejected** | consumers type each field on first sight and cannot change it in place; recovering means rebuilding stored data | + +If a metric's meaning or unit changes, **add a new property** rather than repurposing the old one — +`duration_us` alongside `duration_ms`, not `duration_ms` becoming a float. + +Descriptions, formats, and property ordering are free to change; only `type` is compared. + +## Note for the receiving service + +The numeric properties must be stored as numbers, not strings, or the throughput and duration +metrics cannot be averaged or range-queried — which is the entire point of collecting them. Storing +the `metrics` object as an opaque or string-typed blob defeats the regression alerting this exists +to enable. + +Whatever mapping the receiving service derives should also tolerate a property it has not seen: +consumers pin a tag, so a client released ahead of a consumer rebuild will emit properties the +pinned schema does not contain. Those should be retained and ignored, not rejected. diff --git a/scripts/check_telemetry_schema_compat.py b/scripts/check_telemetry_schema_compat.py new file mode 100755 index 000000000..393130753 --- /dev/null +++ b/scripts/check_telemetry_schema_compat.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Rejects backward-incompatible changes to the telemetry metrics schema. + +Consumers of this schema assign each property a field type on first sight and cannot change it in +place afterwards. That makes the compatibility rules asymmetric: + + * Adding a property is fine - a consumer that has not seen it yet simply ignores it. + * Removing one silently breaks dashboards and alerts built on it. + * Changing one's type breaks ingestion for every document carrying the new type, and recovering + means rebuilding the stored data. + +So this compares the schema on the current branch against a baseline (normally `main`) and fails +on removals and type changes while allowing additions. + +Usage: + check_telemetry_schema_compat.py --baseline [--current ] + check_telemetry_schema_compat.py --baseline-ref origin/main + +Exits 0 when compatible and non-zero otherwise, including on a malformed schema. A missing +baseline is treated as compatible: that is the commit that introduces the schema. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from pathlib import Path + +SCHEMA_PATH = "telemetry/metrics.schema.json" + +# The two direction-specific definitions inside the combined document. +DEFS_KEY = "$defs" + + +def load_json(text: str, source: str) -> dict: + try: + return json.loads(text) + except json.JSONDecodeError as e: + sys.exit(f"error: {source} is not valid JSON: {e}") + + +def read_ref(ref: str, path: str) -> str | None: + """Reads a file at a git ref, or None when it does not exist there.""" + result = subprocess.run( + ["git", "show", f"{ref}:{path}"], + capture_output=True, + text=True, + ) + return result.stdout if result.returncode == 0 else None + + +def properties_by_definition(schema: dict, source: str) -> dict[str, dict[str, str]]: + """Maps definition name -> {property name -> JSON Schema type}. + + Only the `type` is compared. Descriptions, formats, and ordering are free to change: none of + them affect how a consumer types its storage. + """ + defs = schema.get(DEFS_KEY) + if not isinstance(defs, dict): + sys.exit(f"error: {source} has no {DEFS_KEY!r} object; is it the combined metrics schema?") + + out: dict[str, dict[str, str]] = {} + for name, definition in defs.items(): + props = definition.get("properties") + if not isinstance(props, dict): + sys.exit(f"error: {source} definition {name!r} has no properties") + out[name] = { + prop: spec.get("type", "") for prop, spec in props.items() + } + return out + + +def compare(baseline: dict, current: dict) -> tuple[list[str], list[str]]: + """Returns (breaking changes, additions).""" + breaking: list[str] = [] + additions: list[str] = [] + + for definition, base_props in baseline.items(): + cur_props = current.get(definition) + if cur_props is None: + breaking.append( + f"{definition}: definition removed - anything querying this shape breaks" + ) + continue + + for prop, base_type in base_props.items(): + if prop not in cur_props: + breaking.append( + f"{definition}.{prop}: removed (was {base_type}) - " + f"dashboards and alerts using it will silently go empty" + ) + elif cur_props[prop] != base_type: + breaking.append( + f"{definition}.{prop}: type changed {base_type} -> {cur_props[prop]} - " + f"consumers cannot retype a field in place, so this needs a new " + f"property name" + ) + + for prop, cur_type in cur_props.items(): + if prop not in base_props: + additions.append(f"{definition}.{prop}: added ({cur_type})") + + for definition in current: + if definition not in baseline: + additions.append(f"{definition}: new definition") + + return breaking, additions + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--baseline", type=Path, help="baseline schema file") + parser.add_argument("--baseline-ref", help="git ref to read the baseline schema from, e.g. origin/main") + parser.add_argument("--current", type=Path, default=Path(SCHEMA_PATH), help=f"current schema (default: {SCHEMA_PATH})") + args = parser.parse_args() + + if bool(args.baseline) == bool(args.baseline_ref): + parser.error("pass exactly one of --baseline or --baseline-ref") + + if args.baseline_ref: + baseline_text = read_ref(args.baseline_ref, SCHEMA_PATH) + baseline_source = f"{args.baseline_ref}:{SCHEMA_PATH}" + if baseline_text is None: + print( + f"note: {baseline_source} does not exist - treating as the commit that introduces " + f"the schema, so there is nothing to compare against." + ) + return 0 + else: + if not args.baseline.exists(): + print(f"note: {args.baseline} does not exist - nothing to compare against.") + return 0 + baseline_text = args.baseline.read_text() + baseline_source = str(args.baseline) + + if not args.current.exists(): + sys.exit( + f"error: {args.current} is missing. Generate it with:\n" + f" UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib telemetry::schema" + ) + + baseline = properties_by_definition(load_json(baseline_text, baseline_source), baseline_source) + current = properties_by_definition(load_json(args.current.read_text(), str(args.current)), str(args.current)) + + breaking, additions = compare(baseline, current) + + for addition in sorted(additions): + print(f" + {addition}") + + if not breaking: + summary = f"{len(additions)} addition(s)" if additions else "no changes" + print(f"telemetry schema is backward compatible ({summary}).") + return 0 + + print("\nBREAKING telemetry schema changes:", file=sys.stderr) + for change in sorted(breaking): + print(f" - {change}", file=sys.stderr) + print( + "\nConsumers of this schema type each property on first sight and cannot change it in " + "place.\nIf a metric's meaning or unit changed, add a new property rather than " + "repurposing the old one\n(for example duration_us alongside duration_ms).", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/telemetry/metrics.schema.json b/telemetry/metrics.schema.json new file mode 100644 index 000000000..380976122 --- /dev/null +++ b/telemetry/metrics.schema.json @@ -0,0 +1,440 @@ +{ + "$defs": { + "DownloadMetrics": { + "description": "Download documents: [`CommonMetrics`] plus how much the wire bytes expanded on disk.", + "properties": { + "arch": { + "description": "CPU architecture of the client (`x86_64`, `aarch64`, ...).", + "type": "string" + }, + "client_version": { + "description": "Version of the `hf-xet` client that produced this document.", + "type": "string" + }, + "cpu_count": { + "description": "Parallelism available to the client process. 0 when it could not be determined.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "direction": { + "description": "Which half of the transfer this describes: `upload` or `download`.", + "type": "string" + }, + "dry_run": { + "description": "Whether this was a dry run, which performs no real uploads. Always false in practice, since\ndry runs do not report at all; present so the field is never ambiguous.", + "type": "boolean" + }, + "duration_ms": { + "description": "Wall-clock time from session construction to this document being built, in milliseconds.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "endpoint_host": { + "description": "Host component of the CAS endpoint. Never a full URL - a path or query could carry\nsomething sensitive.", + "type": "string" + }, + "error_class": { + "description": "Coarse failure bucket: `none` when nothing went wrong, otherwise `auth`, `network`,\n`timeout`, `rate_limited`, `server_error`, `not_found`, `io`, `format`, `cancelled`,\n`internal`, or `other`. Deliberately coarse - error text could contain file paths.", + "type": "string" + }, + "ewma_throughput_bps": { + "description": "The client's own EWMA estimate, for comparison against the wall-clock figures. Zero rather\nthan absent when the sampler never had enough observations.", + "format": "double", + "type": "number" + }, + "expansion_ratio": { + "description": "Logical bytes produced per wire byte - dedup and compression combined, from the\ndownloader's side.", + "format": "double", + "type": "number" + }, + "logical_throughput_bps": { + "description": "Logical throughput over the whole transfer: `total_bytes_completed` per second. Exceeds\n`throughput_bps` by the dedup and compression factor.", + "format": "double", + "type": "number" + }, + "n_files": { + "description": "Number of files in the transfer.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "os": { + "description": "Operating system the client is running on, as reported by the Rust target\n(`linux`, `macos`, `windows`, ...).", + "type": "string" + }, + "outcome": { + "description": "How the transfer ended: `ok`, `error`, `cancelled`, `aborted`, `dropped`, or\n`in_progress` for a heartbeat.", + "type": "string" + }, + "peak_concurrency": { + "description": "Highest number of concurrent connections the adaptive-concurrency controller reached.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "schema_version": { + "description": "Version of this metric vocabulary. Incremented when keys are added, so a query can tell an\nabsent key from an older client that never emitted it.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "seq": { + "description": "0 for the first document; increments per heartbeat.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "terminal": { + "description": "Exactly one document per `transfer_id` carries `true`. Filter on it to get one row per\ntransfer without any group-by.", + "type": "boolean" + }, + "throughput_bps": { + "description": "Wire throughput over the whole transfer: `transfer_bytes_completed` per second, computed\nfrom `duration_ms`. Deterministic, unlike the EWMA below.", + "format": "double", + "type": "number" + }, + "total_bytes": { + "description": "Logical bytes, before dedup and compression.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "total_bytes_completed": { + "description": "Logical bytes actually finished. Below `total_bytes` for a failed or abandoned transfer.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_bytes": { + "description": "Bytes actually moved over the wire.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_bytes_completed": { + "description": "Wire bytes actually finished.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_id": { + "description": "Unique per transfer. A single `XetSession` id can cover both an upload commit and a\ndownload group, so `session_id` alone does not identify a transfer.", + "type": "string" + } + }, + "required": [ + "schema_version", + "direction", + "transfer_id", + "terminal", + "seq", + "client_version", + "os", + "arch", + "cpu_count", + "endpoint_host", + "dry_run", + "duration_ms", + "outcome", + "error_class", + "n_files", + "total_bytes", + "total_bytes_completed", + "transfer_bytes", + "transfer_bytes_completed", + "throughput_bps", + "logical_throughput_bps", + "ewma_throughput_bps", + "peak_concurrency", + "expansion_ratio" + ], + "title": "DownloadMetrics", + "type": "object" + }, + "UploadMetrics": { + "description": "Upload documents: [`CommonMetrics`] plus dedup effectiveness and shard finalization.", + "properties": { + "arch": { + "description": "CPU architecture of the client (`x86_64`, `aarch64`, ...).", + "type": "string" + }, + "client_version": { + "description": "Version of the `hf-xet` client that produced this document.", + "type": "string" + }, + "compression_ratio": { + "description": "Compressed xorb bytes over the new bytes that produced them: below 1.0 means compression\nhelped. 0.0 when nothing new was uploaded.", + "format": "double", + "type": "number" + }, + "cpu_count": { + "description": "Parallelism available to the client process. 0 when it could not be determined.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "dedup_bytes": { + "description": "Logical bytes that did not need uploading because an identical chunk already existed.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "dedup_chunks": { + "description": "Chunks that already existed and were not uploaded.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "dedup_ratio": { + "description": "Share of logical bytes avoided by dedup: `dedup_bytes / total_bytes`. 0.0 when there was\nnothing to transfer.", + "format": "double", + "type": "number" + }, + "defrag_prevented_dedup_bytes": { + "description": "Bytes that could have been deduplicated but were re-uploaded anyway, because doing\notherwise would have fragmented the xorb past the defrag threshold. The cost of the\nfragmentation-vs-dedup tradeoff.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "defrag_prevented_dedup_chunks": { + "description": "Chunks re-uploaded despite being deduplicable, for the defrag reason above.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "direction": { + "description": "Which half of the transfer this describes: `upload` or `download`.", + "type": "string" + }, + "dry_run": { + "description": "Whether this was a dry run, which performs no real uploads. Always false in practice, since\ndry runs do not report at all; present so the field is never ambiguous.", + "type": "boolean" + }, + "duration_ms": { + "description": "Wall-clock time from session construction to this document being built, in milliseconds.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "endpoint_host": { + "description": "Host component of the CAS endpoint. Never a full URL - a path or query could carry\nsomething sensitive.", + "type": "string" + }, + "error_class": { + "description": "Coarse failure bucket: `none` when nothing went wrong, otherwise `auth`, `network`,\n`timeout`, `rate_limited`, `server_error`, `not_found`, `io`, `format`, `cancelled`,\n`internal`, or `other`. Deliberately coarse - error text could contain file paths.", + "type": "string" + }, + "ewma_throughput_bps": { + "description": "The client's own EWMA estimate, for comparison against the wall-clock figures. Zero rather\nthan absent when the sampler never had enough observations.", + "format": "double", + "type": "number" + }, + "finalize_ms": { + "description": "Shard consolidation, upload, and registration, in milliseconds. 0 for a transfer that\nnever reached finalization.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "global_dedup_bytes": { + "description": "Subset of `dedup_bytes` deduplicated against the global index rather than local state.\nThe rest was matched within this session or from the local shard cache.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "global_dedup_chunks": { + "description": "Subset of `dedup_chunks` matched against the global index.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "ingest_ms": { + "description": "Chunking, hashing, and xorb upload: session start until `finalize` was called, in\nmilliseconds. 0 on a heartbeat, where the phase has not ended.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "logical_throughput_bps": { + "description": "Logical throughput over the whole transfer: `total_bytes_completed` per second. Exceeds\n`throughput_bps` by the dedup and compression factor.", + "format": "double", + "type": "number" + }, + "n_files": { + "description": "Number of files in the transfer.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "new_bytes": { + "description": "Logical bytes that were new and had to be uploaded.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "new_chunks": { + "description": "Chunks that were new and had to be uploaded.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "os": { + "description": "Operating system the client is running on, as reported by the Rust target\n(`linux`, `macos`, `windows`, ...).", + "type": "string" + }, + "outcome": { + "description": "How the transfer ended: `ok`, `error`, `cancelled`, `aborted`, `dropped`, or\n`in_progress` for a heartbeat.", + "type": "string" + }, + "peak_concurrency": { + "description": "Highest number of concurrent connections the adaptive-concurrency controller reached.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "schema_version": { + "description": "Version of this metric vocabulary. Incremented when keys are added, so a query can tell an\nabsent key from an older client that never emitted it.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "seq": { + "description": "0 for the first document; increments per heartbeat.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "shard_bytes_uploaded": { + "description": "Shard (metadata) bytes sent to CAS during finalization.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "shard_validation_entries": { + "description": "File entries the server verified while registering the shards. A proxy for how much work\nfinalization asked of the server.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "shards_completed": { + "description": "Shards the server confirmed as fully registered. Below `shards_total` when finalization\ndid not complete.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "shards_total": { + "description": "Shards this transfer produced.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "terminal": { + "description": "Exactly one document per `transfer_id` carries `true`. Filter on it to get one row per\ntransfer without any group-by.", + "type": "boolean" + }, + "throughput_bps": { + "description": "Wire throughput over the whole transfer: `transfer_bytes_completed` per second, computed\nfrom `duration_ms`. Deterministic, unlike the EWMA below.", + "format": "double", + "type": "number" + }, + "total_bytes": { + "description": "Logical bytes, before dedup and compression.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "total_bytes_completed": { + "description": "Logical bytes actually finished. Below `total_bytes` for a failed or abandoned transfer.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "total_chunks": { + "description": "Chunks the content-defined chunker produced across all files.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_bytes": { + "description": "Bytes actually moved over the wire.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_bytes_completed": { + "description": "Wire bytes actually finished.", + "format": "uint64", + "minimum": 0, + "type": "integer" + }, + "transfer_id": { + "description": "Unique per transfer. A single `XetSession` id can cover both an upload commit and a\ndownload group, so `session_id` alone does not identify a transfer.", + "type": "string" + }, + "xorb_bytes_uploaded": { + "description": "Compressed xorb bytes sent to CAS. This plus `shard_bytes_uploaded` is the real upload\ncost of the transfer.", + "format": "uint64", + "minimum": 0, + "type": "integer" + } + }, + "required": [ + "schema_version", + "direction", + "transfer_id", + "terminal", + "seq", + "client_version", + "os", + "arch", + "cpu_count", + "endpoint_host", + "dry_run", + "duration_ms", + "outcome", + "error_class", + "n_files", + "total_bytes", + "total_bytes_completed", + "transfer_bytes", + "transfer_bytes_completed", + "throughput_bps", + "logical_throughput_bps", + "ewma_throughput_bps", + "peak_concurrency", + "dedup_bytes", + "new_bytes", + "global_dedup_bytes", + "defrag_prevented_dedup_bytes", + "total_chunks", + "dedup_chunks", + "new_chunks", + "global_dedup_chunks", + "defrag_prevented_dedup_chunks", + "xorb_bytes_uploaded", + "shard_bytes_uploaded", + "shards_total", + "shards_completed", + "shard_validation_entries", + "dedup_ratio", + "compression_ratio", + "ingest_ms", + "finalize_ms" + ], + "title": "UploadMetrics", + "type": "object" + } + }, + "$id": "https://raw.githubusercontent.com/huggingface/xet-core/main/telemetry/metrics.schema.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "description": "The `metrics` object of a POST /v1/telemetry document, one shape per transfer direction. This file is the source of truth for the metric vocabulary: every property carries a description of what it measures and its unit.\n\nGenerated from xet_data/src/telemetry/payload.rs - edit the Rust structs, not this file, and regenerate with `UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib telemetry::schema`.\n\nCompatibility: adding a property is backward compatible. Removing one, or changing its type, is NOT - consumers typically assign a column or field type on first sight and cannot change it in place afterwards. If a metric's meaning or unit changes, add a new property rather than repurposing the existing one (`duration_us` alongside `duration_ms`, not `duration_ms` becoming a float). CI enforces this.\n\nEvery value is a scalar - never null, never nested, never an array. Integers fit in an unsigned 64-bit integer; numbers are finite doubles. No file names, paths, hashes, repository ids, or user ids appear anywhere in this vocabulary.", + "oneOf": [ + { + "$ref": "#/$defs/UploadMetrics" + }, + { + "$ref": "#/$defs/DownloadMetrics" + } + ], + "title": "Xet client transfer telemetry metrics" +} diff --git a/xet_data/Cargo.toml b/xet_data/Cargo.toml index 1e8cc3930..3feec64a6 100644 --- a/xet_data/Cargo.toml +++ b/xet_data/Cargo.toml @@ -84,6 +84,7 @@ ctor = { workspace = true } dirs = { workspace = true } rand = { workspace = true } regex = { workspace = true } +schemars = { workspace = true } serial_test = { workspace = true } tempfile = { workspace = true } tracing-test = { workspace = true } diff --git a/xet_data/src/telemetry/mod.rs b/xet_data/src/telemetry/mod.rs index 3018f72c9..9dbb8dd92 100644 --- a/xet_data/src/telemetry/mod.rs +++ b/xet_data/src/telemetry/mod.rs @@ -20,6 +20,9 @@ mod emit; mod outcome; #[cfg(not(target_family = "wasm"))] mod payload; +// Generates the schema from the payload types, so it cannot outlive them on wasm. +#[cfg(all(test, not(target_family = "wasm")))] +mod schema; #[cfg(not(target_family = "wasm"))] pub(crate) use emit::{ diff --git a/xet_data/src/telemetry/payload.rs b/xet_data/src/telemetry/payload.rs index 41c9c3e4a..d03f2ad2f 100644 --- a/xet_data/src/telemetry/payload.rs +++ b/xet_data/src/telemetry/payload.rs @@ -1,5 +1,10 @@ //! The metric vocabulary sent to `POST /v1/telemetry`. //! +//! This module is the source of truth for what the client measures. It is exported to +//! `telemetry/metrics.schema.json` by `schema.rs`, and that schema is the only definition anyone +//! outside this repo has - so **every field needs a doc comment**, which becomes its `description` +//! there. A test enforces it. +//! //! # Rules for changing anything in this file //! //! Consumers assign each property a field type on first sight and cannot change it in place @@ -13,7 +18,7 @@ //! //! `test_upload_key_set_is_exact` / `test_download_key_set_is_exact` pin the key sets and //! `test_numeric_types_stable` pins the types, so any of the above fails the build rather than -//! production. +//! production. CI additionally diffs the generated schema against the merge target. //! //! Every value is a `u64`, `f64`, `bool`, or `String` - never null, never nested, never an array. //! No file names, paths, hashes, repository ids, or user ids appear here; the server derives @@ -63,9 +68,21 @@ fn finite(value: f64) -> f64 { } /// Keys present in every telemetry document, in both directions, always. +/// +/// Every field carries a doc comment, and a test enforces that. The comments become the +/// `description` of each property in the published `telemetry/metrics.schema.json`, which is the +/// only definition of this vocabulary anyone outside this repo has - so a field documented only by +/// its name here is undocumented everywhere. +/// +/// `JsonSchema` is derived only under `cfg(test)`, from a dev-dependency: the schema is generated +/// by a test and committed, so nothing ships in release builds. See `schema.rs`. #[derive(Debug, Clone, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] pub struct CommonMetrics { + /// Version of this metric vocabulary. Incremented when keys are added, so a query can tell an + /// absent key from an older client that never emitted it. pub schema_version: u64, + /// Which half of the transfer this describes: `upload` or `download`. pub direction: &'static str, /// Unique per transfer. A single `XetSession` id can cover both an upload commit and a /// download group, so `session_id` alone does not identify a transfer. @@ -76,32 +93,53 @@ pub struct CommonMetrics { /// 0 for the first document; increments per heartbeat. pub seq: u64, + /// Version of the `hf-xet` client that produced this document. pub client_version: &'static str, + /// Operating system the client is running on, as reported by the Rust target + /// (`linux`, `macos`, `windows`, ...). pub os: &'static str, + /// CPU architecture of the client (`x86_64`, `aarch64`, ...). pub arch: &'static str, + /// Parallelism available to the client process. 0 when it could not be determined. pub cpu_count: u64, - /// Host component only. + /// Host component of the CAS endpoint. Never a full URL - a path or query could carry + /// something sensitive. pub endpoint_host: String, + /// Whether this was a dry run, which performs no real uploads. Always false in practice, since + /// dry runs do not report at all; present so the field is never ambiguous. pub dry_run: bool, + /// Wall-clock time from session construction to this document being built, in milliseconds. pub duration_ms: u64, + /// How the transfer ended: `ok`, `error`, `cancelled`, `aborted`, `dropped`, or + /// `in_progress` for a heartbeat. pub outcome: &'static str, + /// Coarse failure bucket: `none` when nothing went wrong, otherwise `auth`, `network`, + /// `timeout`, `rate_limited`, `server_error`, `not_found`, `io`, `format`, `cancelled`, + /// `internal`, or `other`. Deliberately coarse - error text could contain file paths. pub error_class: &'static str, + /// Number of files in the transfer. pub n_files: u64, /// Logical bytes, before dedup and compression. pub total_bytes: u64, + /// Logical bytes actually finished. Below `total_bytes` for a failed or abandoned transfer. pub total_bytes_completed: u64, /// Bytes actually moved over the wire. pub transfer_bytes: u64, + /// Wire bytes actually finished. pub transfer_bytes_completed: u64, - /// Wire throughput over the whole transfer. Deterministic, unlike the EWMA below. + /// Wire throughput over the whole transfer: `transfer_bytes_completed` per second, computed + /// from `duration_ms`. Deterministic, unlike the EWMA below. pub throughput_bps: f64, + /// Logical throughput over the whole transfer: `total_bytes_completed` per second. Exceeds + /// `throughput_bps` by the dedup and compression factor. pub logical_throughput_bps: f64, /// The client's own EWMA estimate, for comparison against the wall-clock figures. Zero rather /// than absent when the sampler never had enough observations. pub ewma_throughput_bps: f64, + /// Highest number of concurrent connections the adaptive-concurrency controller reached. pub peak_concurrency: u64, } @@ -181,35 +219,61 @@ impl CommonMetrics { /// Upload documents: [`CommonMetrics`] plus dedup effectiveness and shard finalization. #[derive(Debug, Clone, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] pub struct UploadMetrics { + /// Flattened into the same object; these keys appear alongside the upload-specific ones. #[serde(flatten)] pub common: CommonMetrics, + /// Logical bytes that did not need uploading because an identical chunk already existed. pub dedup_bytes: u64, + /// Logical bytes that were new and had to be uploaded. pub new_bytes: u64, + /// Subset of `dedup_bytes` deduplicated against the global index rather than local state. + /// The rest was matched within this session or from the local shard cache. pub global_dedup_bytes: u64, + /// Bytes that could have been deduplicated but were re-uploaded anyway, because doing + /// otherwise would have fragmented the xorb past the defrag threshold. The cost of the + /// fragmentation-vs-dedup tradeoff. pub defrag_prevented_dedup_bytes: u64, + /// Chunks the content-defined chunker produced across all files. pub total_chunks: u64, + /// Chunks that already existed and were not uploaded. pub dedup_chunks: u64, + /// Chunks that were new and had to be uploaded. pub new_chunks: u64, + /// Subset of `dedup_chunks` matched against the global index. pub global_dedup_chunks: u64, + /// Chunks re-uploaded despite being deduplicable, for the defrag reason above. pub defrag_prevented_dedup_chunks: u64, + /// Compressed xorb bytes sent to CAS. This plus `shard_bytes_uploaded` is the real upload + /// cost of the transfer. pub xorb_bytes_uploaded: u64, + /// Shard (metadata) bytes sent to CAS during finalization. pub shard_bytes_uploaded: u64, + /// Shards this transfer produced. pub shards_total: u64, + /// Shards the server confirmed as fully registered. Below `shards_total` when finalization + /// did not complete. pub shards_completed: u64, + /// File entries the server verified while registering the shards. A proxy for how much work + /// finalization asked of the server. pub shard_validation_entries: u64, - /// Share of logical bytes avoided by dedup. + /// Share of logical bytes avoided by dedup: `dedup_bytes / total_bytes`. 0.0 when there was + /// nothing to transfer. pub dedup_ratio: f64, - /// Compressed xorb bytes over the new bytes that produced them. + /// Compressed xorb bytes over the new bytes that produced them: below 1.0 means compression + /// helped. 0.0 when nothing new was uploaded. pub compression_ratio: f64, - /// Chunking, hashing, and xorb upload: session start until `finalize` was called. + /// Chunking, hashing, and xorb upload: session start until `finalize` was called, in + /// milliseconds. 0 on a heartbeat, where the phase has not ended. pub ingest_ms: u64, - /// Shard consolidation, upload, and registration. + /// Shard consolidation, upload, and registration, in milliseconds. 0 for a transfer that + /// never reached finalization. pub finalize_ms: u64, } @@ -255,7 +319,9 @@ impl UploadMetrics { /// Download documents: [`CommonMetrics`] plus how much the wire bytes expanded on disk. #[derive(Debug, Clone, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] pub struct DownloadMetrics { + /// Flattened into the same object; these keys appear alongside the download-specific ones. #[serde(flatten)] pub common: CommonMetrics, diff --git a/xet_data/src/telemetry/schema.rs b/xet_data/src/telemetry/schema.rs new file mode 100644 index 000000000..679ed2abb --- /dev/null +++ b/xet_data/src/telemetry/schema.rs @@ -0,0 +1,280 @@ +//! Generates the committed telemetry schema from the payload structs. +//! +//! The Rust types in [`payload`](super::payload) are the single source of truth; +//! `telemetry/metrics.schema.json` is a generated export of them. `schemars` is a dev-dependency +//! and the `JsonSchema` derives are `cfg(test)`, so none of this exists in a release build. +//! +//! # Regenerating +//! +//! ```text +//! UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib telemetry::schema +//! ``` +//! +//! Without that variable the same test *asserts* the committed file matches, so changing a payload +//! struct without committing the regenerated schema fails the build rather than silently shipping +//! a stale contract. +//! +//! # Scope +//! +//! This publishes *what the client emits* - the metric names, their JSON types, and what each one +//! means. How the receiving service stores, indexes, or aggregates those documents is its own +//! concern and is deliberately not described here: this repo is public, and it has no way to keep +//! a description of someone else's storage layer correct. +//! +//! A consumer that needs to type its own storage derives that from each property's `type`. The +//! schema states the compatibility rules that make doing so safe. +//! +//! # Why every field must be documented +//! +//! The generated schema is the only definition of this vocabulary that anyone outside this repo +//! has. A field whose doc comment is missing arrives there with no `description`, and is then +//! undocumented everywhere - so [`test_every_property_is_documented`] fails the build instead. +//! +//! [`test_every_property_is_documented`]: tests::test_every_property_is_documented + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +use schemars::JsonSchema; +use serde_json::{Value, json}; + +use super::payload::{DownloadMetrics, UploadMetrics}; + +/// Committed artifact: the JSON Schema for both directions. +const SCHEMA_PATH: &str = "telemetry/metrics.schema.json"; + +/// Set this to rewrite the schema instead of asserting on it. +const UPDATE_ENV: &str = "UPDATE_TELEMETRY_SCHEMA"; + +/// Repo root, derived from this crate's manifest directory. +fn repo_root() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("xet_data has a parent") + .to_path_buf() +} + +fn schema_of() -> Value { + serde_json::to_value(schemars::schema_for!(T)).expect("schemars emits valid JSON") +} + +/// Strips the per-schema preamble so the two definitions nest cleanly under `$defs`. +fn as_definition(mut schema: Value) -> Value { + if let Some(object) = schema.as_object_mut() { + object.remove("$schema"); + } + schema +} + +/// The combined JSON Schema document. +fn metrics_json_schema() -> Value { + json!({ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://raw.githubusercontent.com/huggingface/xet-core/main/telemetry/metrics.schema.json", + "title": "Xet client transfer telemetry metrics", + "description": + "The `metrics` object of a POST /v1/telemetry document, one shape per transfer \ + direction. This file is the source of truth for the metric vocabulary: every property \ + carries a description of what it measures and its unit.\n\n\ + Generated from xet_data/src/telemetry/payload.rs - edit the Rust structs, not this \ + file, and regenerate with `UPDATE_TELEMETRY_SCHEMA=1 cargo test -p xet-data --lib \ + telemetry::schema`.\n\n\ + Compatibility: adding a property is backward compatible. Removing one, or changing \ + its type, is NOT - consumers typically assign a column or field type on first sight \ + and cannot change it in place afterwards. If a metric's meaning or unit changes, add \ + a new property rather than repurposing the existing one (`duration_us` alongside \ + `duration_ms`, not `duration_ms` becoming a float). CI enforces this.\n\n\ + Every value is a scalar - never null, never nested, never an array. Integers fit in \ + an unsigned 64-bit integer; numbers are finite doubles. No file names, paths, hashes, \ + repository ids, or user ids appear anywhere in this vocabulary.", + "oneOf": [ + { "$ref": "#/$defs/UploadMetrics" }, + { "$ref": "#/$defs/DownloadMetrics" }, + ], + "$defs": { + "UploadMetrics": as_definition(schema_of::()), + "DownloadMetrics": as_definition(schema_of::()), + }, + }) +} + +/// Every property across both directions, mapped to its JSON Schema type. +/// +/// `BTreeMap` for a stable, diffable ordering. A property present in both directions must agree on +/// its type: a consumer storing both in one place could not represent a disagreement. +fn all_metric_types() -> BTreeMap { + let mut merged: BTreeMap = BTreeMap::new(); + + for schema in [schema_of::(), schema_of::()] { + let properties = schema + .get("properties") + .and_then(Value::as_object) + .cloned() + .unwrap_or_else(|| panic!("metrics schema has no properties; did serde(flatten) stop inlining?")); + + for (name, property) in properties { + let ty = property + .get("type") + .and_then(Value::as_str) + .unwrap_or_else(|| panic!("metric {name:?} has no JSON Schema type")) + .to_owned(); + + if let Some(existing) = merged.insert(name.clone(), ty.clone()) { + assert_eq!( + existing, ty, + "metric {name:?} is {existing} in one direction and {ty} in the other; a \ + consumer storing both in one place could not represent that" + ); + } + } + } + + merged +} + +/// Serializes with a trailing newline, so the file is well-formed for git and editors. +fn to_pretty(value: &Value) -> String { + let mut text = serde_json::to_string_pretty(value).expect("the generated schema serializes"); + text.push('\n'); + text +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Walks every property of both definitions. + fn each_property(schema: &Value, mut visit: impl FnMut(&str, &str, &Value)) { + for definition in ["UploadMetrics", "DownloadMetrics"] { + let properties = schema["$defs"][definition]["properties"] + .as_object() + .unwrap_or_else(|| panic!("{definition} has no properties")); + for (name, spec) in properties { + visit(definition, name, spec); + } + } + } + + #[test] + fn test_committed_schema_is_current() { + let path = repo_root().join(SCHEMA_PATH); + let generated = to_pretty(&metrics_json_schema()); + + if std::env::var_os(UPDATE_ENV).is_some() { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).expect("create telemetry/ directory"); + } + std::fs::write(&path, &generated).unwrap_or_else(|e| panic!("writing {}: {e}", path.display())); + return; + } + + let committed = std::fs::read_to_string(&path).unwrap_or_else(|e| { + panic!( + "cannot read {}: {e}\n\nGenerate it with:\n {UPDATE_ENV}=1 cargo test -p xet-data --lib telemetry::schema", + path.display() + ) + }); + + assert_eq!( + committed, generated, + "\n{SCHEMA_PATH} is out of date with the payload structs.\n\nRegenerate with:\n \ + {UPDATE_ENV}=1 cargo test -p xet-data --lib telemetry::schema\n\nThen review the diff: \ + adding a property is backward compatible, but removing one or changing its type is a \ + breaking change for consumers.\n" + ); + } + + /// The schema is the only definition of this vocabulary outside this repo, so a property + /// without a description is a metric nobody can interpret. + #[test] + fn test_every_property_is_documented() { + let schema = metrics_json_schema(); + let mut undocumented = Vec::new(); + + each_property(&schema, |definition, name, spec| { + let described = spec + .get("description") + .and_then(Value::as_str) + .is_some_and(|d| !d.trim().is_empty()); + if !described { + undocumented.push(format!("{definition}.{name}")); + } + }); + + assert!( + undocumented.is_empty(), + "these metrics have no description, so they arrive undocumented for every consumer: \ + {undocumented:?}\n\nAdd a `///` doc comment to the field in payload.rs and regenerate." + ); + } + + /// Only scalars. A nested or nullable value would break consumers that flatten the object into + /// columns, and `serde_json` renders NaN and infinity as null. + #[test] + fn test_every_property_is_a_scalar() { + let schema = metrics_json_schema(); + + each_property(&schema, |definition, name, spec| { + let ty = spec.get("type").and_then(Value::as_str); + assert!( + matches!(ty, Some("integer" | "number" | "boolean" | "string")), + "{definition}.{name} has type {ty:?}; the vocabulary is scalars only" + ); + }); + } + + #[test] + fn test_schema_documents_both_directions() { + let schema = metrics_json_schema(); + assert!(schema["$defs"]["UploadMetrics"]["properties"]["dedup_ratio"].is_object()); + assert!(schema["$defs"]["DownloadMetrics"]["properties"]["expansion_ratio"].is_object()); + // Nested `$schema` keys would make the combined document ambiguous. + assert!(schema["$defs"]["UploadMetrics"].get("$schema").is_none()); + } + + /// Shared properties must agree across directions; `all_metric_types` asserts that, and this + /// proves the merge actually runs over both. + #[test] + fn test_shared_properties_have_one_type_across_directions() { + let merged = all_metric_types(); + assert_eq!(merged.get("duration_ms").map(String::as_str), Some("integer")); + assert_eq!(merged.get("throughput_bps").map(String::as_str), Some("number")); + + // Direction-specific properties are present alongside the shared ones. + assert_eq!(merged.get("dedup_ratio").map(String::as_str), Some("number")); + assert_eq!(merged.get("expansion_ratio").map(String::as_str), Some("number")); + + // The union is strictly larger than either direction alone. + let upload_count = schema_of::()["properties"].as_object().unwrap().len(); + assert!(merged.len() > upload_count, "download-only properties were not merged in"); + } + + /// The numeric metrics that regression alerting depends on must be typed as numbers, not + /// strings - a consumer typing its storage off this schema needs them aggregatable. + #[test] + fn test_alerting_metrics_are_numeric() { + let types = all_metric_types(); + + for name in [ + "duration_ms", + "total_bytes", + "transfer_bytes", + "peak_concurrency", + "n_files", + ] { + assert_eq!(types.get(name).map(String::as_str), Some("integer"), "{name} must be numeric"); + } + for name in [ + "throughput_bps", + "logical_throughput_bps", + "ewma_throughput_bps", + "dedup_ratio", + ] { + assert_eq!(types.get(name).map(String::as_str), Some("number"), "{name} must be numeric"); + } + for name in ["direction", "outcome", "error_class", "transfer_id"] { + assert_eq!(types.get(name).map(String::as_str), Some("string"), "{name} must be groupable"); + } + assert_eq!(types.get("terminal").map(String::as_str), Some("boolean")); + } +} From 78197420270b9a9160c1971b25a598320a8ee666 Mon Sep 17 00:00:00 2001 From: Sam Horradarn Date: Thu, 30 Jul 2026 17:21:28 -0700 Subject: [PATCH 2/2] fix(telemetry): schema drift test fails on a Windows checkout `test_committed_schema_is_current` compares the committed `telemetry/metrics.schema.json` byte-for-byte against a freshly generated string. Git checks that file out with CRLF on Windows under the default `core.autocrlf=true`, while the generated string always uses `\n`, so the comparison failed there and only there - reporting the schema as out of date when it was identical. This was latent: `build_and_test-win` was already failing to compile for an unrelated reason, so the test never got far enough to run. Fixing that compile error surfaced this. Normalizes line endings before comparing, and adds a `.gitattributes` entry pinning the file to LF. The test does not rely on the latter, since `.gitattributes` only governs fresh checkouts. Verified by rewriting the committed schema with CRLF locally: the test fails without the normalization and passes with it. Co-Authored-By: Claude Opus 5 --- .gitattributes | 4 ++++ xet_data/src/telemetry/schema.rs | 19 +++++++++++++------ 2 files changed, 17 insertions(+), 6 deletions(-) create mode 100644 .gitattributes diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 000000000..ebf2b1c7a --- /dev/null +++ b/.gitattributes @@ -0,0 +1,4 @@ +# The telemetry schema is compared byte-for-byte against a generated string in +# `test_committed_schema_is_current`, and is read by `scripts/check_telemetry_schema_compat.py`. +# Pin it to LF so a Windows checkout does not rewrite it to CRLF. +telemetry/metrics.schema.json text eol=lf diff --git a/xet_data/src/telemetry/schema.rs b/xet_data/src/telemetry/schema.rs index 679ed2abb..7a1e35d41 100644 --- a/xet_data/src/telemetry/schema.rs +++ b/xet_data/src/telemetry/schema.rs @@ -168,12 +168,19 @@ mod tests { return; } - let committed = std::fs::read_to_string(&path).unwrap_or_else(|e| { - panic!( - "cannot read {}: {e}\n\nGenerate it with:\n {UPDATE_ENV}=1 cargo test -p xet-data --lib telemetry::schema", - path.display() - ) - }); + // Line endings are normalized before comparing: git checks this file out with CRLF on + // Windows under the default `core.autocrlf=true`, while the generated string always uses + // `\n`. Without this the test fails there and only there, for a reason that has nothing to + // do with the schema being out of date. `.gitattributes` pins the file to LF as well, but + // that only governs fresh checkouts, so the test does not rely on it. + let committed = std::fs::read_to_string(&path) + .unwrap_or_else(|e| { + panic!( + "cannot read {}: {e}\n\nGenerate it with:\n {UPDATE_ENV}=1 cargo test -p xet-data --lib telemetry::schema", + path.display() + ) + }) + .replace("\r\n", "\n"); assert_eq!( committed, generated,