diff --git a/Cargo.lock b/Cargo.lock index ed04aa938..1e8f8287c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3151,6 +3151,12 @@ dependencies = [ "encoding_rs", ] +[[package]] +name = "endian-type" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d" + [[package]] name = "enum-as-inner" version = "0.6.1" @@ -4721,7 +4727,7 @@ dependencies = [ "jiff", "nom 8.0.0", "num-traits", - "ordered-float", + "ordered-float 5.3.0", "rand 0.9.4", "serde", "serde_json", @@ -5162,6 +5168,7 @@ dependencies = [ "lance-core", "lance-namespace", "log", + "metrics", "moka", "object_store", "object_store_opendal", @@ -5404,6 +5411,8 @@ dependencies = [ "lance-testing", "lazy_static", "log", + "metrics", + "metrics-util", "moka", "num-traits", "object_store", @@ -5861,6 +5870,36 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + +[[package]] +name = "metrics-util" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8496cc523d1f94c1385dd8f0f0c2c480b2b8aeccb5b7e4485ad6365523ae376" +dependencies = [ + "aho-corasick", + "crossbeam-epoch", + "crossbeam-utils", + "hashbrown 0.15.5", + "indexmap 2.14.0", + "metrics", + "ordered-float 4.6.0", + "quanta", + "radix_trie", + "rand 0.9.4", + "rand_xoshiro", + "sketches-ddsketch", +] + [[package]] name = "mime" version = "0.3.17" @@ -6076,6 +6115,15 @@ dependencies = [ "rawpointer", ] +[[package]] +name = "nibble_vec" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a5d83df9f36fe23f0c3648c6bbb8b0298bb5f1939c8f2704431371f4b84d43" +dependencies = [ + "smallvec", +] + [[package]] name = "nix" version = "0.26.4" @@ -6640,6 +6688,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "ordered-float" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7bb71e1b3fa6ca1c61f383464aaf2bb0e2f8e772a1f01d486832464de363b951" +dependencies = [ + "num-traits", +] + [[package]] name = "ordered-float" version = "5.3.0" @@ -7679,6 +7736,21 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "quanta" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7" +dependencies = [ + "crossbeam-utils", + "libc", + "once_cell", + "raw-cpuid", + "wasi 0.11.1+wasi-snapshot-preview1", + "web-sys", + "winapi", +] + [[package]] name = "quick-xml" version = "0.26.0" @@ -7790,6 +7862,16 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" +[[package]] +name = "radix_trie" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c069c179fcdc6a2fe24d8d18305cf085fdbd4f922c041943e203685d6a1c58fd" +dependencies = [ + "endian-type", + "nibble_vec", +] + [[package]] name = "rancor" version = "0.1.1" @@ -7924,6 +8006,15 @@ version = "1.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "973443cf09a9c8656b574a866ab68dfa19f0867d0340648c7d2f6a71b8a8ea68" +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + [[package]] name = "raw-cpuid" version = "11.6.0" @@ -9101,6 +9192,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" +[[package]] +name = "sketches-ddsketch" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b" + [[package]] name = "slab" version = "0.4.12" diff --git a/Cargo.toml b/Cargo.toml index 712a0b6c5..e0a764dc1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -54,6 +54,8 @@ half = { "version" = "2.7.1", default-features = false, features = [ ] } futures = "0" log = "0.4" +metrics = "0.24" +metrics-util = "0.19" moka = { version = "0.12", features = ["future"] } object_store = "0.13.2" pin-project = "1.0.7" diff --git a/docs/src/js/functions/instrumentLanceDbMetrics.md b/docs/src/js/functions/instrumentLanceDbMetrics.md new file mode 100644 index 000000000..b45861106 --- /dev/null +++ b/docs/src/js/functions/instrumentLanceDbMetrics.md @@ -0,0 +1,42 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / instrumentLanceDbMetrics + +# Function: instrumentLanceDbMetrics() + +```ts +function instrumentLanceDbMetrics(meterProvider?): boolean +``` + +Register LanceDB metrics as OpenTelemetry observable instruments. + +Installs a process-global metrics recorder and creates one observable +instrument per LanceDB metric (currently object store request counts, bytes, +latency, errors, and throttles) on the given (or global) `MeterProvider`. The +configured `MetricReader` then collects them on its own schedule. + +Counters and gauges map directly to observable counters/gauges. Because +OpenTelemetry has no asynchronous histogram instrument, each histogram is +exported Prometheus-style as cumulative `le` bucket counts (`_bucket`, +with an `le` attribute) plus `_count` and `_sum`. + +Requires `@opentelemetry/api` (a dependency) and, to actually export, an +OpenTelemetry SDK such as `@opentelemetry/sdk-metrics`. + +## Parameters + +* **meterProvider?**: `MeterProvider` + The provider to register instruments on. Defaults to the + global provider from `@opentelemetry/api`. + +## Returns + +`boolean` + +`true` if the recorder is installed and instruments are registered. + `false` if a different `metrics` recorder is already installed in this + process (only one global recorder is permitted), in which case a warning is + emitted and no instruments are created. Calling this more than once is safe; + instruments are created only on the first successful call. diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index 79d842346..7baaa4f92 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -131,6 +131,7 @@ - [RecordBatchIterator](functions/RecordBatchIterator.md) - [connect](functions/connect.md) - [connectNamespace](functions/connectNamespace.md) +- [instrumentLanceDbMetrics](functions/instrumentLanceDbMetrics.md) - [makeArrowTable](functions/makeArrowTable.md) - [packBits](functions/packBits.md) - [permutationBuilder](functions/permutationBuilder.md) diff --git a/nodejs/Cargo.toml b/nodejs/Cargo.toml index 31d62c10c..563464265 100644 --- a/nodejs/Cargo.toml +++ b/nodejs/Cargo.toml @@ -44,6 +44,6 @@ aws-lc-rs = "=1.16.3" napi-build = "2.3.1" [features] -default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/goosefs"] +default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/goosefs", "lancedb/metrics-otel"] fp16kernels = ["lancedb/fp16kernels"] remote = ["lancedb/remote"] diff --git a/nodejs/__test__/otel.test.ts b/nodejs/__test__/otel.test.ts new file mode 100644 index 000000000..2e6be6b64 --- /dev/null +++ b/nodejs/__test__/otel.test.ts @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +import { + MeterProvider, + type MetricData, + MetricReader, +} from "@opentelemetry/sdk-metrics"; +import * as tmp from "tmp"; +import { connect, instrumentLanceDbMetrics } from "../lancedb"; +// snapshotLancedbMetrics is internal plumbing (not part of the public API), so +// it is imported from the native module rather than the package entry point. +import { snapshotLancedbMetrics } from "../lancedb/native"; + +// The metrics recorder is process-global and installed once, so the whole +// bridge is exercised in a single test to avoid cross-test global-state coupling. + +// A minimal pull-based reader whose `collect()` we drive directly, invoking the +// observable-instrument callbacks. `@opentelemetry/sdk-metrics` ships no +// in-memory reader, so we subclass the abstract base. +class TestMetricReader extends MetricReader { + protected async onForceFlush(): Promise { + // no-op: collection is driven directly via collect() + } + protected async onShutdown(): Promise { + // no-op: nothing to release + } +} + +async function metricsByName( + reader: TestMetricReader, +): Promise> { + const collected = await reader.collect(); + const result = new Map(); + for (const scope of collected.resourceMetrics.scopeMetrics) { + for (const metric of scope.metrics) { + result.set(metric.descriptor.name, metric); + } + } + return result; +} + +describe("OpenTelemetry metrics bridge", () => { + let tmpDir: tmp.DirResult; + beforeEach(() => { + tmpDir = tmp.dirSync({ unsafeCleanup: true }); + }); + afterEach(() => tmpDir.removeCallback()); + + it("snapshot is safe to call regardless of install state", () => { + expect(Array.isArray(snapshotLancedbMetrics())).toBe(true); + }); + + it("exports object store metrics via observable instruments", async () => { + const reader = new TestMetricReader(); + const provider = new MeterProvider({ readers: [reader] }); + expect(instrumentLanceDbMetrics(provider)).toBe(true); + + // Generate object store activity on the local filesystem (scheme "file"). + const db = await connect(tmpDir.name); + const data = Array.from({ length: 256 }, (_, i) => ({ id: i })); + const table = await db.createTable("t", data); + expect(await table.countRows()).toBe(256); + + const metrics = await metricsByName(reader); + + const requests = metrics.get("lance_object_store_requests_total"); + expect(requests).toBeDefined(); + // biome-ignore lint/suspicious/noExplicitAny: SDK point shape + const requestPoints = (requests!.dataPoints as any[]) ?? []; + expect(requestPoints.length).toBeGreaterThan(0); + for (const p of requestPoints) { + // Labelled by `operation` and `base` (the store scheme by default). + expect(p.attributes).toHaveProperty("base"); + expect(p.attributes).toHaveProperty("operation"); + } + const totalRequests = requestPoints.reduce((acc, p) => acc + p.value, 0); + expect(totalRequests).toBeGreaterThan(0); + + // Histograms are decomposed into bucket / count / sum observable counters. + const bucket = metrics.get( + "lance_object_store_request_duration_seconds_bucket", + ); + expect(bucket).toBeDefined(); + // biome-ignore lint/suspicious/noExplicitAny: SDK point shape + const bucketPoints = (bucket!.dataPoints as any[]) ?? []; + expect(bucketPoints.length).toBeGreaterThan(0); + expect(bucketPoints.every((p) => "le" in p.attributes)).toBe(true); + // The implicit +Inf bucket must be present. + expect(bucketPoints.some((p) => p.attributes.le === "+Inf")).toBe(true); + + const count = metrics.get( + "lance_object_store_request_duration_seconds_count", + ); + expect(count).toBeDefined(); + // biome-ignore lint/suspicious/noExplicitAny: SDK point shape + const countPoints = (count!.dataPoints as any[]) ?? []; + expect(countPoints.reduce((acc, p) => acc + p.value, 0)).toBeGreaterThan(0); + + const sum = metrics.get("lance_object_store_request_duration_seconds_sum"); + expect(sum).toBeDefined(); + // biome-ignore lint/suspicious/noExplicitAny: SDK point shape + const sumPoints = (sum!.dataPoints as any[]) ?? []; + expect(sumPoints.reduce((acc, p) => acc + p.value, 0)).toBeGreaterThan(0); + + // Unit handling: only `_sum` keeps the histogram's unit (seconds); `_bucket` + // and `_count` observe cumulative counts and are unitless. + expect(sum!.descriptor.unit).toBe("s"); + expect(bucket!.descriptor.unit).toBe(""); + expect(count!.descriptor.unit).toBe(""); + + await provider.shutdown(); + }); +}); diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index b9939726d..ebd638634 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -20,6 +20,11 @@ import { HeaderProvider } from "./header"; // Re-export native header provider for use with connectWithHeaderProvider export { JsHeaderProvider as NativeJsHeaderProvider } from "./native.js"; +// OpenTelemetry metrics bridge. Only the high-level entry point is public; the +// underlying recorder/catalog/snapshot functions remain internal plumbing that +// `otel.ts` consumes from the native module. +export { instrumentLanceDbMetrics } from "./otel"; + export { AddColumnsSql, ConnectionOptions, diff --git a/nodejs/lancedb/otel.ts b/nodejs/lancedb/otel.ts new file mode 100644 index 000000000..b3b329831 --- /dev/null +++ b/nodejs/lancedb/otel.ts @@ -0,0 +1,137 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +import { + type Attributes, + type MeterProvider, + type ObservableResult, + metrics, +} from "@opentelemetry/api"; + +import { + lancedbMetricsCatalog, + registerLancedbMetricsRecorder, + snapshotLancedbMetrics, +} from "./native"; + +let instrumented = false; + +/** + * Register LanceDB metrics as OpenTelemetry observable instruments. + * + * Installs a process-global metrics recorder and creates one observable + * instrument per LanceDB metric (currently object store request counts, bytes, + * latency, errors, and throttles) on the given (or global) `MeterProvider`. The + * configured `MetricReader` then collects them on its own schedule. + * + * Counters and gauges map directly to observable counters/gauges. Because + * OpenTelemetry has no asynchronous histogram instrument, each histogram is + * exported Prometheus-style as cumulative `le` bucket counts (`_bucket`, + * with an `le` attribute) plus `_count` and `_sum`. + * + * Requires `@opentelemetry/api` (a dependency) and, to actually export, an + * OpenTelemetry SDK such as `@opentelemetry/sdk-metrics`. + * + * @param meterProvider The provider to register instruments on. Defaults to the + * global provider from `@opentelemetry/api`. + * @returns `true` if the recorder is installed and instruments are registered. + * `false` if a different `metrics` recorder is already installed in this + * process (only one global recorder is permitted), in which case a warning is + * emitted and no instruments are created. Calling this more than once is safe; + * instruments are created only on the first successful call. + */ +export function instrumentLanceDbMetrics( + meterProvider?: MeterProvider, +): boolean { + if (!registerLancedbMetricsRecorder()) { + console.warn( + "Could not install the LanceDB metrics recorder: another `metrics` " + + "recorder is already installed in this process. LanceDB metrics will " + + "not be exported via OpenTelemetry.", + ); + return false; + } + + if (instrumented) { + return true; + } + + const provider = meterProvider ?? metrics.getMeterProvider(); + const meter = provider.getMeter("lancedb"); + + const scalarCallback = (metricName: string) => (result: ObservableResult) => { + for (const point of snapshotLancedbMetrics()) { + if (point.name === metricName && point.value != null) { + result.observe(point.value, point.attributes); + } + } + }; + + const bucketCallback = (metricName: string) => (result: ObservableResult) => { + for (const point of snapshotLancedbMetrics()) { + if (point.name !== metricName || point.buckets == null) { + continue; + } + for (const bucket of point.buckets) { + const attributes: Attributes = { + ...point.attributes, + le: bucket.le, + }; + result.observe(bucket.cumulativeCount, attributes); + } + } + }; + + const fieldCallback = + (metricName: string, field: "count" | "sum") => + (result: ObservableResult) => { + for (const point of snapshotLancedbMetrics()) { + if (point.name !== metricName) { + continue; + } + const value = point[field]; + if (value != null) { + result.observe(value, point.attributes); + } + } + }; + + for (const desc of lancedbMetricsCatalog()) { + const unit = desc.unit ?? ""; + if (desc.kind === "counter") { + const counter = meter.createObservableCounter(desc.name, { + unit, + description: desc.description, + }); + counter.addCallback(scalarCallback(desc.name)); + } else if (desc.kind === "gauge") { + const gauge = meter.createObservableGauge(desc.name, { + unit, + description: desc.description, + }); + gauge.addCallback(scalarCallback(desc.name)); + } else if (desc.kind === "histogram") { + // `_bucket` and `_count` observe cumulative sample counts, not the + // histogram's measured quantity, so they are unitless; only `_sum` + // carries the histogram's unit. + const bucket = meter.createObservableCounter(`${desc.name}_bucket`, { + description: `${desc.description} (cumulative buckets)`, + }); + bucket.addCallback(bucketCallback(desc.name)); + + const count = meter.createObservableCounter(`${desc.name}_count`, { + description: `${desc.description} (count)`, + }); + count.addCallback(fieldCallback(desc.name, "count")); + + const sum = meter.createObservableCounter(`${desc.name}_sum`, { + unit, + description: `${desc.description} (sum)`, + }); + sum.addCallback(fieldCallback(desc.name, "sum")); + } + } + + instrumented = true; + return true; +} diff --git a/nodejs/package.json b/nodejs/package.json index cf0cb5f02..e11d5b5f6 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -44,6 +44,7 @@ "@biomejs/biome": "^1.7.3", "@jest/globals": "^29.7.0", "@napi-rs/cli": "3.7.0", + "@opentelemetry/sdk-metrics": "^1.30.0", "@types/axios": "^0.14.0", "@types/jest": "^29.1.2", "@types/node": "22.7.4", @@ -92,6 +93,7 @@ "version": "napi version" }, "dependencies": { + "@opentelemetry/api": "^1.9.0", "reflect-metadata": "^0.2.2" }, "optionalDependencies": { diff --git a/nodejs/pnpm-lock.yaml b/nodejs/pnpm-lock.yaml index 580331248..c21c636d2 100644 --- a/nodejs/pnpm-lock.yaml +++ b/nodejs/pnpm-lock.yaml @@ -8,6 +8,9 @@ importers: .: dependencies: + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 apache-arrow: specifier: '>=15.0.0 <=18.1.0' version: 18.1.0 @@ -33,6 +36,9 @@ importers: '@napi-rs/cli': specifier: 3.7.0 version: 3.7.0(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)(@types/node@22.7.4) + '@opentelemetry/sdk-metrics': + specifier: ^1.30.0 + version: 1.30.1(@opentelemetry/api@1.9.1) '@types/axios': specifier: ^0.14.0 version: 0.14.4 @@ -1307,6 +1313,32 @@ packages: '@octokit/types@16.0.0': resolution: {integrity: sha512-sKq+9r1Mm4efXW1FCk7hFSeJo4QKreL/tTbR0rz/qx/r1Oa2VV83LTA/H/MuCOX7uCIJmQVRKBcbmWoySjAnSg==} + '@opentelemetry/api@1.9.1': + resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==} + engines: {node: '>=8.0.0'} + + '@opentelemetry/core@1.30.1': + resolution: {integrity: sha512-OOCM2C/QIURhJMuKaekP3TRBxBKxG/TWWA0TL2J6nXUtDnuCtccy49LUJF8xPFXMX+0LMcxFpCo8M9cGY1W6rQ==} + engines: {node: '>=14'} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/resources@1.30.1': + resolution: {integrity: sha512-5UxZqiAgLYGFjS4s9qm5mBVo433u+dSPUFWVWXmLAD4wB65oMCoXaJP1KJa9DIYYMeHu3z4BZcStG3LC593cWA==} + engines: {node: '>=14'} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/sdk-metrics@1.30.1': + resolution: {integrity: sha512-q9zcZ0Okl8jRgmy7eNW3Ku1XSgg3sDLa5evHZpCwjspw7E8Is4K/haRPDJrBcX3YSn/Y7gUvFnByNYEKQNbNog==} + engines: {node: '>=14'} + peerDependencies: + '@opentelemetry/api': '>=1.3.0 <1.10.0' + + '@opentelemetry/semantic-conventions@1.28.0': + resolution: {integrity: sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA==} + engines: {node: '>=14'} + '@protobufjs/aspromise@1.1.2': resolution: {integrity: sha512-j+gKExEuLmKwvz3OgROXtrJ2UG2x8Ch2YZUxahh+s1F2HZ+wAceUNLkvy6zKCPVRkU++ZWQrdxsUeQXmcg4uoQ==} @@ -4925,6 +4957,27 @@ snapshots: dependencies: '@octokit/openapi-types': 27.0.0 + '@opentelemetry/api@1.9.1': {} + + '@opentelemetry/core@1.30.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/semantic-conventions': 1.28.0 + + '@opentelemetry/resources@1.30.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 1.30.1(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.28.0 + + '@opentelemetry/sdk-metrics@1.30.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 1.30.1(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 1.30.1(@opentelemetry/api@1.9.1) + + '@opentelemetry/semantic-conventions@1.28.0': {} + '@protobufjs/aspromise@1.1.2': optional: true diff --git a/nodejs/src/lib.rs b/nodejs/src/lib.rs index b95602f7b..93bb2044a 100644 --- a/nodejs/src/lib.rs +++ b/nodejs/src/lib.rs @@ -12,6 +12,7 @@ mod header; mod index; mod iterator; pub mod merge; +pub mod otel; pub mod permutation; mod query; pub mod remote; diff --git a/nodejs/src/otel.rs b/nodejs/src/otel.rs new file mode 100644 index 000000000..0824c8a87 --- /dev/null +++ b/nodejs/src/otel.rs @@ -0,0 +1,119 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Node.js bindings over [`lancedb::metrics_otel`]. +//! +//! The aggregation, catalog, and histogram bucketing all live in the LanceDB +//! core crate; this module only converts the core snapshot types into napi +//! objects and exposes the three entry points to JavaScript, where +//! `lancedb/otel.ts` bridges them into the user's OpenTelemetry `MeterProvider`. + +use std::collections::HashMap; + +use lancedb::metrics_otel::{MetricPoint as CoreMetricPoint, MetricValue}; +use napi_derive::napi; + +/// One cumulative histogram bucket: all samples with value `<= le`. +#[napi(object)] +pub struct MetricBucket { + /// The inclusive upper bound of the bucket, or `"+Inf"` for the final bucket. + pub le: String, + /// Cumulative number of samples less than or equal to `le`. + pub cumulative_count: f64, +} + +/// One aggregated metric data point. For counters and gauges only `value` is +/// set; for histograms `buckets` (cumulative `le` counts), `count`, and `sum` +/// are set. +#[napi(object)] +pub struct MetricPoint { + pub name: String, + pub kind: String, + pub attributes: HashMap, + pub value: Option, + pub buckets: Option>, + pub count: Option, + pub sum: Option, +} + +impl From for MetricPoint { + fn from(point: CoreMetricPoint) -> Self { + let kind = point.kind.as_str().to_string(); + let (value, buckets, count, sum) = match point.value { + MetricValue::Scalar(v) => (Some(v), None, None, None), + MetricValue::Histogram { + buckets, + count, + sum, + } => ( + None, + Some( + buckets + .into_iter() + // Counts stay well within the f64-exact integer range + // (2^53), so this cast is lossless in practice and keeps + // the values plain JS numbers for OpenTelemetry. + .map(|(le, cumulative_count)| MetricBucket { + le, + cumulative_count: cumulative_count as f64, + }) + .collect(), + ), + Some(count as f64), + Some(sum), + ), + }; + Self { + name: point.name, + kind, + attributes: point.attributes, + value, + buckets, + count, + sum, + } + } +} + +/// A described metric, used by the JavaScript layer to create instruments up front. +#[napi(object)] +pub struct MetricDescription { + pub name: String, + pub kind: String, + pub unit: Option, + pub description: String, +} + +/// Install the LanceDB metrics recorder as the process-global `metrics` recorder. +/// +/// Returns `true` if the recorder is installed (now or previously). Returns +/// `false` if a *different* recorder is already installed — `metrics` allows +/// only one global recorder per process, so LanceDB cannot coexist with another. +#[napi] +pub fn register_lancedb_metrics_recorder() -> bool { + lancedb::metrics_otel::register_metrics_recorder() +} + +/// The catalog of described LanceDB metrics. Empty until the recorder is installed. +#[napi] +pub fn lancedb_metrics_catalog() -> Vec { + lancedb::metrics_otel::metrics_catalog() + .into_iter() + .map(|desc| MetricDescription { + name: desc.name, + kind: desc.kind.as_str().to_string(), + unit: desc.unit, + description: desc.description, + }) + .collect() +} + +/// A point-in-time snapshot of every recorded metric. Empty until the recorder +/// is installed. +#[napi] +pub fn snapshot_lancedb_metrics() -> Vec { + lancedb::metrics_otel::snapshot_metrics() + .into_iter() + .map(MetricPoint::from) + .collect() +} diff --git a/python/Cargo.toml b/python/Cargo.toml index f8cf9b666..5495647de 100644 --- a/python/Cargo.toml +++ b/python/Cargo.toml @@ -47,6 +47,6 @@ pyo3-build-config = { version = "0.28", features = [ ] } [features] -default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs"] +default = ["remote", "lancedb/aws", "lancedb/gcs", "lancedb/azure", "lancedb/dynamodb", "lancedb/oss", "lancedb/huggingface", "lancedb/cos", "lancedb/goosefs", "lancedb/metrics-otel"] fp16kernels = ["lancedb/fp16kernels"] remote = ["lancedb/remote"] diff --git a/python/pyproject.toml b/python/pyproject.toml index 5f0ab3eec..7806daed1 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -47,6 +47,10 @@ repository = "https://github.com/lancedb/lancedb" pylance = [ "pylance>=5.0.0b5", ] +# A library only needs the OpenTelemetry API; the application supplies and +# configures the SDK (the actual exporter/reader). See +# https://opentelemetry.io/docs/languages/python/instrumentation/ +otel = ["opentelemetry-api"] tests = [ "aiohttp>=3.9.0", "boto3>=1.28.57", @@ -61,6 +65,7 @@ tests = [ "pylance>=5.0.0b5", "requests>=2.31.0", "datafusion>=52,<53", + "opentelemetry-sdk>=1.30.0", ] dev = [ "ruff>=0.3.0", diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 8667b32c3..18757ebfb 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -30,6 +30,25 @@ IvfHnswPq: type[HnswPq] = HnswPq IvfHnswSq: type[HnswSq] = HnswSq IvfHnswFlat: type[HnswFlat] = HnswFlat +class MetricPoint: + name: str + kind: str + attributes: Dict[str, str] + value: Optional[float] + buckets: Optional[List[Tuple[str, int]]] + count: Optional[int] + sum: Optional[float] + +class MetricDescription: + name: str + kind: str + unit: Optional[str] + description: str + +def register_lancedb_metrics_recorder() -> bool: ... +def lancedb_metrics_catalog() -> List[MetricDescription]: ... +def snapshot_lancedb_metrics() -> List[MetricPoint]: ... + class PyExpr: """A type-safe DataFusion expression node (Rust-side handle).""" diff --git a/python/python/lancedb/otel.py b/python/python/lancedb/otel.py new file mode 100644 index 000000000..e2b8533f3 --- /dev/null +++ b/python/python/lancedb/otel.py @@ -0,0 +1,170 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +"""Bridge LanceDB's internal metrics into OpenTelemetry. + +LanceDB (through Lance core) publishes metrics (currently object store request +counts, bytes, latency, errors, and throttles) through the Rust ``metrics`` +facade. This module installs a process-global recorder that aggregates them and +registers OpenTelemetry observable instruments that report the aggregated values +into the user's ``MeterProvider``. + +The bridge is generic: every metric LanceDB describes is surfaced automatically, +with no per-metric Python code. Histograms have no asynchronous OpenTelemetry +instrument, so each is exported Prometheus-style as cumulative ``le`` buckets +plus ``_count`` and ``_sum`` observable counters. +""" + +from __future__ import annotations + +import warnings +from typing import TYPE_CHECKING, Optional + +from ._lancedb import ( + lancedb_metrics_catalog, + register_lancedb_metrics_recorder, + snapshot_lancedb_metrics, +) + +if TYPE_CHECKING: + from opentelemetry.metrics import MeterProvider + +_INSTRUMENTED = False + + +def instrument_lancedb_metrics( + meter_provider: Optional["MeterProvider"] = None, +) -> bool: + """Register LanceDB metrics as OpenTelemetry observable instruments. + + Installs a process-global metrics recorder and creates one observable + instrument per LanceDB metric on the given (or global) ``MeterProvider``. The + user's configured ``MetricReader`` then collects them on its own schedule. + + Counters and gauges map directly to observable counters/gauges. Each + histogram is exported as cumulative ``le`` bucket counts (``_bucket``, + with an ``le`` attribute) plus ``_count`` and ``_sum``. + + Parameters + ---------- + meter_provider : opentelemetry.metrics.MeterProvider, optional + The provider to register instruments on. Defaults to the global provider + from ``opentelemetry.metrics.get_meter_provider()``. + + Returns + ------- + bool + ``True`` if the recorder is installed and instruments are registered. + ``False`` if a different ``metrics`` recorder is already installed in + this process (``metrics`` permits only one global recorder), in which + case a warning is emitted and no instruments are created. + + Notes + ----- + Requires the OpenTelemetry API (``pip install lancedb[otel]``) and, to + actually export, an OpenTelemetry SDK (``pip install opentelemetry-sdk``) + configured by the application. Calling this more than once is safe; + instruments are created only on the first successful call. + """ + global _INSTRUMENTED + + try: + from opentelemetry.metrics import Observation, get_meter_provider + except ImportError as exc: + raise ImportError( + "instrument_lancedb_metrics requires the OpenTelemetry API/SDK. " + "Install it with `pip install lancedb[otel]` or " + "`pip install opentelemetry-sdk`." + ) from exc + + if not register_lancedb_metrics_recorder(): + warnings.warn( + "Could not install the LanceDB metrics recorder: another `metrics` " + "recorder is already installed in this process. LanceDB metrics will " + "not be exported via OpenTelemetry.", + stacklevel=2, + ) + return False + + if _INSTRUMENTED: + return True + + provider = meter_provider or get_meter_provider() + meter = provider.get_meter("lancedb") + + def scalar_callback(metric_name: str): + def callback(_options): + return [ + Observation(point.value, point.attributes) + for point in snapshot_lancedb_metrics() + if point.name == metric_name and point.value is not None + ] + + return callback + + def bucket_callback(metric_name: str): + def callback(_options): + observations = [] + for point in snapshot_lancedb_metrics(): + if point.name != metric_name or point.buckets is None: + continue + for le, cumulative in point.buckets: + attributes = dict(point.attributes) + attributes["le"] = le + observations.append(Observation(cumulative, attributes)) + return observations + + return callback + + def field_callback(metric_name: str, field: str): + def callback(_options): + observations = [] + for point in snapshot_lancedb_metrics(): + if point.name != metric_name: + continue + value = getattr(point, field) + if value is not None: + observations.append(Observation(value, point.attributes)) + return observations + + return callback + + for desc in lancedb_metrics_catalog(): + unit = desc.unit or "" + if desc.kind == "counter": + meter.create_observable_counter( + desc.name, + callbacks=[scalar_callback(desc.name)], + unit=unit, + description=desc.description, + ) + elif desc.kind == "gauge": + meter.create_observable_gauge( + desc.name, + callbacks=[scalar_callback(desc.name)], + unit=unit, + description=desc.description, + ) + elif desc.kind == "histogram": + # `_bucket` and `_count` observe cumulative sample counts, not the + # histogram's measured quantity, so they are unitless; only `_sum` + # carries the histogram's unit. + meter.create_observable_counter( + f"{desc.name}_bucket", + callbacks=[bucket_callback(desc.name)], + description=f"{desc.description} (cumulative buckets)", + ) + meter.create_observable_counter( + f"{desc.name}_count", + callbacks=[field_callback(desc.name, "count")], + description=f"{desc.description} (count)", + ) + meter.create_observable_counter( + f"{desc.name}_sum", + callbacks=[field_callback(desc.name, "sum")], + unit=unit, + description=f"{desc.description} (sum)", + ) + + _INSTRUMENTED = True + return True diff --git a/python/src/lib.rs b/python/src/lib.rs index 5f3039e25..024ef292f 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -27,6 +27,7 @@ pub mod header; pub mod index; pub mod namespace; pub mod oauth; +pub mod otel; pub mod permutation; pub mod query; pub mod runtime; @@ -61,6 +62,15 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + // OpenTelemetry metrics bridge + m.add_class::()?; + m.add_class::()?; + m.add_function(wrap_pyfunction!( + otel::register_lancedb_metrics_recorder, + m + )?)?; + m.add_function(wrap_pyfunction!(otel::lancedb_metrics_catalog, m)?)?; + m.add_function(wrap_pyfunction!(otel::snapshot_lancedb_metrics, m)?)?; m.add_function(wrap_pyfunction!(connect, m)?)?; m.add_function(wrap_pyfunction!(connect_namespace, m)?)?; m.add_function(wrap_pyfunction!(connect_namespace_client, m)?)?; diff --git a/python/src/otel.rs b/python/src/otel.rs new file mode 100644 index 000000000..b610867c9 --- /dev/null +++ b/python/src/otel.rs @@ -0,0 +1,96 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Python-facing wrappers over [`lancedb::metrics_otel`]. +//! +//! The aggregation, catalog, and histogram bucketing all live in the LanceDB +//! core crate; this module only converts the core snapshot types into PyO3 +//! classes and exposes the three entry points to Python, where +//! `lancedb/otel.py` bridges them into the user's OpenTelemetry `MeterProvider`. + +use std::collections::HashMap; + +use lancedb::metrics_otel::{MetricPoint, MetricValue}; +use pyo3::prelude::*; + +/// One metric data point exposed to Python. For counters and gauges only +/// `value` is set; for histograms `buckets` (cumulative `le` counts), `count`, +/// and `sum` are set. +#[pyclass(name = "MetricPoint", get_all)] +pub struct PyMetricPoint { + name: String, + kind: String, + attributes: HashMap, + value: Option, + buckets: Option>, + count: Option, + sum: Option, +} + +impl From for PyMetricPoint { + fn from(point: MetricPoint) -> Self { + let kind = point.kind.as_str().to_string(); + let (value, buckets, count, sum) = match point.value { + MetricValue::Scalar(v) => (Some(v), None, None, None), + MetricValue::Histogram { + buckets, + count, + sum, + } => (None, Some(buckets), Some(count), Some(sum)), + }; + Self { + name: point.name, + kind, + attributes: point.attributes, + value, + buckets, + count, + sum, + } + } +} + +/// A described metric, used by the Python layer to create instruments up front. +#[pyclass(name = "MetricDescription", get_all)] +pub struct PyMetricDescription { + name: String, + kind: String, + unit: Option, + description: String, +} + +/// Install the LanceDB metrics recorder as the process-global `metrics` recorder. +/// +/// Returns `True` if the recorder is installed (now or previously). Returns +/// `False` if a *different* recorder is already installed — `metrics` allows +/// only one global recorder per process, so LanceDB cannot coexist with another. +#[pyfunction] +pub fn register_lancedb_metrics_recorder() -> bool { + lancedb::metrics_otel::register_metrics_recorder() +} + +/// The catalog of described LanceDB metrics. Empty until the recorder is installed. +#[pyfunction] +pub fn lancedb_metrics_catalog() -> Vec { + lancedb::metrics_otel::metrics_catalog() + .into_iter() + .map(|desc| PyMetricDescription { + name: desc.name, + kind: desc.kind.as_str().to_string(), + unit: desc.unit, + description: desc.description, + }) + .collect() +} + +/// A point-in-time snapshot of every recorded metric. Empty until the recorder +/// is installed. +/// +/// The read is lock-free but not O(1): it walks every registered series and +/// allocates owned copies of their names and labels. The GIL is released across +/// that work so a periodic collection doesn't stall other Python threads. +#[pyfunction] +pub fn snapshot_lancedb_metrics(py: Python<'_>) -> Vec { + let points = py.detach(lancedb::metrics_otel::snapshot_metrics); + points.into_iter().map(PyMetricPoint::from).collect() +} diff --git a/python/tests/test_otel.py b/python/tests/test_otel.py new file mode 100644 index 000000000..c182550c7 --- /dev/null +++ b/python/tests/test_otel.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +import lancedb +import pyarrow as pa +import pytest + +# The metrics recorder is process-global and installed once, so the whole +# bridge is exercised in a single test to avoid cross-test global-state coupling. + + +def _metrics_by_name(reader): + data = reader.get_metrics_data() + result = {} + for resource_metrics in data.resource_metrics: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + result[metric.name] = metric + return result + + +def test_instrument_lancedb_metrics_exports_object_store_metrics(tmp_path): + pytest.importorskip("opentelemetry.sdk.metrics") + from lancedb.otel import instrument_lancedb_metrics + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import InMemoryMetricReader + + reader = InMemoryMetricReader() + provider = MeterProvider(metric_readers=[reader]) + assert instrument_lancedb_metrics(provider) + + # The catalog is populated once the recorder is installed. + from lancedb._lancedb import lancedb_metrics_catalog + + catalog = {desc.name: desc for desc in lancedb_metrics_catalog()} + # Every metric kind emitted by the object store must be described so it is + # surfaced by the bridge (counter, histogram, and gauge). + assert catalog["lance_object_store_requests_total"].kind == "counter" + assert catalog["lance_object_store_request_duration_seconds"].kind == "histogram" + assert catalog["lance_object_store_in_flight_requests"].kind == "gauge" + assert catalog["lance_object_store_retryable_responses_total"].kind == "counter" + + # Generate object store activity on the local filesystem (scheme "file"). + db = lancedb.connect(str(tmp_path)) + table = db.create_table("t", pa.table({"id": pa.array(range(256))})) + assert table.count_rows() == 256 + assert table.to_arrow().num_rows == 256 + + metrics = _metrics_by_name(reader) + + requests = metrics["lance_object_store_requests_total"] + points = list(requests.data.data_points) + assert points, "expected at least one request data point" + # Object store metrics are labelled by `operation` and `base` (the store + # scheme, e.g. "file", by default). + assert all("base" in p.attributes and "operation" in p.attributes for p in points) + assert sum(p.value for p in points) > 0 + + # Histograms are decomposed into bucket / count / sum observable counters. + bucket = metrics["lance_object_store_request_duration_seconds_bucket"] + bucket_points = list(bucket.data.data_points) + assert bucket_points + assert all("le" in p.attributes for p in bucket_points) + # The implicit +Inf bucket must be present and is the cumulative maximum. + assert any(p.attributes["le"] == "+Inf" for p in bucket_points) + + count = metrics["lance_object_store_request_duration_seconds_count"] + assert sum(p.value for p in count.data.data_points) > 0 + + # The `_sum` instrument must also be wired and report positive latency. + duration_sum = metrics["lance_object_store_request_duration_seconds_sum"] + assert sum(p.value for p in duration_sum.data.data_points) > 0 + + # Unit handling: only `_sum` keeps the histogram's unit (seconds); `_bucket` + # and `_count` observe cumulative counts and are unitless. + assert duration_sum.unit == "s" + assert bucket.unit == "" + assert count.unit == "" + + +def test_snapshot_empty_before_install_is_safe(): + # snapshot is callable regardless of installation state and never raises. + from lancedb._lancedb import snapshot_lancedb_metrics + + assert isinstance(snapshot_lancedb_metrics(), list) + + +def test_instrument_warns_when_recorder_unavailable(monkeypatch): + # A foreign `metrics` recorder already installed -> register returns False; + # instrument_lancedb_metrics must warn and return False without instrumenting. + pytest.importorskip("opentelemetry.sdk.metrics") + import lancedb.otel as otel + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import InMemoryMetricReader + + monkeypatch.setattr(otel, "register_lancedb_metrics_recorder", lambda: False) + + reader = InMemoryMetricReader() + provider = MeterProvider(metric_readers=[reader]) + with pytest.warns(UserWarning, match="recorder"): + assert otel.instrument_lancedb_metrics(provider) is False diff --git a/python/uv.lock b/python/uv.lock index 760a6d264..c45b3b5ba 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -780,7 +780,7 @@ name = "cuda-bindings" version = "13.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cuda-pathfinder" }, + { name = "cuda-pathfinder", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/a9/21/8464d133752951c154feafb3b65c297e7d80f301183d220bec4c830f1441/cuda_bindings-13.3.1-cp310-cp310-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:120fcc53d57903df529c3486962c56528cba5b7d6c57c99537320ed9922c8b86", size = 6073403, upload-time = "2026-05-29T23:11:36.22Z" }, @@ -815,37 +815,37 @@ wheels = [ [package.optional-dependencies] cublas = [ - { name = "nvidia-cublas", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] cudart = [ - { name = "nvidia-cuda-runtime", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-runtime", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] cufft = [ - { name = "nvidia-cufft", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cufft", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] cufile = [ { name = "nvidia-cufile", marker = "sys_platform == 'linux'" }, ] cupti = [ - { name = "nvidia-cuda-cupti", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-cupti", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] curand = [ - { name = "nvidia-curand", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-curand", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] cusolver = [ - { name = "nvidia-cusolver", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cusolver", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] cusparse = [ - { name = "nvidia-cusparse", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cusparse", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] nvjitlink = [ - { name = "nvidia-nvjitlink", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] nvrtc = [ - { name = "nvidia-cuda-nvrtc", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-nvrtc", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] nvtx = [ - { name = "nvidia-nvtx", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-nvtx", marker = "(python_full_version < '3.14' and sys_platform == 'win32') or sys_platform == 'linux'" }, ] [[package]] @@ -1909,6 +1909,9 @@ embeddings = [ { name = "sentencepiece" }, { name = "torch" }, ] +otel = [ + { name = "opentelemetry-api" }, +] pylance = [ { name = "pylance" }, ] @@ -1923,6 +1926,7 @@ tests = [ { name = "boto3" }, { name = "datafusion" }, { name = "duckdb" }, + { name = "opentelemetry-sdk" }, { name = "pandas", version = "2.2.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "pandas", version = "2.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11' and python_full_version < '3.14'" }, { name = "pandas", version = "3.0.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.14'" }, @@ -1963,6 +1967,8 @@ requires-dist = [ { name = "open-clip-torch", marker = "extra == 'clip'" }, { name = "open-clip-torch", marker = "extra == 'embeddings'", specifier = ">=2.20.0" }, { name = "openai", marker = "extra == 'embeddings'", specifier = ">=1.6.1" }, + { name = "opentelemetry-api", marker = "extra == 'otel'" }, + { name = "opentelemetry-sdk", marker = "extra == 'tests'", specifier = ">=1.30.0" }, { name = "overrides", marker = "python_full_version < '3.12'", specifier = ">=0.7" }, { name = "packaging", specifier = ">=23.0" }, { name = "pandas", marker = "extra == 'tests'", specifier = ">=1.4" }, @@ -1994,7 +2000,7 @@ requires-dist = [ { name = "transformers", marker = "extra == 'siglip'", specifier = ">=4.41.0" }, { name = "typing-extensions", marker = "python_full_version < '3.11' and extra == 'dev'", specifier = ">=4.0.0" }, ] -provides-extras = ["azure", "clip", "dev", "docs", "embeddings", "pylance", "siglip", "tests"] +provides-extras = ["azure", "clip", "dev", "docs", "embeddings", "otel", "pylance", "siglip", "tests"] [[package]] name = "lomond" @@ -2775,7 +2781,7 @@ name = "nvidia-cudnn-cu13" version = "9.19.0.56" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas" }, + { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/f1/84/26025437c1e6b61a707442184fa0c03d083b661adf3a3eecfd6d21677740/nvidia_cudnn_cu13-9.19.0.56-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:6ed29ffaee1176c612daf442e4dd6cfeb6a0caa43ddcbeb59da94953030b1be4", size = 433781201, upload-time = "2026-02-03T20:40:53.805Z" }, @@ -2787,7 +2793,7 @@ name = "nvidia-cufft" version = "12.0.0.61" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink" }, + { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/8b/ae/f417a75c0259e85c1d2f83ca4e960289a5f814ed0cea74d18c353d3e989d/nvidia_cufft-12.0.0.61-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:2708c852ef8cd89d1d2068bdbece0aa188813a0c934db3779b9b1faa8442e5f5", size = 214053554, upload-time = "2025-09-04T08:31:38.196Z" }, @@ -2817,9 +2823,9 @@ name = "nvidia-cusolver" version = "12.0.4.66" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas" }, - { name = "nvidia-cusparse" }, - { name = "nvidia-nvjitlink" }, + { name = "nvidia-cublas", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, + { name = "nvidia-cusparse", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, + { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/c8/c3/b30c9e935fc01e3da443ec0116ed1b2a009bb867f5324d3f2d7e533e776b/nvidia_cusolver-12.0.4.66-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:02c2457eaa9e39de20f880f4bd8820e6a1cfb9f9a34f820eb12a155aa5bc92d2", size = 223467760, upload-time = "2025-09-04T08:33:04.222Z" }, @@ -2831,7 +2837,7 @@ name = "nvidia-cusparse" version = "12.6.3.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink" }, + { name = "nvidia-nvjitlink", marker = "(python_full_version < '3.14' and sys_platform == 'emscripten') or (python_full_version < '3.14' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/f8/94/5c26f33738ae35276672f12615a64bd008ed5be6d1ebcb23579285d960a9/nvidia_cusparse-12.6.3.3-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:80bcc4662f23f1054ee334a15c72b8940402975e0eab63178fc7e670aa59472c", size = 162155568, upload-time = "2025-09-04T08:33:42.864Z" }, @@ -2934,6 +2940,45 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f6/46/180e14be801a75bc13f234cb1b594b232adeb9c84e60a9ab1832e8333591/openai-2.40.0-py3-none-any.whl", hash = "sha256:2b205637ff214477f9ce9ab035e9f494db0e3fa8f1e599008953735fbf6ff1ff", size = 1350935, upload-time = "2026-06-01T21:48:21.462Z" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.43.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ae/cc/e4c9584181f86494df0f6bdec1a4f3280c50db44704dc2a407e994fc87bb/opentelemetry_api-1.43.0.tar.gz", hash = "sha256:107d0d03857ea8fc7c5fcbbbd83f800c281f0d560553d61c1d675fccfd1761c1", size = 73476, upload-time = "2026-06-24T15:19:55.323Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/17/83/6dba32b85f31868400440dc7ad2ca1eab94cbbf3a7b0459ed39f8311a9e2/opentelemetry_api-1.43.0-py3-none-any.whl", hash = "sha256:20acf45e9b21851926835292e4045d290acade1edd2ff3de86d2f069687ba1fd", size = 61912, upload-time = "2026-06-24T15:19:35.434Z" }, +] + +[[package]] +name = "opentelemetry-sdk" +version = "1.43.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "opentelemetry-semantic-conventions" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/3e/eb/5041074274ac0956b03637cc039d434569112468e875eddfcc9a0674ce06/opentelemetry_sdk-1.43.0.tar.gz", hash = "sha256:d8187c81c162df9913e4003dd6485f7390d9a24fc17026ec7387b8b8218b08e9", size = 254744, upload-time = "2026-06-24T15:20:08.467Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/49/e3/b17be23af124201c9f52eececd4cc8ddfed1597d37b4ee771895d325805c/opentelemetry_sdk-1.43.0-py3-none-any.whl", hash = "sha256:d1323a547c1ce69d6a069a17a44b7da82bb8b332051ecb074041f87642c86823", size = 178852, upload-time = "2026-06-24T15:19:52.169Z" }, +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.64b0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "opentelemetry-api" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/5a/30/5f26df29509eccd86b99b481ac9ffa39da49ba9577cc69071c552ae30447/opentelemetry_semantic_conventions-0.64b0.tar.gz", hash = "sha256:72f76fb2d1582d9d033dd1fcd84532e961e6ff3d90d24ba6fabc72975a83864c", size = 148340, upload-time = "2026-06-24T15:20:09.267Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f2/ca/23ba87a221b574a7c5a99d48849d80bfe8b047624681357e2b002e566187/opentelemetry_semantic_conventions-0.64b0-py3-none-any.whl", hash = "sha256:ea77e85e354b8f604ddbe5f3d9135216f982fa4d77e5859ac30f6d8a50505aa6", size = 203713, upload-time = "2026-06-24T15:19:53.339Z" }, +] + [[package]] name = "overrides" version = "7.7.0" diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index 59c1ec6e4..bcc922261 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -49,6 +49,8 @@ lance-encoding = { workspace = true } lance-arrow = { workspace = true } lance-namespace = { workspace = true } lance-namespace-impls = { workspace = true } +metrics = { workspace = true, optional = true } +metrics-util = { workspace = true, optional = true } moka = { workspace = true } pin-project = { workspace = true } tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] } @@ -147,6 +149,15 @@ remote = [ "lance-namespace-impls/rest", "lance-namespace-impls/rest-adapter", ] +# Publish LanceDB's internal metrics (currently object store request counts, +# bytes, latency, errors, and throttles) through the `metrics` crate facade, +# and re-export the `metrics` crate as `lancedb::metrics`. Install any +# `metrics`-compatible recorder to collect them. +metrics = ["dep:metrics", "lance/metrics", "lance-io/metrics"] +# Additional adapter on top of `metrics` that installs a process-global recorder +# and exposes a pull-based snapshot/catalog API (see `lancedb::metrics_otel`) +# for bridging metrics into OpenTelemetry or other pull-based exporters. +metrics-otel = ["metrics", "dep:metrics-util"] fp16kernels = ["lance-linalg/fp16kernels"] s3-test = [] bedrock = ["dep:aws-sdk-bedrockruntime"] diff --git a/rust/lancedb/src/lib.rs b/rust/lancedb/src/lib.rs index 2aa4ab77c..f4100ae84 100644 --- a/rust/lancedb/src/lib.rs +++ b/rust/lancedb/src/lib.rs @@ -33,6 +33,11 @@ //! - `remote` - Enable remote client to connect to LanceDB cloud. //! - `huggingface` - Enable HuggingFace Hub integration for loading datasets from the Hub. //! - `fp16kernels` - Enable FP16 kernels for faster vector search on CPU. +//! - `metrics` - Publish LanceDB's internal metrics through the +//! [`metrics`](https://docs.rs/metrics) crate facade and re-export that crate. +//! Install any `metrics`-compatible recorder to collect them. +//! - `metrics-otel` - Add a pull-based adapter (the `metrics_otel` module) over +//! the `metrics` facade for bridging metrics into OpenTelemetry or similar. //! //! ### Quick Start //! @@ -174,6 +179,8 @@ pub mod expr; pub mod index; pub mod io; pub mod ipc; +#[cfg(feature = "metrics-otel")] +pub mod metrics_otel; #[cfg(feature = "polars")] mod polars_arrow_convertors; pub mod query; @@ -194,6 +201,12 @@ pub use connection::{ConnectNamespaceBuilder, Connection}; pub use error::{Error, Result}; use lance_index::vector::ApproxMode as LanceApproxMode; use lance_linalg::distance::DistanceType as LanceDistanceType; +/// Re-export of the [`metrics`](https://docs.rs/metrics) crate facade. Enable +/// the `metrics` feature to publish LanceDB's internal metrics; install any +/// `metrics`-compatible recorder to collect them. See also [`metrics_otel`] for +/// a built-in pull-based adapter. +#[cfg(feature = "metrics")] +pub use metrics; pub use table::Table; #[derive(Debug, Copy, Clone, PartialEq, Serialize, Deserialize, Default)] diff --git a/rust/lancedb/src/metrics_otel.rs b/rust/lancedb/src/metrics_otel.rs new file mode 100644 index 000000000..85ecb3af2 --- /dev/null +++ b/rust/lancedb/src/metrics_otel.rs @@ -0,0 +1,594 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! A pull-based adapter over the [`metrics`] crate facade. +//! +//! LanceDB (through Lance core) publishes metrics — currently object store +//! request counts, bytes, latency, errors, and throttles — through the global +//! [`metrics`] facade without choosing a backend. This module installs a +//! process-global [`metrics::Recorder`] that aggregates those metrics into +//! lock-free cumulative storage and exposes that state as a snapshot, so callers +//! can feed it into a pull-based exporter such as OpenTelemetry. +//! +//! The language bindings build their OpenTelemetry integrations on top of the +//! three public entry points here: [`register_metrics_recorder`], +//! [`metrics_catalog`], and [`snapshot_metrics`]. +//! +//! The recorder is *generic*: it records any metric emitted through the facade, +//! keyed by name and labels. Object store metrics are the first producer, but +//! nothing here is specific to them. New metrics flow through automatically; +//! they only need to be described (via the `metrics` `describe_*!` macros) so +//! callers can discover their name, kind, and unit up front. +//! +//! ## Why pull, not push +//! +//! OpenTelemetry collects on its own schedule and invokes observable-instrument +//! callbacks at collection time. Cumulative counters map directly onto OTel's +//! `ObservableCounter` semantics. So the adapter aggregates in Rust and lets the +//! collection thread pull a [`snapshot`](snapshot_metrics) on demand. +//! +//! ## Histograms +//! +//! OpenTelemetry has no asynchronous histogram instrument, so histograms cannot +//! be pulled as-is. Instead each histogram is aggregated into fixed buckets +//! (Prometheus style) and exposed as cumulative `le` bucket counts plus a count +//! and sum, which the caller can surface as observable counters. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, LazyLock, Mutex, OnceLock, RwLock}; + +use metrics::{Counter, Gauge, Histogram, Key, KeyName, Metadata, Recorder, SharedString, Unit}; +use metrics_util::registry::{Registry, Storage}; + +/// Bucket boundaries used when a histogram has no registered bounds. Covers a +/// broad latency range so unknown histograms still produce useful buckets. +const DEFAULT_BOUNDS: &[f64] = &[ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, +]; + +/// The kind of a metric, mirroring the three `metrics` instrument types. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetricKind { + Counter, + Gauge, + Histogram, +} + +impl MetricKind { + /// The lowercase name of this kind (`"counter"`, `"gauge"`, `"histogram"`). + pub fn as_str(self) -> &'static str { + match self { + Self::Counter => "counter", + Self::Gauge => "gauge", + Self::Histogram => "histogram", + } + } +} + +/// A described metric, used to create one exporter instrument per metric up front. +#[derive(Debug, Clone)] +pub struct MetricDescription { + /// The metric name (e.g. `lance_object_store_requests_total`). + pub name: String, + /// Whether the metric is a counter, gauge, or histogram. + pub kind: MetricKind, + /// The canonical unit label, if the producer described one. + pub unit: Option, + /// Human-readable help text describing the metric. + pub description: String, +} + +/// The aggregated value of a metric at snapshot time. +#[derive(Debug, Clone)] +pub enum MetricValue { + /// A counter or gauge value. + Scalar(f64), + /// A histogram, decomposed into cumulative `le` buckets plus count and sum. + Histogram { + /// Cumulative `(le, count)` buckets, ending in the implicit `+Inf` bucket. + buckets: Vec<(String, u64)>, + /// Total number of recorded samples. + count: u64, + /// Sum of all recorded sample values. + sum: f64, + }, +} + +/// One aggregated metric data point exposed to a caller. +#[derive(Debug, Clone)] +pub struct MetricPoint { + /// The metric name. + pub name: String, + /// Whether the point is a counter, gauge, or histogram. + pub kind: MetricKind, + /// The label set for this point (e.g. `operation`, `base`). + pub attributes: HashMap, + /// The aggregated value. + pub value: MetricValue, +} + +/// Catalog of described metrics, keyed by metric name. +static CATALOG: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +struct CatalogEntry { + kind: MetricKind, + unit: Option, + description: String, +} + +/// Per-metric histogram bucket boundaries, keyed by metric name. Producers +/// register their recommended bounds before any metric is recorded. +static HISTOGRAM_BOUNDS: LazyLock>>> = + LazyLock::new(|| RwLock::new(HashMap::new())); + +/// The installed recorder's registry, available once installation succeeds. +static REGISTRY: OnceLock>> = OnceLock::new(); + +fn bounds_for(name: &str) -> Arc<[f64]> { + HISTOGRAM_BOUNDS + .read() + .unwrap() + .get(name) + .cloned() + .unwrap_or_else(|| Arc::from(DEFAULT_BOUNDS)) +} + +/// A histogram that buckets samples at record time into fixed boundaries, +/// keeping a cumulative count and sum. Bucketing eagerly keeps memory bounded +/// (unlike retaining raw samples) and produces Prometheus-style `le` buckets. +struct BucketedHistogram { + /// Sorted, finite upper bounds. A sample `v` falls in the first bucket whose + /// bound is `>= v`; samples above all bounds fall in the implicit `+Inf` + /// bucket stored as the final entry of `counts`. + bounds: Arc<[f64]>, + /// Per-bucket (non-cumulative) counts; length is `bounds.len() + 1`. + counts: Box<[AtomicU64]>, + count: AtomicU64, + /// Running sum of recorded values, stored as `f64` bits (there is no atomic + /// f64, so the bit pattern is held in a `u64`; see [`Self::add_to_sum`]). + sum_bits: AtomicU64, +} + +// All atomics here use `Ordering::Relaxed`: each metric counter is independent, +// so no happens-before relationship is needed between them, and a snapshot +// reader tolerates slightly stale values. This matches `metrics_util`'s +// `AtomicStorage`. + +impl BucketedHistogram { + fn new(bounds: Arc<[f64]>) -> Self { + let counts = (0..bounds.len() + 1) + .map(|_| AtomicU64::new(0)) + .collect::>() + .into_boxed_slice(); + Self { + bounds, + counts, + count: AtomicU64::new(0), + sum_bits: AtomicU64::new(0), + } + } + + fn add_to_sum(&self, value: f64) { + // No atomic offers an f64 add, so read the current bit pattern, add in + // float space, and CAS it back, retrying if another thread won the race. + let mut current = self.sum_bits.load(Ordering::Relaxed); + loop { + let updated = (f64::from_bits(current) + value).to_bits(); + match self.sum_bits.compare_exchange_weak( + current, + updated, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(actual) => current = actual, + } + } + } + + /// Cumulative `le` buckets, total count, and sum at this instant. + fn snapshot(&self) -> MetricValue { + let mut cumulative = 0u64; + let mut buckets = Vec::with_capacity(self.bounds.len() + 1); + for (i, bound) in self.bounds.iter().enumerate() { + cumulative += self.counts[i].load(Ordering::Relaxed); + buckets.push((format!("{}", bound), cumulative)); + } + cumulative += self.counts[self.bounds.len()].load(Ordering::Relaxed); + buckets.push(("+Inf".to_string(), cumulative)); + MetricValue::Histogram { + buckets, + count: self.count.load(Ordering::Relaxed), + sum: f64::from_bits(self.sum_bits.load(Ordering::Relaxed)), + } + } +} + +impl metrics::HistogramFn for BucketedHistogram { + fn record(&self, value: f64) { + let idx = self.bounds.partition_point(|&bound| bound < value); + self.counts[idx].fetch_add(1, Ordering::Relaxed); + self.count.fetch_add(1, Ordering::Relaxed); + self.add_to_sum(value); + } +} + +/// Storage backing the registry. Counters and gauges are plain atomics (as in +/// `metrics_util`'s `AtomicStorage`); histograms use [`BucketedHistogram`]. +struct LanceStorage; + +impl Storage for LanceStorage { + type Counter = Arc; + type Gauge = Arc; + type Histogram = Arc; + + fn counter(&self, _key: &Key) -> Self::Counter { + Arc::new(AtomicU64::new(0)) + } + + fn gauge(&self, _key: &Key) -> Self::Gauge { + // The `metrics` facade writes the f64 bit pattern into this `u64` (the + // snapshot decodes it with `f64::from_bits`), matching `AtomicStorage`. + // `0` decodes to `0.0`, the correct initial value. + Arc::new(AtomicU64::new(0)) + } + + fn histogram(&self, key: &Key) -> Self::Histogram { + Arc::new(BucketedHistogram::new(bounds_for(key.name()))) + } +} + +struct LanceRecorder { + registry: Arc>, +} + +impl LanceRecorder { + fn describe( + &self, + key: KeyName, + kind: MetricKind, + unit: Option, + description: SharedString, + ) { + CATALOG.lock().unwrap().insert( + key.as_str().to_string(), + CatalogEntry { + kind, + unit: unit.map(|u| u.as_canonical_label().to_string()), + description: description.into_owned(), + }, + ); + } +} + +impl Recorder for LanceRecorder { + fn describe_counter(&self, key: KeyName, unit: Option, description: SharedString) { + self.describe(key, MetricKind::Counter, unit, description); + } + + fn describe_gauge(&self, key: KeyName, unit: Option, description: SharedString) { + self.describe(key, MetricKind::Gauge, unit, description); + } + + fn describe_histogram(&self, key: KeyName, unit: Option, description: SharedString) { + self.describe(key, MetricKind::Histogram, unit, description); + } + + fn register_counter(&self, key: &Key, _metadata: &Metadata<'_>) -> Counter { + self.registry + .get_or_create_counter(key, |c| Counter::from_arc(c.clone())) + } + + fn register_gauge(&self, key: &Key, _metadata: &Metadata<'_>) -> Gauge { + self.registry + .get_or_create_gauge(key, |g| Gauge::from_arc(g.clone())) + } + + fn register_histogram(&self, key: &Key, _metadata: &Metadata<'_>) -> Histogram { + self.registry + .get_or_create_histogram(key, |h| Histogram::from_arc(h.clone())) + } +} + +/// Register the recommended histogram bounds for every metric-emitting +/// subsystem. New subsystems add their `histogram_bounds()` here. +fn register_bounds() { + let mut bounds = HISTOGRAM_BOUNDS.write().unwrap(); + for (name, values) in lance_io::object_store::metrics::histogram_bounds() { + bounds.insert((*name).to_string(), Arc::from(*values)); + } +} + +/// Describe every metric-emitting subsystem so the catalog is populated. Must +/// run after the recorder is installed. New subsystems add their +/// `describe_metrics()` here. +fn describe_all() { + lance_io::object_store::metrics::describe_metrics(); +} + +fn labels(key: &Key) -> HashMap { + key.labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect() +} + +fn collect_points(registry: &Registry) -> Vec { + let mut points = Vec::new(); + for (key, handle) in registry.get_counter_handles() { + points.push(MetricPoint { + name: key.name().to_string(), + kind: MetricKind::Counter, + attributes: labels(&key), + // OpenTelemetry observations are float; counts stay well within the + // f64-exact integer range (2^53), so this cast is lossless in practice. + value: MetricValue::Scalar(handle.load(Ordering::Relaxed) as f64), + }); + } + for (key, handle) in registry.get_gauge_handles() { + points.push(MetricPoint { + name: key.name().to_string(), + kind: MetricKind::Gauge, + attributes: labels(&key), + value: MetricValue::Scalar(f64::from_bits(handle.load(Ordering::Relaxed))), + }); + } + for (key, handle) in registry.get_histogram_handles() { + points.push(MetricPoint { + name: key.name().to_string(), + kind: MetricKind::Histogram, + attributes: labels(&key), + value: handle.snapshot(), + }); + } + points +} + +/// Install the LanceDB metrics recorder as the process-global `metrics` recorder. +/// +/// Returns `true` if the recorder is installed (now or previously). Returns +/// `false` if a *different* recorder is already installed — `metrics` allows +/// only one global recorder per process, so LanceDB cannot coexist with another. +pub fn register_metrics_recorder() -> bool { + if REGISTRY.get().is_some() { + return true; + } + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { + registry: registry.clone(), + }; + // Register bounds *before* installing the recorder. Bounds don't depend on + // the recorder, and once it is installed a concurrent histogram emission + // could otherwise create a handle with the fallback bounds and keep them for + // the process lifetime. + register_bounds(); + match metrics::set_global_recorder(recorder) { + Ok(()) => { + let _ = REGISTRY.set(registry); + // Describe metrics only after install so the `describe_*!` macros + // route through this recorder and populate the catalog. + describe_all(); + true + } + Err(_) => false, + } +} + +/// The catalog of described LanceDB metrics. Empty until the recorder is installed. +pub fn metrics_catalog() -> Vec { + CATALOG + .lock() + .unwrap() + .iter() + .map(|(name, entry)| MetricDescription { + name: name.clone(), + kind: entry.kind, + unit: entry.unit.clone(), + description: entry.description.clone(), + }) + .collect() +} + +/// A point-in-time snapshot of every recorded metric. Empty until the recorder +/// is installed. The read is lock-free. +pub fn snapshot_metrics() -> Vec { + let Some(registry) = REGISTRY.get() else { + return Vec::new(); + }; + collect_points(registry) +} + +#[cfg(test)] +mod tests { + use super::*; + use metrics::HistogramFn; + + fn bucket_count(buckets: &[(String, u64)], le: &str) -> u64 { + buckets + .iter() + .find(|(b, _)| b == le) + .map(|(_, c)| *c) + .unwrap_or_else(|| panic!("no bucket with le={le}")) + } + + #[test] + fn bucketed_histogram_records_cumulative_buckets() { + let hist = BucketedHistogram::new(Arc::from([0.1f64, 1.0, 10.0].as_slice())); + hist.record(0.05); // le=0.1 + hist.record(0.5); // le=1 + hist.record(0.5); // le=1 + hist.record(50.0); // +Inf + + let MetricValue::Histogram { + buckets, + count, + sum, + } = hist.snapshot() + else { + panic!("expected histogram"); + }; + + // Buckets are cumulative (Prometheus `le` semantics). + assert_eq!(bucket_count(&buckets, "0.1"), 1); + assert_eq!(bucket_count(&buckets, "1"), 3); + assert_eq!(bucket_count(&buckets, "10"), 3); + assert_eq!(bucket_count(&buckets, "+Inf"), 4); + assert_eq!(count, 4); + assert!((sum - 51.05).abs() < 1e-9); + } + + #[test] + fn bucketed_histogram_boundary_is_inclusive() { + let hist = BucketedHistogram::new(Arc::from([1.0f64].as_slice())); + hist.record(1.0); // exactly the bound -> le=1, not +Inf + let MetricValue::Histogram { buckets, .. } = hist.snapshot() else { + panic!("expected histogram"); + }; + assert_eq!(bucket_count(&buckets, "1"), 1); + assert_eq!(bucket_count(&buckets, "+Inf"), 1); + } + + #[test] + fn bucketed_histogram_boundary_is_inclusive_mid_range() { + // A value equal to a middle bound lands in that bucket, not the next. + let hist = BucketedHistogram::new(Arc::from([0.1f64, 1.0, 10.0].as_slice())); + hist.record(1.0); + let MetricValue::Histogram { buckets, .. } = hist.snapshot() else { + panic!("expected histogram"); + }; + assert_eq!(bucket_count(&buckets, "0.1"), 0); + assert_eq!(bucket_count(&buckets, "1"), 1); + assert_eq!(bucket_count(&buckets, "10"), 1); // cumulative, so still 1 + assert_eq!(bucket_count(&buckets, "+Inf"), 1); + } + + #[test] + fn recorder_aggregates_counters_with_labels() { + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { + registry: registry.clone(), + }; + metrics::with_local_recorder(&recorder, || { + metrics::counter!("test_requests_total", "operation" => "get", "scheme" => "s3") + .increment(2); + metrics::counter!("test_requests_total", "operation" => "get", "scheme" => "s3") + .increment(3); + // A distinct label set must produce a separate point, not merge. + metrics::counter!("test_requests_total", "operation" => "put", "scheme" => "gs") + .increment(7); + }); + + let scalar = |attrs: &[(&str, &str)]| { + let points = collect_points(®istry); + let point = points + .into_iter() + .find(|p| { + p.name == "test_requests_total" + && attrs + .iter() + .all(|(k, v)| p.attributes.get(*k).map(String::as_str) == Some(*v)) + }) + .expect("counter recorded for label set"); + assert_eq!(point.kind, MetricKind::Counter); + match point.value { + MetricValue::Scalar(v) => v, + _ => panic!("expected scalar"), + } + }; + + // Same labels aggregate; distinct labels stay separate. + assert!((scalar(&[("operation", "get"), ("scheme", "s3")]) - 5.0).abs() < 1e-9); + assert!((scalar(&[("operation", "put"), ("scheme", "gs")]) - 7.0).abs() < 1e-9); + } + + #[test] + fn recorder_records_gauges() { + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { + registry: registry.clone(), + }; + // Gauges store the f64 bit pattern in a u64; the snapshot must decode it. + metrics::with_local_recorder(&recorder, || { + metrics::gauge!("test_gauge", "scheme" => "s3").set(3.5); + }); + + let points = collect_points(®istry); + let point = points + .iter() + .find(|p| p.name == "test_gauge") + .expect("gauge recorded"); + assert_eq!(point.kind, MetricKind::Gauge); + assert!(matches!(point.value, MetricValue::Scalar(v) if (v - 3.5).abs() < 1e-9)); + } + + #[test] + fn recorder_falls_back_to_default_bounds() { + // A histogram with no registered bounds uses DEFAULT_BOUNDS. + let name = "test_unregistered_histogram"; + assert!(!HISTOGRAM_BOUNDS.read().unwrap().contains_key(name)); + + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { + registry: registry.clone(), + }; + metrics::with_local_recorder(&recorder, || { + metrics::histogram!(name).record(0.02); + }); + + let points = collect_points(®istry); + let point = points.iter().find(|p| p.name == name).expect("recorded"); + let MetricValue::Histogram { buckets, count, .. } = &point.value else { + panic!("expected histogram"); + }; + assert_eq!(*count, 1); + // DEFAULT_BOUNDS yields one bucket per bound plus the implicit `+Inf`. + assert_eq!(buckets.len(), DEFAULT_BOUNDS.len() + 1); + // 0.02 falls in the le=0.025 bucket (the third DEFAULT_BOUNDS entry). + assert_eq!(bucket_count(buckets, "0.025"), 1); + assert_eq!(bucket_count(buckets, "0.01"), 0); + assert_eq!(bucket_count(buckets, "+Inf"), 1); + } + + #[test] + fn recorder_uses_registered_histogram_bounds() { + let name = "test_recorder_bounds_seconds"; + HISTOGRAM_BOUNDS + .write() + .unwrap() + .insert(name.to_string(), Arc::from([0.1f64, 1.0].as_slice())); + + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { + registry: registry.clone(), + }; + metrics::with_local_recorder(&recorder, || { + metrics::histogram!(name).record(0.05); + metrics::histogram!(name).record(5.0); + }); + + let points = collect_points(®istry); + let point = points.iter().find(|p| p.name == name).expect("recorded"); + let MetricValue::Histogram { buckets, count, .. } = &point.value else { + panic!("expected histogram"); + }; + assert_eq!(*count, 2); + assert_eq!(bucket_count(buckets, "0.1"), 1); + assert_eq!(bucket_count(buckets, "+Inf"), 2); + } + + #[test] + fn describe_populates_catalog() { + let name = "test_describe_catalog_total"; + let registry = Arc::new(Registry::new(LanceStorage)); + let recorder = LanceRecorder { registry }; + metrics::with_local_recorder(&recorder, || { + metrics::describe_counter!(name, Unit::Count, "a test counter"); + }); + + let catalog = CATALOG.lock().unwrap(); + let entry = catalog.get(name).expect("described"); + assert_eq!(entry.kind, MetricKind::Counter); + assert_eq!(entry.description, "a test counter"); + } +}