diff --git a/CHANGELOG.md b/CHANGELOG.md index ac5a8d3d..0d051827 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,11 @@ project adheres to [Semantic Versioning](http://semver.org/). ### Added +- Opt-in native histograms with configurable exponential buckets, a zero bucket, + bucket-count limits, exemplars, and worker/cluster aggregation. +- Prometheus protobuf registries for native and classic metrics, with public + content-type constants and TypeScript support for binary output. + ## [0.16.0] - 2026-08-24 This release marks our first release as a Prometheus subproject. diff --git a/README.md b/README.md index fa0fa14c..26a5f066 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,39 @@ xhrRequest(function (err, res) { }); ``` +##### Native histograms + +Enable native buckets with `nativeHistogramBucketFactor` and expose the registry +using Prometheus protobuf: + +```js +const registry = new client.Registry( + client.Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE, +); +const histogram = new client.Histogram({ + name: 'request_duration_seconds', + help: 'Time spent handling requests', + nativeHistogramBucketFactor: 1.1, + buckets: [], + registers: [registry], +}); +histogram.observe(0.125); + +res.setHeader('Content-Type', registry.contentType); +res.end(await registry.metrics()); // A Buffer for protobuf registries +``` + +Native buckets cover positive and negative values using exponential buckets and +a zero bucket. The default zero threshold is `2 ** -128`, configurable with +`nativeHistogramZeroThreshold`. The default budget of 160 populated buckets per +label set can be configured with `nativeHistogramMaxBucketNumber` (0 disables +the budget). When needed, resolution is reduced down to schema -4; at that +minimum resolution the budget is a soft limit. + +Classic buckets are retained by default. Set `buckets: []` for native-only +protobuf output. Prometheus text and OpenMetrics 1.0 text expose only the classic +representation. Prometheus must also be configured to scrape native histograms. + #### Summary Summaries calculate percentiles of observed values. @@ -397,11 +430,12 @@ enabled. They get a single object with the format `{labels, value, exemplarLabels}`. When using exemplars, the registry used for metrics should be set to OpenMetrics -type (including the global or default registry if no registries are specified). +or Prometheus protobuf (including the global or default registry if no registries +are specified). ### Registry type -The library supports both the old Prometheus format and the OpenMetrics format. +The library supports Prometheus text, OpenMetrics text, and Prometheus protobuf. The format can be set per registry. For default metrics: ```js @@ -419,9 +453,14 @@ this is currently the default registry type. **OPENMETRICS_CONTENT_TYPE** - defaults to version 1.0.0 of the [OpenMetrics standard](https://github.com/OpenObservability/OpenMetrics/blob/d99b705f611b75fec8f450b05e344e02eea6921d/specification/OpenMetrics.md). +**PROMETHEUS_PROTOBUF_CONTENT_TYPE** - length-delimited Prometheus protobuf, +including native histograms. Registry serialization methods return a `Buffer` +for this format. + The HTTP Content-Type string for each registry type is exposed both at module -level (`prometheusContentType` and `openMetricsContentType`) and as static -properties on the `Registry` object. +level (`prometheusContentType`, `openMetricsContentType`, and +`prometheusProtobufContentType`) and as static properties on the `Registry` +object. The `contentType` constant exposed by the module returns the default content type when creating a new registry, currently defaults to Prometheus type. diff --git a/example/native-histogram.js b/example/native-histogram.js new file mode 100644 index 00000000..1fb3be14 --- /dev/null +++ b/example/native-histogram.js @@ -0,0 +1,54 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const http = require('node:http'); +const client = require('../index'); + +const registry = new client.Registry( + client.Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE, +); +client.collectDefaultMetrics({ register: registry }); + +const duration = new client.Histogram({ + name: 'http_request_duration_seconds', + help: 'Time spent handling requests', + labelNames: ['method'], + nativeHistogramBucketFactor: 1.1, + nativeHistogramMaxBucketNumber: 160, + buckets: [], + registers: [registry], +}); + +http + .createServer(async (req, res) => { + if (req.url === '/metrics') { + try { + const metrics = await registry.metrics(); + res.writeHead(200, { 'Content-Type': registry.contentType }); + res.end(metrics); + } catch (error) { + res.writeHead(500); + res.end(error.message); + } + return; + } + + const end = duration.startTimer({ method: req.method }); + res.writeHead(204); + res.end(); + end(); + }) + .listen(Number(process.env.PORT ?? 3000)); diff --git a/index.d.ts b/index.d.ts index 6e60e8bf..c8ea55ff 100644 --- a/index.d.ts +++ b/index.d.ts @@ -28,24 +28,32 @@ export type OpenMetricsContentType = export type PrometheusContentType = `${PrometheusMIME}; version=${PrometheusMetricsVersion}; charset=${Charset}`; +export type PrometheusProtobufContentType = + 'application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; encoding=delimited'; + export type RegistryContentType = - PrometheusContentType | OpenMetricsContentType; + | PrometheusContentType + | OpenMetricsContentType + | PrometheusProtobufContentType; + +export type RegistryMetrics = + T extends PrometheusProtobufContentType ? Uint8Array : string; /** * Container for all registered metrics */ export class Registry< - BoundRegistryContentType extends RegistryContentType = RegistryContentType, + BoundRegistryContentType extends RegistryContentType = PrometheusContentType, > { /** * @param regContentType The content type of the registry */ - constructor(regContentType?: RegistryContentType); + constructor(regContentType?: BoundRegistryContentType); /** - * Get string representation for all metrics + * Get metrics as text, or a Buffer for a Prometheus protobuf registry */ - metrics(): Promise; + metrics(): Promise>; /** * Remove all metrics from the registry @@ -77,8 +85,8 @@ export class Registry< ): Promise>[]>; /** - * Get string representation for a metric - * @param metric Metric to convert to a string + * Get a metric as text + * @param metric Metric to serialize */ getMetricsAsString(metric: Metric): Promise; @@ -113,7 +121,7 @@ export class Registry< setDefaultLabels(labels: object): void; /** - * Get a string representation of a single metric by name + * Get a single metric as text * @param name The name of the metric */ getSingleMetricAsString(name: string): Promise; @@ -121,22 +129,22 @@ export class Registry< /** * Gets the Content-Type of the metrics for use in the response headers. */ - readonly contentType: PrometheusContentType | OpenMetricsContentType; + readonly contentType: RegistryContentType; /** - * Set the content type of a registry. Used to change between Prometheus and - * OpenMetrics versions. + * Set the content type of a registry. The returned registry has the new + * serialization return type; use it when switching between text and protobuf. * @param contentType The type of the registry */ - setContentType( - contentType: PrometheusContentType | OpenMetricsContentType, - ): void; + setContentType(contentType: T): Registry; /** * Merge registers * @param registers The registers you want to merge together */ - static merge(registers: Registry[]): Registry; + static merge( + registers: Registry[], + ): Registry; /** * Creates a new Registry instance from an array of metrics that were @@ -147,9 +155,10 @@ export class Registry< * `registry.getMetricsAsJSON()`. * @returns {Registry} aggregated registry. */ - static aggregate( + static aggregate( metricsArr: Array, - ): Registry; // TODO Promise? + registryType?: T, + ): Registry; /** * HTTP Prometheus Content-Type for metrics response headers. @@ -160,6 +169,11 @@ export class Registry< * HTTP OpenMetrics Content-Type for metrics response headers. */ static OPENMETRICS_CONTENT_TYPE: OpenMetricsContentType; + + /** + * HTTP Content-Type for length-delimited Prometheus protobuf metrics. + */ + static PROMETHEUS_PROTOBUF_CONTENT_TYPE: PrometheusProtobufContentType; } export type Collector = () => void; @@ -184,15 +198,23 @@ export const prometheusContentType: PrometheusContentType; */ export const openMetricsContentType: OpenMetricsContentType; +/** + * HTTP Content-Type for length-delimited Prometheus protobuf metrics. + */ +export const prometheusProtobufContentType: PrometheusProtobufContentType; + export class ClusterRegistry< - T extends RegistryContentType, + T extends RegistryContentType = PrometheusContentType, > extends Registry { + setContentType( + contentType: U, + ): ClusterRegistry; + /** * Gets aggregated metrics for all workers. - * @returns {Promise} Promise that resolves with the aggregated - * metrics. + * @returns Promise that resolves with text or protobuf bytes. */ - clusterMetrics(): Promise; + clusterMetrics(): Promise>; /** * Sets the registry or registries to be aggregated. Call from workers to @@ -202,22 +224,22 @@ export class ClusterRegistry< * @returns {void} */ static setRegistries( - regs: - | Array< - Registry | Registry - > - | Registry - | Registry, + regs: Registry[] | Registry, ): void; } -export class WorkerRegistry extends Registry { +export class WorkerRegistry< + T extends RegistryContentType = PrometheusContentType, +> extends Registry { + setContentType( + contentType: U, + ): WorkerRegistry; + /** * Gets aggregated metrics for all workers. - * @returns {Promise} Promise that resolves with the aggregated - * metrics. + * @returns Promise that resolves with text or protobuf bytes. */ - workerMetrics(): Promise; + workerMetrics(): Promise>; /** * Orderly shutdown of the registry. @@ -238,12 +260,7 @@ export class WorkerRegistry extends Registry { * @returns {void} */ static setRegistries( - regs: - | Array< - Registry | Registry - > - | Registry - | Registry, + regs: Registry[] | Registry, ): void; } @@ -251,14 +268,17 @@ export class WorkerRegistry extends Registry { * @deprecated */ export class AggregatorRegistry< - T extends RegistryContentType, + T extends RegistryContentType = PrometheusContentType, > extends Registry { + setContentType( + contentType: U, + ): AggregatorRegistry; + /** * Gets aggregated metrics for all workers. - * @returns {Promise} Promise that resolves with the aggregated - * metrics. + * @returns Promise that resolves with text or protobuf bytes. */ - clusterMetrics(): Promise; + clusterMetrics(): Promise>; /** * Orderly shutdown of the registry. @@ -279,12 +299,7 @@ export class AggregatorRegistry< * @returns {void} */ static setRegistries( - regs: - | Array< - Registry | Registry - > - | Registry - | Registry, + regs: Registry[] | Registry, ): void; } @@ -321,6 +336,36 @@ export interface MetricObjectWithValues< T extends MetricValue, > extends MetricObject { values: T[]; + nativeHistograms?: NativeHistogramValue[]; +} + +export interface NativeHistogramSpan { + offset: number; + length: number; +} + +export interface NativeHistogramExemplar { + labelSet: Record; + value: number; + /** Unix timestamp in seconds. */ + timestamp: number; +} + +/** A snapshot of one label set's native histogram, suitable for JSON and IPC. */ +export interface NativeHistogramValue { + labels: LabelValues; + count: number; + sum: number; + schema: number; + zeroThreshold: number; + zeroCount: number; + positiveSpans: NativeHistogramSpan[]; + positiveDeltas: number[]; + negativeSpans: NativeHistogramSpan[]; + negativeDeltas: number[]; + /** Unix timestamp in seconds. */ + createdTimestamp: number; + exemplars: NativeHistogramExemplar[]; } export type MetricValue = { @@ -340,9 +385,7 @@ interface MetricConfiguration { name: string; help: string; labelNames?: T[] | readonly T[]; - registers?: ( - Registry | Registry - )[]; + registers?: Registry[]; aggregator?: Aggregator; collect?: CollectFunction; enableExemplars?: boolean; @@ -363,7 +406,7 @@ export interface IncreaseDataWithExemplar { export interface ObserveDataWithExemplar { value: number; labels?: LabelValues; - exemplarLabels?: LabelValues; + exemplarLabels?: Record; } /** @@ -582,9 +625,23 @@ export namespace Gauge { } } -export interface HistogramConfiguration< - T extends string, -> extends MetricConfiguration { +export interface NativeHistogramConfiguration { + /** + * Enable native buckets with an upper bound on their growth factor. + * Values <= 1 disable native buckets (the default). A value of 1.1 is recommended. + */ + nativeHistogramBucketFactor?: number; + /** Absolute values <= this threshold go into the zero bucket. Default: 2^-128. */ + nativeHistogramZeroThreshold?: number; + /** + * Limit populated positive and negative buckets per label set by reducing + * resolution, down to schema -4. Default: 160. Zero disables the limit. + */ + nativeHistogramMaxBucketNumber?: number; +} + +export interface HistogramConfiguration + extends MetricConfiguration, NativeHistogramConfiguration { buckets?: number[]; collect?: CollectFunction>; } @@ -640,8 +697,11 @@ export class Histogram { */ startTimer( labels?: LabelValues, - exemplarLabels?: LabelValues, - ): (labels?: LabelValues, exemplarLabels?: LabelValues) => number; + exemplarLabels?: Record, + ): ( + labels?: LabelValues, + exemplarLabels?: Record, + ) => number; /** * Reset histogram values @@ -694,10 +754,13 @@ export namespace Histogram { * @returns Function to invoke when timer should be stopped. The value it * returns is the timed duration. */ - startTimer(): (labels?: LabelValues) => void; + startTimer(): ( + labels?: LabelValues, + exemplarLabels?: Record, + ) => number; } - interface Config { + interface Config extends NativeHistogramConfiguration { /** * Buckets used in the histogram */ diff --git a/index.js b/index.js index 52ca3631..96319260 100644 --- a/index.js +++ b/index.js @@ -33,6 +33,8 @@ Object.defineProperty(exports, 'contentType', { }); exports.prometheusContentType = exports.Registry.PROMETHEUS_CONTENT_TYPE; exports.openMetricsContentType = exports.Registry.OPENMETRICS_CONTENT_TYPE; +exports.prometheusProtobufContentType = + exports.Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE; exports.validateMetricName = require('./lib/validation').validateMetricName; exports.Counter = require('./lib/counter'); diff --git a/lib/cluster.js b/lib/cluster.js index b3b04004..0e4a3692 100644 --- a/lib/cluster.js +++ b/lib/cluster.js @@ -62,10 +62,11 @@ class AggregatorRegistry extends Registry { /** * Gets aggregated metrics for all workers. The optional callback and * returned Promise resolve with the same value; either may be used. - * @returns {Promise} Promise that resolves with the aggregated + * @returns {Promise} Promise that resolves with the aggregated * metrics. */ async clusterMetrics() { + const contentType = this.contentType; const requestId = requestCtr++; const orderedWorkers = [...workers.values()] .filter(worker => worker.isConnected()) @@ -91,7 +92,7 @@ class AggregatorRegistry extends Registry { const request = { responseHandlers, promise: waitFor( - this.#gather(requestId, metricSnapshot, responsePromises), + this.#gather(requestId, metricSnapshot, responsePromises, contentType), 5_000, ), }; @@ -128,15 +129,13 @@ class AggregatorRegistry extends Registry { * @param requestId {number} * @param {any[]} historical - Previously collected values * @param promises {Promise[]} - * @returns {Promise} + * @param contentType {string} + * @returns {Promise} */ - async #gather(requestId, historical, promises) { + async #gather(requestId, historical, promises, contentType) { const responses = await Promise.all(promises); const metrics = responses.flatMap(response => response.metrics); - return Registry.aggregate( - [historical, ...metrics], - this.contentType, - ).metrics(); + return Registry.aggregate([historical, ...metrics], contentType).metrics(); } get contentType() { diff --git a/lib/histogram.js b/lib/histogram.js index ee1f8ae5..2eee93f6 100644 --- a/lib/histogram.js +++ b/lib/histogram.js @@ -27,11 +27,16 @@ const { } = require('./util'); const { Metric } = require('./metric'); const Exemplar = require('./exemplar'); +const { + NativeHistogram, + resolveNativeHistogramConfig, +} = require('./nativeHistogram'); class Histogram extends Metric { constructor(config) { super(config, { buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], + nativeHistogramConfig: resolveNativeHistogramConfig(config), }); this.type = 'histogram'; @@ -68,10 +73,7 @@ class Histogram extends Metric { if (this.labelNames.length === 0) { this.store = new LabelMap(this.labelNames); - this.store.merge( - {}, - createBaseValues({}, this.bucketValues, this.bucketExemplars), - ); + this.store.merge({}, createBaseValues({}, this)); } } @@ -107,6 +109,7 @@ class Histogram extends Metric { exemplar.labelSet = exemplarLabels; exemplar.value = value; exemplar.timestamp = nowTimestamp(); + this.store.entry(labels).nativeHistogram?.addExemplar(exemplar); } async get() { @@ -125,13 +128,19 @@ class Histogram extends Metric { .map(extractBucketValuesForExport(this)) .reduce(addSumAndCountForExport(this), []); - return { + const data = { name: this.name, help: this.help, type: this.type, values, aggregator: this.aggregator, }; + if (this.nativeHistogramConfig) { + data.nativeHistograms = Array.from(this.store.values(), entry => + entry.nativeHistogram.snapshot(entry), + ); + } + return data; } reset() { @@ -145,10 +154,7 @@ class Histogram extends Metric { */ zero(labels) { this.store.validate(labels); - this.store.merge( - labels, - createBaseValues(labels, this.bucketValues, this.bucketExemplars), - ); + this.store.merge(labels, createBaseValues(labels, this)); } /** @@ -174,7 +180,7 @@ class Histogram extends Metric { this.store.validate(labels); return { observe: value => observe(this, labels, value), - startTimer: () => startTimer(this, labels), + startTimer: () => this.startTimer(labels), }; } @@ -248,11 +254,7 @@ function observe(histogram, labels, value) { if (entry === undefined) { entry = histogram.store.merge( labelValuePair.labels, - createBaseValues( - labelValuePair.labels, - histogram.bucketValues, - histogram.bucketExemplars, - ), + createBaseValues(labelValuePair.labels, histogram), ); } @@ -260,21 +262,27 @@ function observe(histogram, labels, value) { entry.sum += labelValuePair.value; entry.count += 1; + entry.nativeHistogram?.observe(labelValuePair.value); if (Object.hasOwn(entry.bucketValues, b)) { entry.bucketValues[b] += 1; } } -function createBaseValues(labels, bucketValues, bucketExemplars) { +function createBaseValues(labels, histogram) { const result = { labels, - bucketValues: { ...bucketValues }, + bucketValues: { ...histogram.bucketValues }, sum: 0, count: 0, }; - if (bucketExemplars) { - result.bucketExemplars = { ...bucketExemplars }; + if (histogram.bucketExemplars) { + result.bucketExemplars = { ...histogram.bucketExemplars }; + } + if (histogram.nativeHistogramConfig) { + result.nativeHistogram = new NativeHistogram( + histogram.nativeHistogramConfig, + ); } return result; } diff --git a/lib/metric.js b/lib/metric.js index 0c519de0..c3cc4f0c 100644 --- a/lib/metric.js +++ b/lib/metric.js @@ -76,7 +76,7 @@ class Metric { register.contentType === Registry.PROMETHEUS_CONTENT_TYPE ) { throw new TypeError( - 'Exemplars are supported only on OpenMetrics registries', + 'Exemplars are supported only on OpenMetrics or Prometheus protobuf registries', ); } register.registerMetric(this); diff --git a/lib/metricAggregators.js b/lib/metricAggregators.js index 6f897c16..9c3e7672 100644 --- a/lib/metricAggregators.js +++ b/lib/metricAggregators.js @@ -14,16 +14,29 @@ 'use strict'; -const { LabelGrouper } = require('./util'); +const { LabelGrouper, Grouper } = require('./util'); +const { + copyNativeHistogram, + sumNativeHistograms, + nativeHistogramKey, +} = require('./nativeHistogram'); /** * Returns a new function that applies the `aggregatorFn` to the values. * @param {Function} aggregatorFn function to apply to values. + * @param {Function} [nativeAggregatorFn] function to apply to native histograms. * @returns {Function} aggregator function */ -function AggregatorFactory(aggregatorFn) { +function AggregatorFactory(aggregatorFn, nativeAggregatorFn) { return metrics => { if (metrics.length === 0) return; + // Same-named metrics must agree on whether native histograms are enabled. + const hasNativeHistograms = metrics[0].nativeHistograms !== undefined; + if (hasNativeHistograms && !nativeAggregatorFn) { + throw new TypeError( + 'Native histograms support only sum, first, or omit aggregation', + ); + } const result = { help: metrics[0].help, name: metrics[0].name, @@ -38,10 +51,14 @@ function AggregatorFactory(aggregatorFn) { const name = value.metricName ?? ''; let group = byNames.get(name); if (group === undefined) { - group = new LabelGrouper(); + group = hasNativeHistograms ? new Grouper() : new LabelGrouper(); byNames.set(name, group); } - group.add(value); + if (hasNativeHistograms) { + group.add(nativeHistogramKey(value.labels), value); + } else { + group.add(value); + } }); }); // Apply aggregator function to gathered metrics. @@ -59,6 +76,18 @@ function AggregatorFactory(aggregatorFn) { result.values.push(valObj); }); }); + if (hasNativeHistograms) { + const byLabels = new Grouper(); + for (const metric of metrics) { + for (const histogram of metric.nativeHistograms) { + byLabels.add(nativeHistogramKey(histogram.labels), histogram); + } + } + result.nativeHistograms = Array.from( + byLabels.values(), + nativeAggregatorFn, + ); + } return result; }; } @@ -72,11 +101,17 @@ exports.aggregators = { /** * @returns The sum of values. */ - sum: AggregatorFactory(v => v.reduce((p, c) => p + c.value, 0)), + sum: AggregatorFactory( + v => v.reduce((p, c) => p + c.value, 0), + sumNativeHistograms, + ), /** * @returns The first value. */ - first: AggregatorFactory(v => v[0].value), + first: AggregatorFactory( + v => v[0].value, + v => copyNativeHistogram(v[0]), + ), /** * @returns {undefined} Undefined; omits the metric. */ diff --git a/lib/metrics.json b/lib/metrics.json new file mode 100644 index 00000000..2e3d7795 --- /dev/null +++ b/lib/metrics.json @@ -0,0 +1,344 @@ +{ + "nested": { + "io": { + "nested": { + "prometheus": { + "nested": { + "client": { + "options": { + "java_package": "io.prometheus.client", + "go_package": "github.com/prometheus/client_model/go;io_prometheus_client" + }, + "nested": { + "LabelPair": { + "edition": "proto2", + "fields": { + "name": { + "type": "string", + "id": 1 + }, + "value": { + "type": "string", + "id": 2 + } + } + }, + "MetricType": { + "edition": "proto2", + "values": { + "COUNTER": 0, + "GAUGE": 1, + "SUMMARY": 2, + "UNTYPED": 3, + "HISTOGRAM": 4, + "GAUGE_HISTOGRAM": 5 + } + }, + "Gauge": { + "edition": "proto2", + "fields": { + "value": { + "type": "double", + "id": 1 + } + } + }, + "Counter": { + "edition": "proto2", + "fields": { + "value": { + "type": "double", + "id": 1 + }, + "exemplar": { + "type": "Exemplar", + "id": 2 + }, + "createdTimestamp": { + "type": "google.protobuf.Timestamp", + "id": 3, + "protoName": "created_timestamp" + } + } + }, + "Quantile": { + "edition": "proto2", + "fields": { + "quantile": { + "type": "double", + "id": 1 + }, + "value": { + "type": "double", + "id": 2 + } + } + }, + "Summary": { + "edition": "proto2", + "fields": { + "sampleCount": { + "type": "uint64", + "id": 1, + "protoName": "sample_count" + }, + "sampleSum": { + "type": "double", + "id": 2, + "protoName": "sample_sum" + }, + "quantile": { + "rule": "repeated", + "type": "Quantile", + "id": 3 + }, + "createdTimestamp": { + "type": "google.protobuf.Timestamp", + "id": 4, + "protoName": "created_timestamp" + } + } + }, + "Untyped": { + "edition": "proto2", + "fields": { + "value": { + "type": "double", + "id": 1 + } + } + }, + "Histogram": { + "edition": "proto2", + "fields": { + "sampleCount": { + "type": "uint64", + "id": 1, + "protoName": "sample_count" + }, + "sampleCountFloat": { + "type": "double", + "id": 4, + "protoName": "sample_count_float" + }, + "sampleSum": { + "type": "double", + "id": 2, + "protoName": "sample_sum" + }, + "bucket": { + "rule": "repeated", + "type": "Bucket", + "id": 3 + }, + "createdTimestamp": { + "type": "google.protobuf.Timestamp", + "id": 15, + "protoName": "created_timestamp" + }, + "schema": { + "type": "sint32", + "id": 5 + }, + "zeroThreshold": { + "type": "double", + "id": 6, + "protoName": "zero_threshold" + }, + "zeroCount": { + "type": "uint64", + "id": 7, + "protoName": "zero_count" + }, + "zeroCountFloat": { + "type": "double", + "id": 8, + "protoName": "zero_count_float" + }, + "negativeSpan": { + "rule": "repeated", + "type": "BucketSpan", + "id": 9, + "protoName": "negative_span" + }, + "negativeDelta": { + "rule": "repeated", + "type": "sint64", + "id": 10, + "protoName": "negative_delta" + }, + "negativeCount": { + "rule": "repeated", + "type": "double", + "id": 11, + "protoName": "negative_count" + }, + "positiveSpan": { + "rule": "repeated", + "type": "BucketSpan", + "id": 12, + "protoName": "positive_span" + }, + "positiveDelta": { + "rule": "repeated", + "type": "sint64", + "id": 13, + "protoName": "positive_delta" + }, + "positiveCount": { + "rule": "repeated", + "type": "double", + "id": 14, + "protoName": "positive_count" + }, + "exemplars": { + "rule": "repeated", + "type": "Exemplar", + "id": 16 + } + } + }, + "Bucket": { + "edition": "proto2", + "fields": { + "cumulativeCount": { + "type": "uint64", + "id": 1, + "protoName": "cumulative_count" + }, + "cumulativeCountFloat": { + "type": "double", + "id": 4, + "protoName": "cumulative_count_float" + }, + "upperBound": { + "type": "double", + "id": 2, + "protoName": "upper_bound" + }, + "exemplar": { + "type": "Exemplar", + "id": 3 + } + } + }, + "BucketSpan": { + "edition": "proto2", + "fields": { + "offset": { + "type": "sint32", + "id": 1 + }, + "length": { + "type": "uint32", + "id": 2 + } + } + }, + "Exemplar": { + "edition": "proto2", + "fields": { + "label": { + "rule": "repeated", + "type": "LabelPair", + "id": 1 + }, + "value": { + "type": "double", + "id": 2 + }, + "timestamp": { + "type": "google.protobuf.Timestamp", + "id": 3 + } + } + }, + "Metric": { + "edition": "proto2", + "fields": { + "label": { + "rule": "repeated", + "type": "LabelPair", + "id": 1 + }, + "gauge": { + "type": "Gauge", + "id": 2 + }, + "counter": { + "type": "Counter", + "id": 3 + }, + "summary": { + "type": "Summary", + "id": 4 + }, + "untyped": { + "type": "Untyped", + "id": 5 + }, + "histogram": { + "type": "Histogram", + "id": 7 + }, + "timestampMs": { + "type": "int64", + "id": 6, + "protoName": "timestamp_ms" + } + } + }, + "MetricFamily": { + "edition": "proto2", + "fields": { + "name": { + "type": "string", + "id": 1 + }, + "help": { + "type": "string", + "id": 2 + }, + "type": { + "type": "MetricType", + "id": 3 + }, + "metric": { + "rule": "repeated", + "type": "Metric", + "id": 4 + }, + "unit": { + "type": "string", + "id": 5 + } + } + } + } + } + } + } + } + }, + "google": { + "nested": { + "protobuf": { + "nested": { + "Timestamp": { + "fields": { + "seconds": { + "type": "int64", + "id": 1 + }, + "nanos": { + "type": "int32", + "id": 2 + } + } + } + } + } + } + } + } +} diff --git a/lib/metrics.proto b/lib/metrics.proto new file mode 100644 index 00000000..2f2bff0c --- /dev/null +++ b/lib/metrics.proto @@ -0,0 +1,156 @@ +// Copyright 2013 Prometheus Team +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto2"; + +package io.prometheus.client; +option java_package = "io.prometheus.client"; +option go_package = "github.com/prometheus/client_model/go;io_prometheus_client"; + +import "google/protobuf/timestamp.proto"; + +message LabelPair { + optional string name = 1; + optional string value = 2; +} + +enum MetricType { + // COUNTER must use the Metric field "counter". + COUNTER = 0; + // GAUGE must use the Metric field "gauge". + GAUGE = 1; + // SUMMARY must use the Metric field "summary". + SUMMARY = 2; + // UNTYPED must use the Metric field "untyped". + UNTYPED = 3; + // HISTOGRAM must use the Metric field "histogram". + HISTOGRAM = 4; + // GAUGE_HISTOGRAM must use the Metric field "histogram". + GAUGE_HISTOGRAM = 5; +} + +message Gauge { + optional double value = 1; +} + +message Counter { + optional double value = 1; + optional Exemplar exemplar = 2; + + optional google.protobuf.Timestamp created_timestamp = 3; +} + +message Quantile { + optional double quantile = 1; + optional double value = 2; +} + +message Summary { + optional uint64 sample_count = 1; + optional double sample_sum = 2; + repeated Quantile quantile = 3; + + optional google.protobuf.Timestamp created_timestamp = 4; +} + +message Untyped { + optional double value = 1; +} + +message Histogram { + optional uint64 sample_count = 1; + optional double sample_count_float = 4; // Overrides sample_count if > 0. + optional double sample_sum = 2; + // Buckets for the conventional histogram. + repeated Bucket bucket = 3; // Ordered in increasing order of upper_bound, +Inf bucket is optional. + + optional google.protobuf.Timestamp created_timestamp = 15; + + // Everything below here is for native histograms (formerly known as sparse histograms). + + // schema defines the bucket schema. Currently, valid numbers are -4 <= n <= 8. + // They are all for base-2 bucket schemas, where 1 is a bucket boundary in each case, and + // then each power of two is divided into 2^n logarithmic buckets. + // Or in other words, each bucket boundary is the previous boundary times 2^(2^-n). + // In the future, more bucket schemas may be added using numbers < -4 or > 8. + optional sint32 schema = 5; + optional double zero_threshold = 6; // Breadth of the zero bucket. + optional uint64 zero_count = 7; // Count in zero bucket. + optional double zero_count_float = 8; // Overrides sb_zero_count if > 0. + + // Negative buckets for the native histogram. + repeated BucketSpan negative_span = 9; + // Use either "negative_delta" or "negative_count", the former for + // regular histograms with integer counts, the latter for float + // histograms. + repeated sint64 negative_delta = 10; // Count delta of each bucket compared to previous one (or to zero for 1st bucket). + repeated double negative_count = 11; // Absolute count of each bucket. + + // Positive buckets for the native histogram. + // Use a no-op span (offset 0, length 0) for a native histogram without any + // observations yet and with a zero_threshold of 0. Otherwise, it would be + // indistinguishable from a classic histogram. + repeated BucketSpan positive_span = 12; + // Use either "positive_delta" or "positive_count", the former for + // regular histograms with integer counts, the latter for float + // histograms. + repeated sint64 positive_delta = 13; // Count delta of each bucket compared to previous one (or to zero for 1st bucket). + repeated double positive_count = 14; // Absolute count of each bucket. + + // Only used for native histograms. These exemplars MUST have a timestamp. + repeated Exemplar exemplars = 16; +} + +// A Bucket of a conventional histogram, each of which is treated as +// an individual counter-like time series by Prometheus. +message Bucket { + optional uint64 cumulative_count = 1; // Cumulative in increasing order. + optional double cumulative_count_float = 4; // Overrides cumulative_count if > 0. + optional double upper_bound = 2; // Inclusive. + optional Exemplar exemplar = 3; +} + +// A BucketSpan defines a number of consecutive buckets in a native +// histogram with their offset. Logically, it would be more +// straightforward to include the bucket counts in the Span. However, +// the protobuf representation is more compact in the way the data is +// structured here (with all the buckets in a single array separate +// from the Spans). +message BucketSpan { + optional sint32 offset = 1; // Gap to previous span, or starting point for 1st span (which can be negative). + optional uint32 length = 2; // Length of consecutive buckets. +} + +message Exemplar { + repeated LabelPair label = 1; + optional double value = 2; + optional google.protobuf.Timestamp timestamp = 3; // OpenMetrics-style. +} + +message Metric { + repeated LabelPair label = 1; + optional Gauge gauge = 2; + optional Counter counter = 3; + optional Summary summary = 4; + optional Untyped untyped = 5; + optional Histogram histogram = 7; + optional int64 timestamp_ms = 6; +} + +message MetricFamily { + optional string name = 1; + optional string help = 2; + optional MetricType type = 3; + repeated Metric metric = 4; + optional string unit = 5; +} diff --git a/lib/nativeHistogram.js b/lib/nativeHistogram.js new file mode 100644 index 00000000..2208612d --- /dev/null +++ b/lib/nativeHistogram.js @@ -0,0 +1,360 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const bounds = require('./nativeHistogramBounds'); +const { isObject, nowTimestamp } = require('./util'); + +const MIN_SCHEMA = -4; +const MAX_SCHEMA = 8; +const MAX_EXEMPLARS = 10; +const float = new DataView(new ArrayBuffer(8)); + +function resolveNativeHistogramConfig(config) { + if (!isObject(config)) return undefined; + + const factor = + config.nativeHistogramBucketFactor === undefined + ? 0 + : config.nativeHistogramBucketFactor; + const zeroThreshold = + config.nativeHistogramZeroThreshold === undefined + ? 2 ** -128 + : config.nativeHistogramZeroThreshold; + const maxBucketNumber = + config.nativeHistogramMaxBucketNumber === undefined + ? 160 + : config.nativeHistogramMaxBucketNumber; + + if (!Number.isFinite(factor) || factor < 0) { + throw new TypeError( + 'nativeHistogramBucketFactor must be a finite, non-negative number', + ); + } + if (!Number.isFinite(zeroThreshold) || zeroThreshold < 0) { + throw new TypeError( + 'nativeHistogramZeroThreshold must be a finite, non-negative number', + ); + } + if (!Number.isSafeInteger(maxBucketNumber) || maxBucketNumber < 0) { + throw new TypeError( + 'nativeHistogramMaxBucketNumber must be a non-negative safe integer', + ); + } + if (factor <= 1) return undefined; + + if (!['sum', 'first', 'omit'].includes(config.aggregator ?? 'sum')) { + throw new TypeError( + 'Native histograms support only sum, first, or omit aggregation', + ); + } + + let schema = MIN_SCHEMA; + while (schema < MAX_SCHEMA && 2 ** (2 ** -schema) > factor) { + schema++; + } + return { schema, zeroThreshold, maxBucketNumber }; +} + +class NativeHistogram { + constructor(config) { + this.schema = config.schema; + this.zeroThreshold = config.zeroThreshold; + this.maxBucketNumber = config.maxBucketNumber; + this.zeroCount = 0; + this.positiveBuckets = new Map(); + this.negativeBuckets = new Map(); + this.createdTimestamp = nowTimestamp(); + this.exemplars = []; + } + + observe(value) { + const magnitude = Math.abs(value); + if (magnitude <= this.zeroThreshold) { + this.zeroCount++; + return; + } + + const buckets = value > 0 ? this.positiveBuckets : this.negativeBuckets; + addBucket(buckets, bucketIndex(magnitude, this.schema), 1); + + // Coarsening preserves every observation and does not reset the counter. + // The limit is best effort once the lowest supported schema is reached. + while ( + this.maxBucketNumber > 0 && + this.positiveBuckets.size + this.negativeBuckets.size > + this.maxBucketNumber && + this.schema > MIN_SCHEMA + ) { + this.positiveBuckets = coarsenBuckets(this.positiveBuckets, 1); + this.negativeBuckets = coarsenBuckets(this.negativeBuckets, 1); + this.schema--; + } + } + + addExemplar(exemplar) { + this.exemplars.push(copyExemplar(exemplar)); + if (this.exemplars.length > MAX_EXEMPLARS) this.exemplars.shift(); + } + + snapshot({ labels, count, sum }) { + const positive = encodeBuckets(this.positiveBuckets); + const negative = encodeBuckets(this.negativeBuckets); + + // With no zero threshold, zero count, or spans, the wire format would + // otherwise be indistinguishable from a classic histogram. + if ( + positive.spans.length === 0 && + negative.spans.length === 0 && + this.zeroThreshold === 0 && + this.zeroCount === 0 + ) { + positive.spans.push({ offset: 0, length: 0 }); + } + + return { + labels: { ...labels }, + count, + sum, + schema: this.schema, + zeroThreshold: this.zeroThreshold, + zeroCount: this.zeroCount, + positiveSpans: positive.spans, + positiveDeltas: positive.deltas, + negativeSpans: negative.spans, + negativeDeltas: negative.deltas, + createdTimestamp: this.createdTimestamp, + exemplars: this.exemplars.map(copyExemplar), + }; + } +} + +// Find the inclusive bucket boundary using the same significands as +// Prometheus. Math.ceil(Math.log2(value) * 2**schema) rounds incorrectly +// immediately above many boundaries, especially at large exponents. +function bucketIndex(value, schema) { + let adjustment = 0; + if (value < 2 ** -1022) { + value *= 2 ** 52; + adjustment = -52; + } + float.setFloat64(0, value); + const high = float.getUint32(0); + const exponent = (high >>> 20) - 1022 + adjustment; + float.setUint32(0, (high & 0xfffff) | 0x3fe00000); + const fraction = float.getFloat64(0); + + if (schema <= 0) { + return Math.ceil((exponent - (fraction === 0.5 ? 1 : 0)) / 2 ** -schema); + } + + const size = 2 ** schema; + const stride = 2 ** (MAX_SCHEMA - schema); + let low = 0; + let highIndex = size; + while (low < highIndex) { + const mid = (low + highIndex) >>> 1; + if (fraction <= bounds[mid * stride]) { + highIndex = mid; + } else { + low = mid + 1; + } + } + return low + (exponent - 1) * size; +} + +function upperBound(index, schema) { + if (schema < 0) + return Math.min(2 ** (index * 2 ** -schema), Number.MAX_VALUE); + + const size = 2 ** schema; + const exponent = Math.floor(index / size); + const fraction = + 2 * bounds[(index - exponent * size) * 2 ** (MAX_SCHEMA - schema)]; + if (exponent >= 1024) return Number.MAX_VALUE; + // Scale subnormals in two steps so the power itself doesn't underflow. + if (exponent < -1022) { + return fraction * 2 ** (exponent + 1074) * Number.MIN_VALUE; + } + return fraction * 2 ** exponent; +} + +function addBucket(buckets, index, count) { + buckets.set(index, (buckets.get(index) ?? 0) + count); +} + +function coarsenBuckets(buckets, schemaDifference) { + if (schemaDifference === 0) return buckets; + const result = new Map(); + const divisor = 2 ** schemaDifference; + for (const [index, count] of buckets) { + addBucket(result, Math.ceil(index / divisor), count); + } + return result; +} + +function encodeBuckets(buckets) { + const spans = []; + const deltas = []; + let nextIndex = 0; + let previousCount = 0; + + function append(count) { + spans[spans.length - 1].length++; + deltas.push(count - previousCount); + previousCount = count; + } + + for (const index of [...buckets.keys()].sort((a, b) => a - b)) { + const gap = index - nextIndex; + if (spans.length === 0 || gap > 2) { + spans.push({ offset: gap, length: 0 }); + } else { + for (let i = 0; i < gap; i++) append(0); + } + append(buckets.get(index)); + nextIndex = index + 1; + } + return { spans, deltas }; +} + +function decodeBuckets(spans, deltas) { + const result = new Map(); + let index = 0; + let count = 0; + let deltaIndex = 0; + for (const span of spans) { + index += span.offset; + for (let i = 0; i < span.length; i++) { + count += deltas[deltaIndex++]; + if (count !== 0) result.set(index, count); + index++; + } + } + return result; +} + +function copyExemplar(exemplar) { + return { + labelSet: { ...exemplar.labelSet }, + value: exemplar.value, + timestamp: exemplar.timestamp, + }; +} + +function nativeHistogramKey(labels) { + return JSON.stringify( + Object.keys(labels) + .sort() + .map(name => [name, String(labels[name])]), + ); +} + +function copyNativeHistogram(histogram) { + return { + ...histogram, + labels: { ...histogram.labels }, + positiveSpans: histogram.positiveSpans.map(span => { + return { ...span }; + }), + positiveDeltas: [...histogram.positiveDeltas], + negativeSpans: histogram.negativeSpans.map(span => { + return { ...span }; + }), + negativeDeltas: [...histogram.negativeDeltas], + exemplars: (histogram.exemplars ?? []).map(copyExemplar), + }; +} + +function sumNativeHistograms(histograms) { + const schema = Math.min(...histograms.map(histogram => histogram.schema)); + const merged = new NativeHistogram({ + schema, + zeroThreshold: Math.max( + ...histograms.map(histogram => histogram.zeroThreshold), + ), + maxBucketNumber: 0, + }); + const prepared = histograms.map(histogram => { + return { + histogram, + positive: coarsenBuckets( + decodeBuckets(histogram.positiveSpans, histogram.positiveDeltas), + histogram.schema - schema, + ), + negative: coarsenBuckets( + decodeBuckets(histogram.negativeSpans, histogram.negativeDeltas), + histogram.schema - schema, + ), + }; + }); + + // A larger zero threshold can cut through a populated bucket in another + // histogram. Include that entire bucket; its observations cannot be split. + let widened; + do { + widened = false; + for (const source of prepared) { + if (source.histogram.zeroThreshold === merged.zeroThreshold) continue; + const index = bucketIndex(merged.zeroThreshold, schema); + const bound = upperBound(index, schema); + if ( + bound > merged.zeroThreshold && + (source.positive.has(index) || source.negative.has(index)) + ) { + merged.zeroThreshold = bound; + widened = true; + } + } + } while (widened); + + let count = 0; + let sum = 0; + merged.createdTimestamp = Math.min( + ...histograms.map(histogram => histogram.createdTimestamp), + ); + for (const source of prepared) { + count += source.histogram.count; + sum += source.histogram.sum; + merged.zeroCount += source.histogram.zeroCount; + for (const [buckets, destination] of [ + [source.positive, merged.positiveBuckets], + [source.negative, merged.negativeBuckets], + ]) { + for (const [index, population] of buckets) { + if (upperBound(index, schema) <= merged.zeroThreshold) { + merged.zeroCount += population; + } else { + addBucket(destination, index, population); + } + } + } + } + merged.exemplars = histograms + .flatMap(histogram => histogram.exemplars ?? []) + .sort((a, b) => a.timestamp - b.timestamp) + .slice(-MAX_EXEMPLARS) + .map(copyExemplar); + + return merged.snapshot({ labels: histograms[0].labels, count, sum }); +} + +module.exports = { + NativeHistogram, + resolveNativeHistogramConfig, + copyNativeHistogram, + sumNativeHistograms, + nativeHistogramKey, +}; diff --git a/lib/nativeHistogramBounds.js b/lib/nativeHistogramBounds.js new file mode 100644 index 00000000..1b0ec086 --- /dev/null +++ b/lib/nativeHistogramBounds.js @@ -0,0 +1,100 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +// Schema 8 significand boundaries from prometheus/client_golang, +// prometheus/histogram.go (nativeHistogramBounds). Lower schemas use every +// 2^(8-schema)-th entry. Keep these values rather than recomputing powers: +// rounding differences at bucket boundaries would disagree with Prometheus. +module.exports = Object.freeze([ + 0.5, 0.5013556375251013, 0.5027149505564014, 0.5040779490592088, + 0.5054446430258502, 0.5068150424757447, 0.5081891574554764, 0.509566998038869, + 0.5109485743270583, 0.5123338964485679, 0.5137229745593818, + 0.5151158188430205, 0.5165124395106142, 0.5179128468009786, + 0.5193170509806894, 0.520725062344158, 0.5221368912137069, 0.5235525479396449, + 0.5249720429003435, 0.526395386502313, 0.5278225891802786, 0.5292536613972564, + 0.5306886136446309, 0.5321274564422321, 0.5335702003384117, + 0.5350168559101208, 0.5364674337629877, 0.5379219445313954, + 0.5393803988785598, 0.5408428074966075, 0.5423091811066545, + 0.5437795304588847, 0.5452538663326288, 0.5467321995364429, + 0.5482145409081883, 0.549700901315111, 0.5511912916539204, 0.5526857228508706, + 0.5541842058618393, 0.5556867516724088, 0.5571933712979462, + 0.5587040757836845, 0.5602188762048033, 0.5617377836665098, + 0.5632608093041209, 0.564787964283144, 0.5663192597993595, 0.5678547070789026, + 0.5693943173783458, 0.5709381019847808, 0.572486072215902, 0.5740382394200894, + 0.5755946149764913, 0.5771552102951081, 0.5787200368168754, + 0.5802891060137493, 0.5818624293887887, 0.5834400184762408, 0.585021884841625, + 0.5866080400818185, 0.5881984958251406, 0.5897932637314379, + 0.5913923554921704, 0.5929957828304968, 0.5946035575013605, + 0.5962156912915756, 0.5978321960199137, 0.5994530835371903, + 0.6010783657263515, 0.6027080545025619, 0.6043421618132907, + 0.6059806996384005, 0.6076236799902344, 0.6092711149137041, + 0.6109230164863786, 0.6125793968185725, 0.6142402680534349, + 0.6159056423670379, 0.6175755319684665, 0.6192499490999082, 0.620928906036742, + 0.622612415087629, 0.6243004885946023, 0.6259931389331581, 0.6276903785123455, + 0.6293922197748583, 0.6310986751971253, 0.6328097572894031, + 0.6345254785958666, 0.6362458516947014, 0.637970889198196, 0.6397006037528346, + 0.6414350080393891, 0.6431741147730128, 0.6449179367033329, + 0.6466664866145447, 0.6484197773255048, 0.6501778216898253, + 0.6519406325959679, 0.6537082229673385, 0.6554806057623822, + 0.6572577939746774, 0.659039800633032, 0.6608266388015788, 0.6626183215798706, + 0.6644148621029772, 0.6662162735415805, 0.6680225691020727, + 0.6698337620266515, 0.6716498655934177, 0.6734708931164728, + 0.6752968579460171, 0.6771277734684463, 0.6789636531064505, + 0.6808045103191123, 0.6826503586020058, 0.6845012114872953, + 0.6863570825438342, 0.688217985377265, 0.690083933630119, 0.6919549409819159, + 0.6938310211492645, 0.6957121878859629, 0.6975984549830999, + 0.6994898362691555, 0.7013863456101023, 0.7032879969095076, + 0.7051948041086352, 0.7071067811865475, 0.7090239421602076, + 0.7109463010845827, 0.7128738720527471, 0.7148066691959849, + 0.7167447066838943, 0.718687998724491, 0.7206365595643126, 0.7225904034885232, + 0.7245495448210174, 0.7265139979245261, 0.7284837772007218, + 0.7304588970903234, 0.7324393720732029, 0.7344252166684908, + 0.7364164454346837, 0.7384130729697496, 0.7404151139112358, + 0.7424225829363761, 0.7444354947621984, 0.7464538641456323, + 0.7484777058836176, 0.7505070348132126, 0.7525418658117031, + 0.7545822137967112, 0.7566280937263048, 0.7586795205991071, + 0.7607365094544071, 0.762799075372269, 0.7648672334736434, 0.7669409989204777, + 0.7690203869158282, 0.7711054127039704, 0.7731960915705107, + 0.7752924388424999, 0.7773944698885442, 0.7795022001189185, + 0.7816156449856788, 0.7837348199827764, 0.7858597406461707, + 0.7879904225539431, 0.7901268813264122, 0.7922691326262467, + 0.7944171921585818, 0.7965710756711334, 0.7987307989543135, + 0.8008963778413465, 0.8030678282083853, 0.805245165974627, 0.8074284071024302, + 0.8096175675974316, 0.8118126635086642, 0.8140137109286738, + 0.8162207259936375, 0.8184337248834821, 0.820652723822003, 0.8228777390769823, + 0.8251087869603088, 0.8273458838280969, 0.8295890460808079, + 0.8318382901633681, 0.8340936325652911, 0.8363550898207981, + 0.8386226785089391, 0.8408964152537144, 0.8431763167241966, + 0.8454623996346523, 0.8477546807446661, 0.8500531768592616, + 0.8523579048290255, 0.8546688815502312, 0.8569861239649629, + 0.8593096490612387, 0.8616394738731368, 0.8639756154809185, + 0.8663180910111553, 0.8686669176368529, 0.871022112577578, 0.8733836930995842, + 0.8757516765159389, 0.8781260801866495, 0.8805069215187917, + 0.8828942179666361, 0.8852879870317771, 0.8876882462632604, 0.890095013257712, + 0.8925083056594671, 0.8949281411607002, 0.8973545375015533, + 0.8997875124702672, 0.9022270839033115, 0.9046732696855155, + 0.9071260877501991, 0.909585556079304, 0.9120516927035263, 0.9145245157024483, + 0.9170040432046711, 0.9194902933879467, 0.9219832844793128, + 0.9244830347552253, 0.9269895625416926, 0.92950288621441, 0.9320230241988943, + 0.9345499949706191, 0.9370838170551498, 0.93962450902828, 0.9421720895161669, + 0.9447265771954693, 0.9472879907934827, 0.9498563490882775, + 0.9524316709088368, 0.9550139751351947, 0.9576032806985735, + 0.9601996065815236, 0.9628029718180622, 0.9654133954938133, + 0.9680308967461471, 0.9706554947643201, 0.9732872087896164, + 0.9759260581154889, 0.9785720620876999, 0.9812252401044634, + 0.9838856116165875, 0.9865531961276168, 0.9892280131939752, + 0.9919100824251095, 0.9945994234836328, 0.9972960560854698, +]); diff --git a/lib/protobuf.js b/lib/protobuf.js new file mode 100644 index 00000000..65eb4d3e --- /dev/null +++ b/lib/protobuf.js @@ -0,0 +1,203 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const { Buffer } = require('node:buffer'); +const protobuf = require('protobufjs/light'); +const { nativeHistogramKey } = require('./nativeHistogram'); + +// Loaded only when a registry is scraped as protobuf. The schema is vendored +// from prometheus/client_model/io/prometheus/client/metrics.proto. +const root = protobuf.Root.fromJSON(require('./metrics.json')); +const MetricFamily = root.lookupType('io.prometheus.client.MetricFamily'); +const metricTypes = root.lookupEnum('io.prometheus.client.MetricType').values; + +function encodeMetricFamily(metric, defaultLabels) { + const type = metricTypes[metric.type.toUpperCase()]; + if (type === undefined) { + throw new TypeError( + `Cannot encode metric type ${metric.type} as Prometheus protobuf`, + ); + } + + const groups = new Map(); + const defaultLabelEntries = Object.entries(defaultLabels); + function labelsWithDefaults(labels, sharedLabels) { + const merged = { ...defaultLabels, ...labels, ...sharedLabels }; + for (const [name, value] of defaultLabelEntries) { + merged[name] ??= value; + } + return merged; + } + + function groupFor(labels) { + const key = nativeHistogramKey(labels); + let group = groups.get(key); + if (!group) { + group = { label: labelPairs(labels), [metric.type]: {} }; + groups.set(key, group); + } + return group[metric.type]; + } + + for (const value of metric.values ?? []) { + const labels = labelsWithDefaults(value.labels, value.sharedLabels); + const metricName = value.metricName ?? metric.name; + let bound; + let quantile; + if (metric.type === 'histogram' && metricName === `${metric.name}_bucket`) { + bound = labels.le === '+Inf' ? Infinity : Number(labels.le); + delete labels.le; + } else if (metric.type === 'summary' && Object.hasOwn(labels, 'quantile')) { + quantile = Number(labels.quantile); + delete labels.quantile; + } + + const data = groupFor(labels); + switch (metric.type) { + case 'counter': + case 'gauge': + case 'untyped': + data.value = value.value; + if (metric.type === 'counter' && value.exemplar) { + data.exemplar = encodeExemplar(value.exemplar); + } + break; + case 'histogram': + case 'summary': + if (metricName === `${metric.name}_sum`) { + data.sampleSum = value.value; + } else if (metricName === `${metric.name}_count`) { + data.sampleCount = value.value; + } else if (metric.type === 'histogram' && bound !== undefined) { + data.bucket ??= []; + const bucket = { cumulativeCount: value.value, upperBound: bound }; + if (value.exemplar) bucket.exemplar = encodeExemplar(value.exemplar); + data.bucket.push(bucket); + } else if (metric.type === 'summary' && quantile !== undefined) { + data.quantile ??= []; + data.quantile.push({ quantile, value: value.value }); + } + break; + default: + throw new TypeError( + `Cannot encode metric type ${metric.type} as Prometheus protobuf`, + ); + } + } + + for (const histogram of metric.nativeHistograms ?? []) { + const data = groupFor(labelsWithDefaults(histogram.labels)); + // Classic and native buckets in the same message share sum and count. + // Reject partial aggregation instead of publishing inconsistent data. + if ( + (data.sampleCount !== undefined && + data.sampleCount !== histogram.count) || + (data.sampleSum !== undefined && + !Object.is(data.sampleSum, histogram.sum)) + ) { + throw new Error( + `Classic and native histogram counts or sums differ for ${metric.name}`, + ); + } + Object.assign(data, { + sampleCount: histogram.count, + sampleSum: histogram.sum, + schema: histogram.schema, + zeroThreshold: histogram.zeroThreshold, + zeroCount: histogram.zeroCount, + positiveSpan: histogram.positiveSpans, + positiveDelta: histogram.positiveDeltas, + negativeSpan: histogram.negativeSpans, + negativeDelta: histogram.negativeDeltas, + createdTimestamp: timestamp(histogram.createdTimestamp), + exemplars: (histogram.exemplars ?? []).map(encodeExemplar), + }); + + // The text representation always includes an implicit +Inf bucket. + // With buckets: [], omit it from protobuf to expose a native-only series. + if (data.bucket?.length === 1 && data.bucket[0].upperBound === Infinity) { + delete data.bucket; + } + } + + for (const group of groups.values()) { + if (group.histogram) { + encodeHistogramCount(group.histogram, 'sampleCount'); + for (const bucket of group.histogram.bucket ?? []) { + encodeHistogramCount(bucket, 'cumulativeCount'); + } + group.histogram.bucket?.sort((a, b) => a.upperBound - b.upperBound); + } + if ( + group.summary?.sampleCount !== undefined && + !Number.isInteger(group.summary.sampleCount) + ) { + throw new TypeError( + `Prometheus protobuf requires an integer sample count for summary ${metric.name}`, + ); + } + group.summary?.quantile?.sort((a, b) => a.quantile - b.quantile); + } + + return Buffer.from( + MetricFamily.encodeDelimited({ + name: metric.name, + help: metric.help, + type, + metric: [...groups.values()], + }).finish(), + ); +} + +// Classic histogram aggregation can produce fractional counts. Protobuf has +// dedicated double fields for them; uint64 encoding would truncate them. +function encodeHistogramCount(message, field) { + const count = message[field]; + if (count !== undefined && !Number.isInteger(count)) { + message[`${field}Float`] = count; + delete message[field]; + } +} + +function labelPairs(labels) { + return Object.keys(labels) + .sort() + .map(name => { + return { + name, + value: String(labels[name]), + }; + }); +} + +function timestamp(seconds) { + const milliseconds = Math.round(seconds * 1000); + const wholeSeconds = Math.floor(milliseconds / 1000); + return { + seconds: wholeSeconds, + nanos: (milliseconds - wholeSeconds * 1000) * 1e6, + }; +} + +function encodeExemplar(exemplar) { + return { + label: labelPairs(exemplar.labelSet), + value: exemplar.value, + timestamp: timestamp(exemplar.timestamp), + }; +} + +module.exports = { encodeMetricFamily }; diff --git a/lib/pushgateway.js b/lib/pushgateway.js index f43487b4..907f981b 100644 --- a/lib/pushgateway.js +++ b/lib/pushgateway.js @@ -75,6 +75,15 @@ async function useGateway(method, job, groupings) { const httpModule = requestParams.protocol === 'https:' ? https : http; const options = { ...this.requestOptions, method }; + if ( + method !== 'DELETE' && + this.registry.contentType === Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE + ) { + options.headers = { + 'Content-Type': this.registry.contentType, + ...options.headers, + }; + } return new Promise((resolve, reject) => { if (method === 'DELETE' && options.headers) { diff --git a/lib/registry.js b/lib/registry.js index 38f3903b..15fd5fdd 100644 --- a/lib/registry.js +++ b/lib/registry.js @@ -26,14 +26,15 @@ class Registry { return 'application/openmetrics-text; version=1.0.0; charset=utf-8'; } + static get PROMETHEUS_PROTOBUF_CONTENT_TYPE() { + return 'application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; encoding=delimited'; + } + constructor(regContentType = Registry.PROMETHEUS_CONTENT_TYPE) { this._metrics = new Map(); this._collectors = []; this._defaultLabels = {}; - if ( - regContentType !== Registry.PROMETHEUS_CONTENT_TYPE && - regContentType !== Registry.OPENMETRICS_CONTENT_TYPE - ) { + if (!isSupportedContentType(regContentType)) { throw new TypeError(`Content type ${regContentType} is unsupported`); } this._contentType = regContentType; @@ -60,12 +61,18 @@ class Registry { } async getMetricsAsString(metrics) { + const isOpenMetrics = + this.contentType === Registry.OPENMETRICS_CONTENT_TYPE; const metric = typeof metrics.getForPromString === 'function' ? await metrics.getForPromString() : await metrics.get(); - const name = escapeString(metric.name); + const name = escapeString( + isOpenMetrics && metric.type === 'counter' + ? standardizeCounterName(metric.name) + : metric.name, + ); const help = `# HELP ${name} ${escapeString(metric.help)}`; const type = `# TYPE ${name} ${metric.type}`; const values = [help, type]; @@ -75,11 +82,9 @@ class Registry { defaultLabelNames = undefined; } - const isOpenMetrics = - this.contentType === Registry.OPENMETRICS_CONTENT_TYPE; - for (const val of metric.values ?? []) { - const { metricName = name, labels = {}, sharedLabels } = val; + const { metricName = name, labels = {} } = val; + let { sharedLabels } = val; const seriesName = isOpenMetrics && metric.type === 'counter' ? `${metricName}_total` @@ -91,6 +96,17 @@ class Registry { if (defaultLabelNames !== undefined) { for (const labelName of defaultLabelNames) { seriesLabels[labelName] ??= this._defaultLabels[labelName]; + if ( + sharedLabels !== undefined && + Object.hasOwn(sharedLabels, labelName) && + (sharedLabels[labelName] === null || + sharedLabels[labelName] === undefined) + ) { + sharedLabels = { + ...sharedLabels, + [labelName]: this._defaultLabels[labelName], + }; + } } } @@ -124,21 +140,30 @@ class Registry { } async metrics() { - const isOpenMetrics = - this.contentType === Registry.OPENMETRICS_CONTENT_TYPE; - - const promises = this.getMetricsAsArray().map(metric => { - if (isOpenMetrics && metric.type === 'counter') { - metric.name = standardizeCounterName(metric.name); - } - return this.getMetricsAsString(metric); - }); - - const resolves = await Promise.all(promises); - - return isOpenMetrics - ? `${resolves.join('\n')}\n# EOF\n` - : `${resolves.join('\n\n')}\n`; + const contentType = this.contentType; + let output; + if (contentType === Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE) { + const { encodeMetricFamily } = require('./protobuf'); + const buffers = await Promise.all( + this.getMetricsAsArray().map(async metric => { + const data = + typeof metric.getForPromString === 'function' + ? await metric.getForPromString() + : await metric.get(); + return encodeMetricFamily(data, this._defaultLabels); + }), + ); + output = Buffer.concat(buffers); + } else { + const strings = await Promise.all( + this.getMetricsAsArray().map(metric => this.getMetricsAsString(metric)), + ); + output = + contentType === Registry.OPENMETRICS_CONTENT_TYPE + ? `${strings.join('\n')}\n# EOF\n` + : `${strings.join('\n\n')}\n`; + } + return output; } registerMetric(metric) { @@ -177,13 +202,15 @@ class Registry { const resolves = await Promise.all(promises); for (const item of resolves) { - if (defaultLabelNames !== undefined && item.values !== undefined) { - for (const val of item.values) { - // Make a copy before mutating - val.labels = { ...val.labels }; - - for (const labelName of defaultLabelNames) { - val.labels[labelName] ??= this._defaultLabels[labelName]; + if (defaultLabelNames !== undefined) { + for (const values of [item.values, item.nativeHistograms]) { + for (const val of values ?? []) { + // Make a copy before mutating + val.labels = { ...val.labels }; + + for (const labelName of defaultLabelNames) { + val.labels[labelName] ??= this._defaultLabels[labelName]; + } } } } @@ -221,11 +248,9 @@ class Registry { } setContentType(metricsContentType) { - if ( - metricsContentType === Registry.OPENMETRICS_CONTENT_TYPE || - metricsContentType === Registry.PROMETHEUS_CONTENT_TYPE - ) { + if (isSupportedContentType(metricsContentType)) { this._contentType = metricsContentType; + return this; } else { throw new Error(`Content type ${metricsContentType} is unsupported`); } @@ -303,6 +328,14 @@ class Registry { } } +function isSupportedContentType(contentType) { + return [ + Registry.PROMETHEUS_CONTENT_TYPE, + Registry.OPENMETRICS_CONTENT_TYPE, + Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE, + ].includes(contentType); +} + function formatLabels(labels, exclude) { const formatted = []; for (const [name, value] of Object.entries(labels)) { diff --git a/lib/worker.js b/lib/worker.js index aa5c3c64..e97b94ed 100644 --- a/lib/worker.js +++ b/lib/worker.js @@ -68,10 +68,11 @@ class WorkerRegistry extends Registry { /** * Gets aggregated metrics for all workers. The optional callback and * returned Promise resolve with the same value; either may be used. - * @returns {Promise} Promise that resolves with the aggregated + * @returns {Promise} Promise that resolves with the aggregated * metrics. */ async workerMetrics() { + const contentType = this.contentType; const requestId = requestCtr++; const orderedWorkers = [...workers.values()].sort( (left, right) => left.threadId - right.threadId, @@ -80,7 +81,9 @@ class WorkerRegistry extends Registry { if (orderedWorkers.length === 0) { if (historicMetrics.length === 0) { debug('No data found for requestId', requestId); - return ''; + return contentType === Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE + ? Buffer.alloc(0) + : ''; } else { debug('No workers found for requestId', requestId); } @@ -101,7 +104,7 @@ class WorkerRegistry extends Registry { const request = { responseHandlers, promise: waitFor( - this.#gather(requestId, metricSnapshot, responsePromises), + this.#gather(requestId, metricSnapshot, responsePromises, contentType), 5_000, ), }; @@ -134,17 +137,15 @@ class WorkerRegistry extends Registry { * @param requestId {number} * @param {any[]} historical - Previously collected values * @param promises {Promise[]} - * @returns {Promise} + * @param contentType {string} + * @returns {Promise} */ - async #gather(requestId, historical, promises) { + async #gather(requestId, historical, promises, contentType) { debug('Gathering data...', requestId); const responses = await Promise.all(promises); debug('Aggregating data...', requestId); const metrics = responses.flatMap(response => response.metrics); - return Registry.aggregate( - [historical, ...metrics], - this.contentType, - ).metrics(); + return Registry.aggregate([historical, ...metrics], contentType).metrics(); } get contentType() { diff --git a/package.json b/package.json index a669dcca..104bca9c 100644 --- a/package.json +++ b/package.json @@ -19,6 +19,7 @@ "run-prettier": "prettier .", "check-prettier": "npm run run-prettier -- --check", "compile-typescript": "tsc --project .", + "generate-protobuf": "node scripts/generate-protobuf.js", "prepare": "husky" }, "repository": { @@ -53,6 +54,7 @@ }, "dependencies": { "@opentelemetry/api": "^1.4.0", + "protobufjs": "^8.8.0", "tdigest": "^0.1.1" }, "types": "./index.d.ts", diff --git a/scripts/generate-protobuf.js b/scripts/generate-protobuf.js new file mode 100644 index 00000000..fb63c234 --- /dev/null +++ b/scripts/generate-protobuf.js @@ -0,0 +1,28 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const fs = require('node:fs'); +const path = require('node:path'); +const protobuf = require('protobufjs'); + +// Keep the runtime descriptor in sync with the vendored upstream schema. +// Encoding uses protobufjs/light and JSON so bundlers can include it without +// runtime filesystem access or parsing .proto files. +const root = protobuf.loadSync(path.join(__dirname, '../lib/metrics.proto')); +fs.writeFileSync( + path.join(__dirname, '../lib/metrics.json'), + `${JSON.stringify(root.toJSON(), null, '\t')}\n`, +); diff --git a/test/clusterTest.js b/test/clusterTest.js index c475266c..ebf5aa77 100644 --- a/test/clusterTest.js +++ b/test/clusterTest.js @@ -116,6 +116,39 @@ describe.each([ } }); + it('keeps pending scrapes in their original content types', async () => { + const { Registry: LocalRegistry, Histogram } = require('../index'); + const { decodeMetricFamilies } = require('./helpers/nativeHistogram'); + const local = new LocalRegistry(); + let finishCollecting; + const collected = new Promise(resolve => { + finishCollecting = resolve; + }); + new Histogram({ + name: 'pending_native', + help: 'Pending native histogram', + registers: [local], + nativeHistogramBucketFactor: 1.1, + async collect() { + await collected; + }, + }).observe(1); + AggregatorRegistry.setRegistries(local); + try { + const registry = new AggregatorRegistry(regType); + const text = registry.clusterMetrics(); + registry.setContentType(Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE); + const binary = registry.clusterMetrics(); + finishCollecting(); + const [textBody, binaryBody] = await Promise.all([text, binary]); + expect(textBody).toContain('pending_native_count 1'); + const [family] = decodeMetricFamilies(binaryBody); + expect(family.metric[0].histogram.sampleCount).toBe(1); + } finally { + AggregatorRegistry.setRegistries(LocalRegistry.globalRegistry); + } + }); + it("listeners don't accumulate", () => { for (let i = 0; i < 30; i++) { jest.resetModules(); diff --git a/test/exemplarsTest.js b/test/exemplarsTest.js index a349de39..cd900b71 100644 --- a/test/exemplarsTest.js +++ b/test/exemplarsTest.js @@ -31,7 +31,9 @@ describe('Exemplars', () => { labelNames: ['method', 'code'], enableExemplars: true, }); - }).toThrow('Exemplars are supported only on OpenMetrics registries'); + }).toThrow( + 'Exemplars are supported only on OpenMetrics or Prometheus protobuf registries', + ); }); describe.each([['OpenMetrics', Registry.OPENMETRICS_CONTENT_TYPE]])( 'with %s registry', diff --git a/test/helpers/nativeHistogram.js b/test/helpers/nativeHistogram.js new file mode 100644 index 00000000..8a389e96 --- /dev/null +++ b/test/helpers/nativeHistogram.js @@ -0,0 +1,61 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const path = require('node:path'); +const protobuf = require('protobufjs'); + +// Decode against the upstream .proto, independently of the runtime JSON +// descriptor and the client's conversion of metric snapshots to protobuf. +const root = protobuf.loadSync(path.join(__dirname, '../../lib/metrics.proto')); +const MetricFamily = root.lookupType('io.prometheus.client.MetricFamily'); + +function decodeMetricFamilies(buffer) { + const reader = protobuf.Reader.create(buffer); + const families = []; + while (reader.pos < reader.len) { + families.push( + MetricFamily.toObject(MetricFamily.decodeDelimited(reader), { + longs: Number, + enums: String, + }), + ); + } + return families; +} + +function bucketCounts(histogram, side) { + const counts = new Map(); + let index = 0; + let count = 0; + let position = 0; + for (const span of histogram[`${side}Spans`]) { + index += span.offset; + for (let i = 0; i < span.length; i++, index++) { + count += histogram[`${side}Deltas`][position++]; + if (count !== 0) counts.set(index, count); + } + } + return counts; +} + +function nextUp(value) { + const data = new DataView(new ArrayBuffer(8)); + data.setFloat64(0, value); + data.setBigUint64(0, data.getBigUint64(0) + 1n); + return data.getFloat64(0); +} + +module.exports = { decodeMetricFamilies, bucketCounts, nextUp }; diff --git a/test/nativeHistogramAggregationTest.js b/test/nativeHistogramAggregationTest.js new file mode 100644 index 00000000..5eb2007f --- /dev/null +++ b/test/nativeHistogramAggregationTest.js @@ -0,0 +1,207 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const { + Registry, + Histogram, + ClusterRegistry, + WorkerRegistry, + aggregators, +} = require('../index'); +const { + bucketCounts, + decodeMetricFamilies, +} = require('./helpers/nativeHistogram'); + +const contentType = Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE; + +function create(config = {}) { + return new Histogram({ + name: 'worker_duration_seconds', + help: 'Worker duration', + registers: [], + buckets: [0.5, 1, 2, 8], + nativeHistogramBucketFactor: 1.1, + nativeHistogramZeroThreshold: 0, + ...config, + }); +} + +function comparable(data) { + return { + count: data.count, + sum: data.sum, + schema: data.schema, + zeroCount: data.zeroCount, + zeroThreshold: data.zeroThreshold, + positive: bucketCounts(data, 'positive'), + negative: bucketCounts(data, 'negative'), + }; +} + +describe('native histogram aggregation', () => { + test.each([Registry, ClusterRegistry, WorkerRegistry])( + '%p aggregates JSON snapshots at the lowest common resolution', + async RegistryClass => { + const first = create(); + const second = create({ nativeHistogramBucketFactor: 2 }); + const firstValues = [0, 0.125, 0.5, 1, 1.5, 2, -0.25, -2]; + const secondValues = [0, 0.25, 0.5, 1, 4, -1, -4]; + firstValues.forEach(value => first.observe(value)); + secondValues.forEach(value => second.observe(value)); + const expected = create({ nativeHistogramBucketFactor: 2 }); + [...firstValues, ...secondValues].forEach(value => + expected.observe(value), + ); + + // Cluster IPC serializes JSON, while worker threads use structured clone. + // Neither Maps nor private collector state should be required here. + const snapshots = JSON.parse( + JSON.stringify([[await first.get()], [await second.get()]]), + ); + const before = JSON.stringify(snapshots); + const aggregated = RegistryClass.aggregate(snapshots, contentType); + const [data] = await aggregated.getMetricsAsJSON(); + expect(comparable(data.nativeHistograms[0])).toEqual( + comparable((await expected.get()).nativeHistograms[0]), + ); + expect(JSON.stringify(snapshots)).toBe(before); + expect( + decodeMetricFamilies(await aggregated.metrics())[0].metric[0].histogram + .sampleCount, + ).toBe(firstValues.length + secondValues.length); + }, + ); + + test.each([ + [0.5, 0.75, [0.6, -0.6], [0.8, -0.8], 1], + [0.5, 0.75, [2, -2], [0.8, -0.8], 0.75], + [0.75, 0.75, [0.8, -0.8], [0.9, -0.9], 0.75], + [0, 0.5, [0.25, -0.25, 1], [0.5, -0.5, 2], 0.5], + ])( + 'reconciles zero thresholds %s and %s without splitting populated buckets', + async (thresholdA, thresholdB, valuesA, valuesB, expectedThreshold) => { + const first = create({ nativeHistogramZeroThreshold: thresholdA }); + const second = create({ + nativeHistogramBucketFactor: 2, + nativeHistogramZeroThreshold: thresholdB, + }); + valuesA.forEach(value => first.observe(value)); + valuesB.forEach(value => second.observe(value)); + const expected = create({ + nativeHistogramBucketFactor: 2, + nativeHistogramZeroThreshold: expectedThreshold, + }); + [...valuesA, ...valuesB].forEach(value => expected.observe(value)); + const aggregated = Registry.aggregate( + [[await first.get()], [await second.get()]], + contentType, + ); + const [data] = await aggregated.getMetricsAsJSON(); + expect(comparable(data.nativeHistograms[0])).toEqual( + comparable((await expected.get()).nativeHistograms[0]), + ); + await expect(aggregated.metrics()).resolves.toBeInstanceOf(Buffer); + }, + ); + + test('handles a zero threshold at the largest finite double', async () => { + const first = create(); + const second = create({ nativeHistogramZeroThreshold: Number.MAX_VALUE }); + first.observe(Number.MAX_VALUE); + second.observe(-Number.MAX_VALUE); + const result = aggregators.sum([await first.get(), await second.get()]); + expect(result.nativeHistograms[0]).toMatchObject({ + count: 2, + sum: 0, + zeroCount: 2, + zeroThreshold: Number.MAX_VALUE, + }); + expect(bucketCounts(result.nativeHistograms[0], 'positive').size).toBe(0); + }); + + test('keeps the earliest creation time and a bounded set of exemplars', async () => { + jest.useFakeTimers(); + try { + jest.setSystemTime(1000); + const first = create({ enableExemplars: true }); + first.observe({ value: 1, exemplarLabels: { trace_id: 'first' } }); + jest.setSystemTime(2000); + const second = create({ enableExemplars: true }); + for (let i = 0; i < 12; i++) { + second.observe({ value: i, exemplarLabels: { trace_id: 'second' } }); + } + const result = aggregators.sum([await first.get(), await second.get()]); + expect(result.nativeHistograms[0].createdTimestamp).toBe(1); + expect(result.nativeHistograms[0].exemplars).toHaveLength(10); + expect(result.nativeHistograms[0].exemplars[0].labelSet).toEqual({ + trace_id: 'second', + }); + } finally { + jest.useRealTimers(); + } + }); + + test('supports first and omit aggregation without mutating source snapshots', async () => { + const first = create({ aggregator: 'first' }); + const second = create({ aggregator: 'first' }); + first.observe(1); + second.observe(2); + const snapshots = [await first.get(), await second.get()]; + const result = aggregators.first(snapshots); + expect(result.nativeHistograms).toEqual(snapshots[0].nativeHistograms); + result.nativeHistograms[0].positiveDeltas[0] = 100; + expect(snapshots[0].nativeHistograms[0].positiveDeltas).toEqual([1]); + expect(aggregators.omit(snapshots)).toBeUndefined(); + }); + + test.each(['average', 'min', 'max'])( + 'rejects %s aggregation of native snapshots', + async method => { + const data = await create().get(); + expect(() => aggregators[method]([data])).toThrow('sum, first, or omit'); + }, + ); + + test('preserves native metadata through repeated aggregation of historic worker snapshots', async () => { + const first = create(); + const second = create({ nativeHistogramBucketFactor: 2 }); + first.observe(1); + second.observe(2); + const historical = Registry.aggregate([ + [await first.get()], + ]).getMetricsAsArray(); + const combined = Registry.aggregate( + [historical, [await second.get()]], + contentType, + ); + const [family] = decodeMetricFamilies(await combined.metrics()); + expect(family.metric[0].histogram).toMatchObject({ + sampleCount: 2, + sampleSum: 3, + schema: 0, + positiveSpan: [{ offset: 0, length: 2 }], + positiveDelta: [1, 0], + }); + }); + + test('returns a Buffer when a protobuf worker registry has no workers', async () => { + const register = new WorkerRegistry(contentType); + const result = await register.workerMetrics(); + expect(Buffer.isBuffer(result)).toBe(true); + expect(result.length).toBe(0); + }); +}); diff --git a/test/nativeHistogramTest.js b/test/nativeHistogramTest.js new file mode 100644 index 00000000..60d7a362 --- /dev/null +++ b/test/nativeHistogramTest.js @@ -0,0 +1,439 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const { Histogram, Registry } = require('../index'); +const { bucketCounts, nextUp } = require('./helpers/nativeHistogram'); + +function create(config = {}) { + return new Histogram({ + name: 'native_histogram', + help: 'Native histogram test', + registers: [], + nativeHistogramBucketFactor: 2, + ...config, + }); +} + +async function snapshot(histogram) { + return (await histogram.get()).nativeHistograms[0]; +} + +afterEach(() => { + jest.useRealTimers(); +}); + +describe('native histogram configuration', () => { + test.each([ + [65536, -4], + [256, -3], + [16, -2], + [4, -1], + [2, 0], + [1.5, 1], + [1.2, 2], + [1.1, 3], + [1.05, 4], + [1.03, 5], + [1.02, 6], + [1.01, 7], + [1.005, 8], + [Number.MAX_VALUE, -4], + [1 + Number.EPSILON, 8], + ])('factor %s selects schema %s', async (factor, schema) => { + expect( + await snapshot(create({ nativeHistogramBucketFactor: factor })), + ).toMatchObject({ schema, count: 0, sum: 0, zeroThreshold: 2 ** -128 }); + }); + + test.each([undefined, 0, 0.5, 1])( + 'factor %s leaves native buckets disabled', + async factor => { + expect( + (await create({ nativeHistogramBucketFactor: factor }).get()) + .nativeHistograms, + ).toBeUndefined(); + }, + ); + + test.each([ + ...[-1, NaN, Infinity, '1.1', null, true].map(value => [ + 'nativeHistogramBucketFactor', + value, + ]), + ...[-1, NaN, Infinity, '0', null, false].map(value => [ + 'nativeHistogramZeroThreshold', + value, + ]), + ...[-1, 1.5, Infinity, NaN, '160', null, Number.MAX_SAFE_INTEGER + 1].map( + value => ['nativeHistogramMaxBucketNumber', value], + ), + ])('rejects invalid %s: %s before registration', (option, value) => { + const register = new Registry(); + expect(() => create({ registers: [register], [option]: value })).toThrow( + option, + ); + expect(register.getMetricsAsArray()).toEqual([]); + }); + + test.each(['min', 'max', 'average'])('rejects %s aggregation', aggregator => { + expect(() => create({ aggregator })).toThrow('sum, first, or omit'); + }); +}); + +describe('native observations', () => { + test('exports signed bucket spans, deltas, sum, and zero count', async () => { + const histogram = create(); + [0, 0.5, 1, 1, 2, 8, -0.5, -2, -2].forEach(value => + histogram.observe(value), + ); + const data = await snapshot(histogram); + expect(data).toMatchObject({ + count: 9, + sum: 8, + zeroCount: 1, + positiveSpans: [{ offset: -1, length: 5 }], + positiveDeltas: [1, 1, -1, -1, 1], + negativeSpans: [{ offset: -1, length: 3 }], + negativeDeltas: [1, -1, 2], + }); + }); + + test('keeps delta counts across gaps between spans', async () => { + const histogram = create(); + [1, 1, 1, 1024, 1024, 2 ** 20].forEach(value => histogram.observe(value)); + expect(await snapshot(histogram)).toMatchObject({ + positiveSpans: [ + { offset: 0, length: 1 }, + { offset: 9, length: 1 }, + { offset: 9, length: 1 }, + ], + positiveDeltas: [3, -1, -1], + }); + }); + + test.each(Array.from({ length: 13 }, (_, i) => i - 4))( + 'uses inclusive boundaries at schema %s', + async schema => { + const histogram = create({ + nativeHistogramBucketFactor: 2 ** (2 ** -schema), + nativeHistogramZeroThreshold: 0, + }); + const boundary = schema < 0 ? 2 ** (2 ** -schema) : 1; + const index = schema < 0 ? 1 : 0; + histogram.observe(boundary); + histogram.observe(nextUp(boundary)); + histogram.observe(-boundary); + histogram.observe(-nextUp(boundary)); + const data = await snapshot(histogram); + for (const side of ['positive', 'negative']) { + expect(bucketCounts(data, side)).toEqual( + new Map([ + [index, 1], + [index + 1, 1], + ]), + ); + } + }, + ); + + test.each([-1074, -1022, -256, -1, 0, 1, 128, 1023])( + 'does not round observations above 2^%s into the preceding bucket', + async exponent => { + const histogram = create({ nativeHistogramZeroThreshold: 0 }); + const boundary = 2 ** exponent; + histogram.observe(boundary); + histogram.observe(nextUp(boundary)); + expect(bucketCounts(await snapshot(histogram), 'positive')).toEqual( + new Map([ + [exponent, 1], + [exponent + 1, 1], + ]), + ); + }, + ); + + test('matches the Prometheus fractional boundary and its adjacent float', async () => { + const histogram = create({ nativeHistogramBucketFactor: 1.5 }); + // The canonical Prometheus boundary differs by one ULP from Math.SQRT2. + const boundary = 2 * 0.7071067811865475; + histogram.observe(boundary); + histogram.observe(nextUp(boundary)); + expect(bucketCounts(await snapshot(histogram), 'positive')).toEqual( + new Map([ + [1, 1], + [2, 1], + ]), + ); + }); + + test('covers the full finite double range on both sides of zero', async () => { + const histogram = create({ + nativeHistogramBucketFactor: 1.1, + nativeHistogramZeroThreshold: 0, + }); + [ + Number.MIN_VALUE, + -Number.MIN_VALUE, + Number.MAX_VALUE, + -Number.MAX_VALUE, + ].forEach(value => histogram.observe(value)); + const data = await snapshot(histogram); + expect(data).toMatchObject({ count: 4, sum: 0, zeroCount: 0 }); + for (const side of ['positive', 'negative']) { + expect(bucketCounts(data, side)).toEqual( + new Map([ + [-8592, 1], + [8192, 1], + ]), + ); + } + }); + + test('includes both endpoints of the zero bucket', async () => { + const histogram = create({ nativeHistogramZeroThreshold: 1 }); + [0, -0, 1, -1, nextUp(1), -nextUp(1)].forEach(value => + histogram.observe(value), + ); + const data = await snapshot(histogram); + expect(data.zeroCount).toBe(4); + expect(bucketCounts(data, 'positive')).toEqual(new Map([[1, 1]])); + expect(bucketCounts(data, 'negative')).toEqual(new Map([[1, 1]])); + }); + + test('defaults to a zero threshold of 2^-128 and allows exactly zero', async () => { + const regular = create(); + const zeroOnly = create({ nativeHistogramZeroThreshold: 0 }); + [0, 2 ** -128, -(2 ** -128)].forEach(value => { + regular.observe(value); + zeroOnly.observe(value); + }); + expect((await snapshot(regular)).zeroCount).toBe(3); + expect((await snapshot(zeroOnly)).zeroCount).toBe(1); + }); + + test('marks an empty native histogram with zero threshold zero', async () => { + expect( + await snapshot(create({ nativeHistogramZeroThreshold: 0 })), + ).toMatchObject({ + positiveSpans: [{ offset: 0, length: 0 }], + positiveDeltas: [], + negativeSpans: [], + count: 0, + }); + }); + + test.each([NaN, Infinity, -Infinity, '1', undefined, null])( + 'rejects invalid observation %s without changing the histogram', + async value => { + const histogram = create(); + const before = await histogram.get(); + expect(() => histogram.observe(value)).toThrow( + 'Value is not a valid number', + ); + expect(await histogram.get()).toEqual(before); + }, + ); + + test('preserves classic histogram values when native buckets are enabled', async () => { + const classic = create({ nativeHistogramBucketFactor: 0 }); + const native = create(); + [-3, 0, 0.1, 1, 100].forEach(value => { + classic.observe(value); + native.observe(value); + }); + expect((await native.get()).values).toEqual((await classic.get()).values); + }); +}); + +describe('native bucket limits', () => { + test.each([ + [ + [1, 2, 4], + [ + [0, 1], + [1, 2], + ], + ], + [ + [0.125, 0.25, 0.5], + [ + [-1, 2], + [0, 1], + ], + ], + ])('coarsens %s while preserving all counts', async (values, counts) => { + const histogram = create({ nativeHistogramMaxBucketNumber: 2 }); + values.forEach(value => histogram.observe(value)); + const data = await snapshot(histogram); + expect(data.schema).toBe(-1); + expect(data.count).toBe(3); + expect(data.sum).toBe(values.reduce((sum, value) => sum + value, 0)); + expect(bucketCounts(data, 'positive')).toEqual(new Map(counts)); + }); + + test('applies the limit to positive and negative buckets together', async () => { + const histogram = create({ nativeHistogramMaxBucketNumber: 2 }); + [1, 2, -1].forEach(value => histogram.observe(value)); + const data = await snapshot(histogram); + expect(data.schema).toBe(-4); + expect(data.count).toBe(3); + expect(bucketCounts(data, 'positive')).toEqual( + new Map([ + [0, 1], + [1, 1], + ]), + ); + expect(bucketCounts(data, 'negative')).toEqual(new Map([[0, 1]])); + }); + + test('uses a default budget of 160 populated buckets', async () => { + const histogram = create({ nativeHistogramBucketFactor: 1.1 }); + for (let i = 0; i < 400; i++) histogram.observe(2 ** i); + const data = await snapshot(histogram); + const positive = bucketCounts(data, 'positive'); + expect(positive.size).toBeLessThanOrEqual(160); + expect([...positive.values()].reduce((sum, count) => sum + count, 0)).toBe( + 400, + ); + }); + + test('can disable the bucket limit', async () => { + const histogram = create({ + nativeHistogramBucketFactor: 1.005, + nativeHistogramMaxBucketNumber: 0, + }); + for (let i = 0; i < 200; i++) histogram.observe(2 ** i); + const data = await snapshot(histogram); + expect(data.schema).toBe(8); + expect(bucketCounts(data, 'positive').size).toBe(200); + }); + + test('limits and resets each label set independently', async () => { + const histogram = create({ + labelNames: ['route'], + nativeHistogramMaxBucketNumber: 2, + }); + [1, 2, 4].forEach(value => histogram.labels('/a').observe(value)); + histogram.labels('/b').observe(1); + expect( + (await histogram.get()).nativeHistograms.map(data => data.schema), + ).toEqual([-1, 0]); + histogram.zero({ route: '/a' }); + const reset = (await histogram.get()).nativeHistograms[0]; + expect(reset).toMatchObject({ schema: 0, count: 0, sum: 0 }); + }); +}); + +describe('native histogram lifecycle', () => { + test('supports labels, timers, zero, remove, and reset', async () => { + jest.useFakeTimers(); + jest.setSystemTime(1000); + const histogram = create({ labelNames: ['method', 'code'] }); + histogram.zero({ method: 'GET', code: 200 }); + expect((await snapshot(histogram)).labels).toEqual({ + method: 'GET', + code: '200', + }); + const end = histogram.labels('GET', '200').startTimer(); + jest.advanceTimersByTime(500); + expect(end()).toBe(0.5); + expect(await snapshot(histogram)).toMatchObject({ + count: 1, + sum: 0.5, + createdTimestamp: 1, + }); + histogram.remove({ method: 'GET', code: '200' }); + expect((await histogram.get()).nativeHistograms).toEqual([]); + histogram.observe({ method: 'POST', code: '201' }, 1); + histogram.reset(); + expect((await histogram.get()).nativeHistograms).toEqual([]); + histogram.observe({ method: 'POST', code: '201' }, 2); + expect(await snapshot(histogram)).toMatchObject({ + count: 1, + sum: 2, + createdTimestamp: 1.5, + }); + }); + + test('collects asynchronous observations before taking a snapshot', async () => { + const histogram = create({ + async collect() { + await Promise.resolve(); + this.observe(3); + }, + }); + expect(await snapshot(histogram)).toMatchObject({ count: 1, sum: 3 }); + }); + + test('returns independent JSON snapshots', async () => { + const histogram = create({ labelNames: ['route'] }); + const labels = { route: '/a' }; + histogram.observe(labels, 2); + labels.route = '/changed'; + const data = await snapshot(histogram); + data.labels.route = '/also_changed'; + data.positiveSpans[0].offset = 100; + data.positiveDeltas[0] = 100; + expect(await snapshot(histogram)).toMatchObject({ + labels: { route: '/a' }, + positiveSpans: [{ offset: 1, length: 1 }], + positiveDeltas: [1], + }); + }); + + test('retains at most ten independent timestamped exemplars', async () => { + jest.useFakeTimers(); + jest.setSystemTime(1234); + const register = new Registry(Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE); + const histogram = create({ enableExemplars: true, registers: [register] }); + const exemplarLabels = { trace_id: 'trace' }; + for (let i = 0; i < 12; i++) { + histogram.observe({ value: i, exemplarLabels }); + } + exemplarLabels.trace_id = 'changed'; + const data = await snapshot(histogram); + expect(data.exemplars).toEqual( + Array.from({ length: 10 }, (_, i) => { + return { + labelSet: { trace_id: 'trace' }, + value: i + 2, + timestamp: 1.234, + }; + }), + ); + data.exemplars[0].labelSet.trace_id = 'mutated'; + expect((await snapshot(histogram)).exemplars[0].labelSet.trace_id).toBe( + 'trace', + ); + }); + + test('supports exemplars on label-bound timers', async () => { + jest.useFakeTimers(); + const register = new Registry(Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE); + const histogram = create({ + registers: [register], + labelNames: ['method'], + enableExemplars: true, + }); + const end = histogram.labels('GET').startTimer(); + jest.advanceTimersByTime(200); + expect(end(undefined, { trace_id: 'timer' })).toBe(0.2); + expect((await snapshot(histogram)).exemplars[0]).toMatchObject({ + value: 0.2, + labelSet: { trace_id: 'timer' }, + }); + }); +}); diff --git a/test/nativeHistogramTypes.ts b/test/nativeHistogramTypes.ts new file mode 100644 index 00000000..bbb777ac --- /dev/null +++ b/test/nativeHistogramTypes.ts @@ -0,0 +1,142 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import { + AggregatorRegistry, + ClusterRegistry, + Histogram, + NativeHistogramValue, + PrometheusProtobufContentType, + Pushgateway, + Registry, + RegistryContentType, + WorkerRegistry, + prometheusProtobufContentType, +} from '../index'; + +const registry = new Registry(Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE); +const histogram = new Histogram({ + name: 'typescript_native_histogram', + help: 'Native histogram TypeScript test', + registers: [registry], + labelNames: ['route'] as const, + buckets: [], + nativeHistogramBucketFactor: 1.1, + nativeHistogramZeroThreshold: 0, + nativeHistogramMaxBucketNumber: 160, +}); + +histogram.observe({ route: '/' }, 0.2); +histogram.labels('/').observe(0.3); +histogram.zero({ route: '/' }); +const end = histogram.startTimer({ route: '/' }, { trace_id: 'abc' }); +const elapsed: number = end(undefined, { span_id: 'def' }); +const childElapsed: number = histogram.labels('/').startTimer()(); +void elapsed; +void childElapsed; + +const type: PrometheusProtobufContentType = prometheusProtobufContentType; +const binary: Promise = registry.metrics(); +const oneText: Promise = registry.getMetricsAsString(histogram); +const singleText: Promise = registry.getSingleMetricAsString( + 'typescript_native_histogram', +); +const merged: Promise = Registry.merge([registry]).metrics(); +const aggregate: Promise = Registry.aggregate([], type).metrics(); +const switched: Promise = new Registry() + .setContentType(type) + .metrics(); +const text: Promise = new Registry().metrics(); +const openMetrics: Promise = new Registry( + Registry.OPENMETRICS_CONTENT_TYPE, +).metrics(); +const unknownFormat: Registry = registry; +const unknownBody: Promise = unknownFormat.metrics(); +void [ + binary, + oneText, + singleText, + merged, + aggregate, + switched, + text, + openMetrics, + unknownBody, +]; + +const cluster = new ClusterRegistry(type); +const workers = new WorkerRegistry(type); +const clusterBytes: Promise = cluster.clusterMetrics(); +const workerBytes: Promise = workers.workerMetrics(); +const switchedClusterBytes: Promise = new ClusterRegistry() + .setContentType(type) + .clusterMetrics(); +const switchedWorkerBytes: Promise = new WorkerRegistry() + .setContentType(type) + .workerMetrics(); +const switchedAggregatorBytes: Promise = new AggregatorRegistry() + .setContentType(type) + .clusterMetrics(); +ClusterRegistry.setRegistries([registry, new Registry()]); +WorkerRegistry.setRegistries(registry); +void [ + clusterBytes, + workerBytes, + switchedClusterBytes, + switchedWorkerBytes, + switchedAggregatorBytes, +]; + +new Pushgateway('http://localhost:9091', registry); +new Histogram({ + name: 'typescript_native_exemplar_histogram', + help: 'Native exemplars', + registers: [registry], + labelNames: ['route'] as const, + nativeHistogramBucketFactor: 1.1, + enableExemplars: true, +}).observe({ + labels: { route: '/' }, + value: 1, + exemplarLabels: { trace_id: 'abc' }, +}); + +async function nativeSnapshotsAreTyped() { + const metric = await histogram.get(); + const snapshot: NativeHistogramValue | undefined = + metric.nativeHistograms?.[0]; + if (snapshot) { + const schema: number = snapshot.schema; + const count: number = snapshot.count; + const delta: number | undefined = snapshot.positiveDeltas[0]; + void [schema, count, delta]; + } + const allMetrics = await registry.getMetricsAsJSON(); + const natives: NativeHistogramValue[] | undefined = + allMetrics[0].nativeHistograms; + void natives; +} +void nativeSnapshotsAreTyped; + +// @ts-expect-error A protobuf registry returns bytes, not text. +const invalidText: Promise = registry.metrics(); +void invalidText; +new Histogram({ + name: 'invalid', + help: 'invalid', + // @ts-expect-error Native bucket factors are numeric. + nativeHistogramBucketFactor: '1.1', +}); +// @ts-expect-error Label names remain checked for native histograms. +histogram.observe({ method: 'GET' }, 1); diff --git a/test/protobufTest.js b/test/protobufTest.js new file mode 100644 index 00000000..d0e763d6 --- /dev/null +++ b/test/protobufTest.js @@ -0,0 +1,522 @@ +// Copyright The Prometheus Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +'use strict'; + +const client = require('../index'); +const { Registry, Counter, Gauge, Histogram, Summary, Pushgateway } = client; +const { decodeMetricFamilies } = require('./helpers/nativeHistogram'); +const nock = require('nock'); + +const contentType = Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE; + +function createHistogram(register, config = {}) { + return new Histogram({ + name: 'native_duration_seconds', + help: 'Duration of a request', + registers: [register], + nativeHistogramBucketFactor: 1.1, + ...config, + }); +} + +afterEach(() => { + jest.useRealTimers(); + nock.cleanAll(); +}); + +describe('Prometheus protobuf exposition', () => { + test('exposes the content type and returns an empty Buffer for an empty registry', async () => { + const register = new Registry(contentType); + expect(register.contentType).toBe(client.prometheusProtobufContentType); + expect(contentType).toBe( + 'application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; encoding=delimited', + ); + const body = await register.metrics(); + expect(Buffer.isBuffer(body)).toBe(true); + expect(body.length).toBe(0); + expect(new Registry().setContentType(contentType).contentType).toBe( + contentType, + ); + }); + + test('encodes every existing metric type as length-delimited MetricFamily messages', async () => { + const register = new Registry(contentType); + const counter = new Counter({ + name: 'requests_total', + help: 'Requests', + registers: [register], + }); + const gauge = new Gauge({ + name: 'temperature', + help: 'Temperature', + registers: [register], + }); + const histogram = new Histogram({ + name: 'classic', + help: 'Classic histogram', + buckets: [1, 2], + registers: [register], + }); + const summary = new Summary({ + name: 'summary', + help: 'Summary', + percentiles: [0.5], + registers: [register], + }); + counter.inc(4.5); + gauge.set(-7.5); + histogram.observe(1.5); + summary.observe(1); + summary.observe(3); + + const families = decodeMetricFamilies(await register.metrics()); + expect(families).toHaveLength(4); + expect(families[0]).toEqual({ + name: 'requests_total', + help: 'Requests', + type: 'COUNTER', + metric: [{ counter: { value: 4.5 } }], + }); + expect(families[1].metric).toEqual([{ gauge: { value: -7.5 } }]); + expect(families[2]).toMatchObject({ + type: 'HISTOGRAM', + metric: [ + { + histogram: { + sampleCount: 1, + sampleSum: 1.5, + bucket: [ + { upperBound: 1, cumulativeCount: 0 }, + { upperBound: 2, cumulativeCount: 1 }, + { upperBound: Infinity, cumulativeCount: 1 }, + ], + }, + }, + ], + }); + expect(families[3]).toMatchObject({ + type: 'SUMMARY', + metric: [ + { + summary: { + sampleCount: 2, + sampleSum: 4, + quantile: [{ quantile: 0.5, value: 2 }], + }, + }, + ], + }); + }); + + test('encodes native and classic buckets together, with labels and a creation timestamp', async () => { + jest.useFakeTimers(); + jest.setSystemTime(1234); + const register = new Registry(contentType); + register.setDefaultLabels({ service: 'frontend', route: 'default' }); + const histogram = createHistogram(register, { + buckets: [1, 5], + labelNames: ['route'], + help: 'Unicode: café "help"\nwith newline', + }); + const labels = { route: '/api/π\n"quoted"\\path' }; + [1, -2, 0].forEach(value => histogram.observe(labels, value)); + + const [family] = decodeMetricFamilies(await register.metrics()); + expect(family.name).toBe('native_duration_seconds'); + expect(family.help).toBe('Unicode: café "help"\nwith newline'); + expect(family.metric).toEqual([ + { + label: [ + { name: 'route', value: labels.route }, + { name: 'service', value: 'frontend' }, + ], + histogram: { + sampleCount: 3, + sampleSum: -1, + bucket: [ + { upperBound: 1, cumulativeCount: 3 }, + { upperBound: 5, cumulativeCount: 3 }, + { upperBound: Infinity, cumulativeCount: 3 }, + ], + schema: 3, + zeroThreshold: 2 ** -128, + zeroCount: 1, + positiveSpan: [{ offset: 0, length: 1 }], + positiveDelta: [1], + negativeSpan: [{ offset: 8, length: 1 }], + negativeDelta: [1], + createdTimestamp: { seconds: 1, nanos: 234000000 }, + }, + }, + ]); + + const text = await register.getMetricsAsString(histogram); + expect(typeof text).toBe('string'); + const textRegistry = new Registry(); + textRegistry.setDefaultLabels({ service: 'frontend', route: 'default' }); + expect(text).toEqual(await textRegistry.getMetricsAsString(histogram)); + expect(text).toContain(`${histogram.name}_count`); + expect(await register.getSingleMetricAsString(histogram.name)).toBe(text); + expect(await register.metrics()).toBeInstanceOf(Buffer); + }); + + test.each([ + [[0.5], [1.5], 1, [0.5, 1, 1]], + [[0.5], [0.5, 1.5], 1.5, [1, 1.5, 1.5]], + ])( + 'preserves fractional classic histogram counts after averaging %p and %p', + async (firstValues, secondValues, expectedCount, expectedBuckets) => { + const snapshots = await Promise.all( + [firstValues, secondValues].map(async values => { + const histogram = new Histogram({ + name: 'averaged_classic', + help: 'Averaged classic histogram', + buckets: [1, 2], + aggregator: 'average', + registers: [], + }); + values.forEach(value => histogram.observe(value)); + return [await histogram.get()]; + }), + ); + const register = Registry.aggregate(snapshots, contentType); + const data = decodeMetricFamilies(await register.metrics())[0].metric[0] + .histogram; + expect(data.sampleCountFloat ?? data.sampleCount).toBe(expectedCount); + expect( + data.bucket.map( + bucket => bucket.cumulativeCountFloat ?? bucket.cumulativeCount, + ), + ).toEqual(expectedBuckets); + }, + ); + + test('rejects fractional summary counts that protobuf cannot represent', async () => { + const snapshots = await Promise.all( + [[1], [1, 2]].map(async values => { + const summary = new Summary({ + name: 'averaged_summary', + help: 'Averaged summary', + aggregator: 'average', + registers: [], + }); + values.forEach(value => summary.observe(value)); + return [await summary.get()]; + }), + ); + const register = Registry.aggregate(snapshots, contentType); + await expect(register.metrics()).rejects.toThrow( + 'Prometheus protobuf requires an integer sample count for summary averaged_summary', + ); + }); + + test('can expose native buckets without classic buckets', async () => { + const register = new Registry(contentType); + const histogram = createHistogram(register, { buckets: [] }); + histogram.observe(2); + const data = decodeMetricFamilies(await register.metrics())[0].metric[0] + .histogram; + expect(data).not.toHaveProperty('bucket'); + expect(data).toMatchObject({ + sampleCount: 1, + sampleSum: 2, + positiveDelta: [1], + schema: 3, + }); + }); + + test('preserves zero-valued fields and the empty no-op span on the wire', async () => { + jest.useFakeTimers(); + jest.setSystemTime(0); + const register = new Registry(contentType); + createHistogram(register, { + buckets: [], + nativeHistogramBucketFactor: 2, + nativeHistogramZeroThreshold: 0, + }); + expect( + decodeMetricFamilies(await register.metrics())[0].metric[0].histogram, + ).toEqual({ + sampleCount: 0, + sampleSum: 0, + schema: 0, + zeroThreshold: 0, + zeroCount: 0, + positiveSpan: [{ offset: 0, length: 0 }], + // Timestamp is proto3: an empty message represents the Unix epoch. + createdTimestamp: {}, + }); + }); + + test('encodes negative schemas, large counts, offsets, and negative deltas', async () => { + const register = new Registry(contentType); + const histogram = createHistogram(register, { + buckets: [], + nativeHistogramBucketFactor: 4, + nativeHistogramZeroThreshold: 0, + }); + for (let i = 0; i < 300; i++) histogram.observe(0.25); + histogram.observe(2 ** 200); + histogram.observe(-(2 ** -1000)); + const data = decodeMetricFamilies(await register.metrics())[0].metric[0] + .histogram; + expect(data).toMatchObject({ + sampleCount: 302, + schema: -1, + positiveSpan: [ + { offset: -1, length: 1 }, + { offset: 100, length: 1 }, + ], + positiveDelta: [300, -299], + negativeSpan: [{ offset: -500, length: 1 }], + negativeDelta: [1], + }); + }); + + test('encodes uint64 counts and sint64 deltas beyond 32 bits without truncating them', async () => { + const register = new Registry(contentType); + const histogram = createHistogram(new Registry(contentType), { + buckets: [], + }); + histogram.observe(1); + const data = await histogram.get(); + const count = 2 ** 40; + data.nativeHistograms[0].count = count; + data.nativeHistograms[0].sum = count; + data.nativeHistograms[0].positiveDeltas[0] = count; + data.values.forEach(value => { + value.value = count; + }); + register.registerMetric({ name: data.name, get: () => data }); + const decoded = decodeMetricFamilies(await register.metrics())[0].metric[0] + .histogram; + expect(decoded.sampleCount).toBe(count); + expect(decoded.positiveDelta).toEqual([count]); + }); + + test('uses default labels consistently in JSON and protobuf without changing the metric', async () => { + const register = new Registry(contentType); + register.setDefaultLabels({ service: 'frontend', code: 'default' }); + const histogram = createHistogram(register, { labelNames: ['code'] }); + histogram.observe({ code: 200 }, 0.5); + const [json] = await register.getMetricsAsJSON(); + expect(json.nativeHistograms[0].labels).toEqual({ + code: '200', + service: 'frontend', + }); + expect((await histogram.get()).nativeHistograms[0].labels).toEqual({ + code: '200', + }); + json.nativeHistograms[0].labels.service = 'changed'; + expect( + decodeMetricFamilies(await register.metrics())[0].metric, + ).toHaveLength(1); + expect( + (await register.getMetricsAsJSON())[0].nativeHistograms[0].labels.service, + ).toBe('frontend'); + }); + + test.each( + [ + [Counter, 'inc'], + [Gauge, 'set'], + [Summary, 'observe'], + [Histogram, 'observe'], + ].flatMap(([Metric, method]) => + [undefined, null].map(value => [Metric, method, value]), + ), + )( + 'uses default labels for %p.%s with a nullish label value %p', + async (Metric, method, value) => { + const register = new Registry(); + register.setDefaultLabels({ service: 'frontend', team: 'platform' }); + const metric = new Metric({ + name: 'default_labels', + help: 'Default labels', + labelNames: ['service'], + registers: [register], + nativeHistogramBucketFactor: 1.1, + }); + metric[method]({ service: value }, 0.5); + expect(await register.metrics()).toContain('service="frontend"'); + register.setContentType(contentType); + const [family] = decodeMetricFamilies(await register.metrics()); + expect(family.metric).toHaveLength(1); + expect(family.metric[0].label).toEqual([ + { name: 'service', value: 'frontend' }, + { name: 'team', value: 'platform' }, + ]); + const [json] = await register.getMetricsAsJSON(); + expect( + json.values.every(sample => sample.labels.service === 'frontend'), + ).toBe(true); + }, + ); + + test('keeps distinct label sets from different workers when they contain separators', async () => { + const register = new Registry(contentType); + const other = new Registry(contentType); + const histogram = createHistogram(register, { labelNames: ['a', 'b'] }); + const otherHistogram = createHistogram(other, { labelNames: ['a', 'b'] }); + histogram.observe({ a: 'a|b', b: 'c' }, 1); + otherHistogram.observe({ a: 'a', b: 'b|c' }, 2); + const aggregated = Registry.aggregate( + [await register.getMetricsAsJSON(), await other.getMetricsAsJSON()], + contentType, + ); + expect( + decodeMetricFamilies(await aggregated.metrics())[0].metric, + ).toHaveLength(2); + }); + + test.each([ + Registry.PROMETHEUS_CONTENT_TYPE, + Registry.OPENMETRICS_CONTENT_TYPE, + ])('retains classic text output for %s registries', async type => { + const register = new Registry(type); + const histogram = createHistogram(register, { buckets: [1, 2] }); + histogram.observe(1.5); + const text = await register.metrics(); + expect(typeof text).toBe('string'); + expect(text).toContain('native_duration_seconds_bucket{le="2"} 1'); + expect(text).toContain('native_duration_seconds_sum 1.5'); + expect(text).not.toContain('schema'); + expect(text).not.toContain('NaN'); + }); + + test('awaits the collector exactly once per protobuf scrape', async () => { + const register = new Registry(contentType); + const collect = jest.fn(async function () { + await Promise.resolve(); + this.observe(2); + }); + createHistogram(register, { collect }); + const [family] = decodeMetricFamilies(await register.metrics()); + expect(collect).toHaveBeenCalledTimes(1); + expect(family.metric[0].histogram.sampleCount).toBe(1); + }); + + test('keeps counter names intact when a metric is shared with an OpenMetrics registry', async () => { + const register = new Registry(contentType); + const openMetrics = new Registry(Registry.OPENMETRICS_CONTENT_TYPE); + const counter = new Counter({ + name: 'shared_requests_total', + help: 'Requests', + registers: [register, openMetrics], + }); + counter.inc(); + expect(await openMetrics.metrics()).toContain('shared_requests_total 1'); + expect(decodeMetricFamilies(await register.metrics())[0].name).toBe( + 'shared_requests_total', + ); + expect(counter.name).toBe('shared_requests_total'); + expect(await openMetrics.metrics()).toContain('shared_requests_total 1'); + }); + + test('keeps each pending scrape in the format selected when it started', async () => { + const register = new Registry(); + let finishCollecting; + const collected = new Promise(resolve => { + finishCollecting = resolve; + }); + new Counter({ + name: 'async_requests_total', + help: 'Requests', + registers: [register], + async collect() { + await collected; + }, + }).inc(1); + const text = register.metrics(); + register.setContentType(contentType); + const binary = register.metrics(); + finishCollecting(); + expect(await text).toContain('async_requests_total 1'); + expect(decodeMetricFamilies(await binary)[0].metric[0].counter.value).toBe( + 1, + ); + }); + + test('merges protobuf registries and retains native data', async () => { + const one = new Registry(contentType); + const two = new Registry(contentType); + createHistogram(one).observe(1); + new Gauge({ name: 'gauge', help: 'gauge', registers: [two] }).set(5); + const merged = Registry.merge([one, two]); + expect(merged.contentType).toBe(contentType); + expect( + decodeMetricFamilies(await merged.metrics()).map(metric => metric.name), + ).toEqual(['native_duration_seconds', 'gauge']); + expect(() => Registry.merge([one, new Registry()])).toThrow( + 'same content type', + ); + }); + + test('exports exemplars for counters, classic buckets, and native histograms', async () => { + jest.useFakeTimers(); + jest.setSystemTime(1234); + const register = new Registry(contentType); + new Counter({ + name: 'requests_total', + help: 'Requests', + enableExemplars: true, + registers: [register], + }).inc({ value: 2, exemplarLabels: { trace_id: 'counter' } }); + createHistogram(register, { enableExemplars: true, buckets: [1] }).observe({ + value: 0.5, + exemplarLabels: { trace_id: 'histogram' }, + }); + const [counter, histogram] = decodeMetricFamilies(await register.metrics()); + const timestamp = { seconds: 1, nanos: 234000000 }; + expect(counter.metric[0].counter.exemplar).toEqual({ + label: [{ name: 'trace_id', value: 'counter' }], + value: 2, + timestamp, + }); + const exemplar = { + label: [{ name: 'trace_id', value: 'histogram' }], + value: 0.5, + timestamp, + }; + expect(histogram.metric[0].histogram.exemplars).toEqual([exemplar]); + expect(histogram.metric[0].histogram.bucket[0].exemplar).toEqual(exemplar); + }); + + test('rejects inconsistent classic/native sample counts', async () => { + const register = new Registry(contentType); + const histogram = createHistogram(new Registry(contentType)); + histogram.observe(1); + const data = await histogram.get(); + data.nativeHistograms[0].count++; + register.registerMetric({ name: data.name, get: () => data }); + await expect(register.metrics()).rejects.toThrow( + 'Classic and native histogram counts or sums differ', + ); + }); + + test('sends protobuf to Pushgateway with the matching content type', async () => { + const register = new Registry(contentType); + createHistogram(register).observe(1); + const body = await register.metrics(); + const gateway = nock('http://localhost:9091') + .matchHeader('Content-Type', contentType) + .put('/metrics/job/native', body) + .reply(202); + await new Pushgateway('http://localhost:9091', register).push({ + jobName: 'native', + }); + expect(gateway.isDone()).toBe(true); + }); +}); diff --git a/test/registerTest.js b/test/registerTest.js index b9cfcdd6..14228c62 100644 --- a/test/registerTest.js +++ b/test/registerTest.js @@ -33,6 +33,21 @@ describe('Register', () => { }).toThrow(expectedContentTypeErrStr); }); + it.each([ + Registry.PROMETHEUS_CONTENT_TYPE, + Registry.OPENMETRICS_CONTENT_TYPE, + Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE, + ])('accepts %s in both the constructor and setter', contentType => { + expect(new Registry(contentType).contentType).toBe(contentType); + const registry = new Registry(); + expect(registry.setContentType(contentType)).toBe(registry); + expect(registry.contentType).toBe(contentType); + expect(() => registry.setContentType(contentTypeTestStr)).toThrow( + expectedContentTypeErrStr, + ); + expect(registry.contentType).toBe(contentType); + }); + describe.each([ ['Prometheus', Registry.PROMETHEUS_CONTENT_TYPE], ['OpenMetrics', Registry.OPENMETRICS_CONTENT_TYPE], @@ -1106,6 +1121,41 @@ describe('Register', () => { }); }); + it('does not rename counters shared with another registry when scraped', async () => { + const other = new Registry(); + const counter = new Counter({ + name: 'shared_requests_total', + help: 'Requests', + registers: [register, other], + }); + counter.inc(); + const first = await register.metrics(); + expect(counter.name).toBe('shared_requests_total'); + expect(await register.metrics()).toBe(first); + expect(await other.metrics()).toContain('shared_requests_total 1'); + expect(await register.getSingleMetricAsString(counter.name)).toContain( + 'shared_requests_total 1', + ); + }); + + it.each([null, undefined])( + 'uses registry defaults for a shared histogram label with value %p', + async value => { + register.setDefaultLabels({ service: 'frontend' }); + const histogram = new Histogram({ + name: 'default_labels', + help: 'Default labels', + labelNames: ['service'], + registers: [register], + }); + histogram.observe({ service: value }, 0.5); + const before = await histogram.getForPromString(); + const labels = { ...before.values[0].sharedLabels }; + expect(await register.metrics()).toContain('service="frontend"'); + expect(before.values[0].sharedLabels).toEqual(labels); + }, + ); + function getMetric(name) { name = name || 'test_metric'; return { diff --git a/test/workerTest.js b/test/workerTest.js index 3bdf6c0b..e9063c82 100644 --- a/test/workerTest.js +++ b/test/workerTest.js @@ -120,6 +120,51 @@ describe.each([ } }); + it('keeps pending scrapes in their original content types', async () => { + const { Histogram } = require('../index'); + const { decodeMetricFamilies } = require('./helpers/nativeHistogram'); + const histogram = new Histogram({ + name: 'pending_native', + help: 'Pending native histogram', + registers: [], + nativeHistogramBucketFactor: 1.1, + }); + histogram.observe(1); + const snapshot = await histogram.get(); + const threadId = 212; + const name = `@prometheus-io/client:worker:${threadId}`; + const channel = new BroadcastChannel(name).unref(); + try { + announcementChannel.postMessage({ type: ANNOUNCEMENT, name, threadId }); + for ( + let attempt = 0; + attempt < 100 && AggregatorRegistry.workerCount() === 0; + attempt++ + ) { + await delay(5); + } + expect(AggregatorRegistry.workerCount()).toBeGreaterThan(0); + announcementChannel.addEventListener('message', event => { + if (event.data.type !== GET_METRICS_REQ) return; + channel.postMessage({ + type: GET_METRICS_RES, + requestId: event.data.requestId, + threadId, + metrics: [[snapshot]], + }); + }); + const text = registry.workerMetrics(); + registry.setContentType(Registry.PROMETHEUS_PROTOBUF_CONTENT_TYPE); + const binary = registry.workerMetrics(); + const [textBody, binaryBody] = await Promise.all([text, binary]); + expect(textBody).toContain('pending_native_count 1'); + const [family] = decodeMetricFamilies(binaryBody); + expect(family.metric[0].histogram.sampleCount).toBe(1); + } finally { + channel.close(); + } + }); + it('aggregates worker responses in thread id order', async () => { const responders = [1, 2, 3].map(threadId => { const name = `@prometheus-io/client:worker:${threadId}`;