diff --git a/.changeset/metric-no-recorded-value.md b/.changeset/metric-no-recorded-value.md new file mode 100644 index 0000000000..a123d3f682 --- /dev/null +++ b/.changeset/metric-no-recorded-value.md @@ -0,0 +1,15 @@ +--- +'@hyperdx/common-utils': patch +'@hyperdx/app': patch +'@hyperdx/api': patch +--- + +fix: leave metric points with no recorded value out of charts and alerts + +When a scraped series goes away or a scrape fails, the collector's `prometheus` +receiver sends a point flagged "no recorded value" (a Prometheus staleness +marker), and the ClickHouse exporter stores it as zeros. Metric charts and +alerts read it as a real value: a gauge's `min` and `avg` dropped and `count` +included the series that went away, and a counter or histogram that came back +after the marker counted its whole total again. Metric queries now skip points +with this flag. diff --git a/packages/common-utils/src/__tests__/metricNoRecordedValue.int.test.ts b/packages/common-utils/src/__tests__/metricNoRecordedValue.int.test.ts new file mode 100644 index 0000000000..904213f539 --- /dev/null +++ b/packages/common-utils/src/__tests__/metricNoRecordedValue.int.test.ts @@ -0,0 +1,366 @@ +import type { ClickHouseClient } from '@clickhouse/client'; +import { createClient } from '@clickhouse/client'; + +import { ClickhouseClient as HdxClickhouseClient } from '@/clickhouse/node'; +import { Metadata, MetadataCache } from '@/core/metadata'; +import { + ChartConfigWithOptDateRange, + DisplayType, + MetricsDataType, +} from '@/types'; + +// A point with OTLP FLAG_NO_RECORDED_VALUE (Flags bit 0) carries no value. The +// collector's prometheus receiver sends one as a staleness marker when a series +// goes away or a scrape fails, and the ClickHouse exporter stores it as zeros. +describe('metric points with no recorded value (integration)', () => { + let client: ClickHouseClient; + let hdxClient: HdxClickhouseClient; + let metadata: Metadata; + + const DATABASE = 'default'; + const GAUGE_TABLE = 'no_recorded_value_gauge_int_test'; + const SUM_TABLE = 'no_recorded_value_sum_int_test'; + const HIST_TABLE = 'no_recorded_value_histogram_int_test'; + const EXP_HIST_TABLE = 'no_recorded_value_exp_histogram_int_test'; + + const NO_RECORDED_VALUE = 1; + + // Inserts are server-local strings; the CI ClickHouse runs in UTC. + const insertTs = (minute: number) => `2025-04-15 10:0${minute}:00`; + const bucket = (minute: number) => `2025-04-15T10:0${minute}:00Z`; + + // Columns of docker/otel-collector/schema/seed/00003_otel_metrics.sql that + // every metric table shares. + const COMMON_COLUMNS = ` + ResourceAttributes Map(LowCardinality(String), String), + ResourceSchemaUrl String, + ScopeName String, + ScopeVersion String, + ScopeAttributes Map(LowCardinality(String), String), + ScopeDroppedAttrCount UInt32, + ScopeSchemaUrl String, + ServiceName LowCardinality(String), + MetricName LowCardinality(String), + MetricDescription String, + MetricUnit String, + Attributes Map(LowCardinality(String), String), + StartTimeUnix DateTime, + TimeUnix DateTime, + Flags UInt32`; + const ENGINE = `ENGINE = MergeTree PARTITION BY toDate(TimeUnix) + ORDER BY (ServiceName, MetricName, toStartOfHour(TimeUnix), cityHash64(Attributes), TimeUnix)`; + + const commonFields = ( + metricName: string, + minute: number, + service: string, + flags = 0, + ) => ({ + ResourceAttributes: { 'service.name': service }, + ResourceSchemaUrl: '', + ScopeName: '', + ScopeVersion: '', + ScopeAttributes: {}, + ScopeDroppedAttrCount: 0, + ScopeSchemaUrl: '', + ServiceName: service, + MetricName: metricName, + MetricDescription: '', + MetricUnit: '', + Attributes: {}, + StartTimeUnix: insertTs(0), + TimeUnix: insertTs(minute), + Flags: flags, + }); + + const metricTables = { + gauge: GAUGE_TABLE, + sum: SUM_TABLE, + histogram: HIST_TABLE, + summary: 'unused', + 'exponential histogram': EXP_HIST_TABLE, + }; + + const runConfig = async ( + metricType: MetricsDataType, + metricName: string, + select: { aggFn: string; level?: number }, + ) => { + const result = await hdxClient.queryChartConfig({ + config: { + displayType: DisplayType.Line, + connection: 'test-connection', + metricTables, + from: { databaseName: DATABASE, tableName: '' }, + select: [ + { + ...select, + aggCondition: '', + aggConditionLanguage: 'sql', + valueExpression: 'Value', + alias: 'Value', + metricName, + metricType, + }, + ], + where: '', + whereLanguage: 'sql', + timestampValueExpression: 'TimeUnix', + dateRange: [new Date('2025-04-14'), new Date('2025-04-16')], + granularity: '1 minute', + } as ChartConfigWithOptDateRange, + metadata, + querySettings: undefined, + }); + return new Map( + (result.data as Array<{ __hdx_time_bucket: string; Value: unknown }>).map( + row => [row.__hdx_time_bucket, Number(row.Value)], + ), + ); + }; + + beforeAll(async () => { + const host = process.env.CLICKHOUSE_HOST || 'http://localhost:8123'; + const username = process.env.CLICKHOUSE_USER || 'default'; + const password = process.env.CLICKHOUSE_PASSWORD || ''; + + client = createClient({ url: host, username, password }); + hdxClient = new HdxClickhouseClient({ host, username, password }); + + await client.command({ + query: `CREATE OR REPLACE TABLE ${DATABASE}.${GAUGE_TABLE} ( + ${COMMON_COLUMNS}, + Value Float64 + ) ${ENGINE}`, + }); + await client.command({ + query: `CREATE OR REPLACE TABLE ${DATABASE}.${SUM_TABLE} ( + ${COMMON_COLUMNS}, + Value Float64, + AggregationTemporality Int32, + IsMonotonic Bool + ) ${ENGINE}`, + }); + await client.command({ + query: `CREATE OR REPLACE TABLE ${DATABASE}.${HIST_TABLE} ( + ${COMMON_COLUMNS}, + Count UInt64, + Sum Float64, + BucketCounts Array(UInt64), + ExplicitBounds Array(Float64), + Min Float64, + Max Float64, + AggregationTemporality Int32 + ) ${ENGINE}`, + }); + await client.command({ + query: `CREATE OR REPLACE TABLE ${DATABASE}.${EXP_HIST_TABLE} ( + ${COMMON_COLUMNS}, + Count UInt64, + Sum Float64, + Scale Int32, + ZeroCount UInt64, + PositiveOffset Int32, + PositiveBucketCounts Array(UInt64), + NegativeOffset Int32, + NegativeBucketCounts Array(UInt64), + Min Float64, + Max Float64, + AggregationTemporality Int32 + ) ${ENGINE}`, + }); + + // Two series report 10 every minute; svc-b goes away after minute 4 and + // leaves a marker at minute 5. + const gaugeRow = ( + minute: number, + service: string, + value = 10, + flags = 0, + ) => ({ + ...commonFields('gauge.cpu', minute, service, flags), + Value: value, + }); + await client.insert({ + table: `${DATABASE}.${GAUGE_TABLE}`, + values: [ + ...[0, 1, 2, 3, 4, 5].map(m => gaugeRow(m, 'svc-a')), + ...[0, 1, 2, 3, 4].map(m => gaugeRow(m, 'svc-b')), + gaugeRow(5, 'svc-b', 0, NO_RECORDED_VALUE), + ], + format: 'JSONEachRow', + }); + + const sumRow = + (metricName: string) => + (minute: number, value: number, flags = 0) => ({ + ...commonFields(metricName, minute, 'svc-a', flags), + Value: value, + AggregationTemporality: 2, + IsMonotonic: true, + }); + // A cumulative counter that goes up by 10 a minute. One scrape fails at + // minute 3, then the same series comes back. + const requests = sumRow('sum.requests'); + // A counter that restarts behind a marker, and the same restart without + // one. + const restart = sumRow('sum.restart'); + const restartControl = sumRow('sum.restart.control'); + await client.insert({ + table: `${DATABASE}.${SUM_TABLE}`, + values: [ + requests(0, 100), + requests(1, 110), + requests(2, 120), + requests(3, 0, NO_RECORDED_VALUE), + requests(4, 130), + restart(0, 100), + restart(1, 120), + restart(2, 0, NO_RECORDED_VALUE), + restart(3, 5), + restartControl(0, 100), + restartControl(1, 120), + restartControl(3, 5), + ], + format: 'JSONEachRow', + }); + + // Cumulative histogram: 10 observations in (0, 10] by minute 1, a marker + // at minute 2, then 5 new observations in (10, 30] by minute 3. + const histRow = (minute: number, bucketCounts: number[], flags = 0) => ({ + ...commonFields('hist.latency', minute, 'svc-a', flags), + Count: bucketCounts.reduce((a, b) => a + b, 0), + Sum: 0, + BucketCounts: bucketCounts, + ExplicitBounds: [10, 30], + Min: 0, + Max: 0, + AggregationTemporality: 2, + }); + await client.insert({ + table: `${DATABASE}.${HIST_TABLE}`, + values: [ + histRow(0, [0, 0, 0]), + histRow(1, [10, 0, 0]), + histRow(2, [0, 0, 0], NO_RECORDED_VALUE), + histRow(3, [10, 5, 0]), + ], + format: 'JSONEachRow', + }); + + // Cumulative native histogram at scale 3, where bucket 8 is (2, 2.18] and + // bucket 16 is (4, 4.36]: 10 observations in bucket 8 by minute 1, a + // marker at minute 2, then 5 new observations in bucket 16 by minute 3. + // The receiver writes a stale point with scale 0 and no buckets. + const expHistRow = (minute: number, buckets: number[], flags = 0) => ({ + ...commonFields('exphist.latency', minute, 'svc-a', flags), + Count: buckets.reduce((a, b) => a + b, 0), + Sum: 0, + Scale: flags ? 0 : 3, + ZeroCount: 0, + PositiveOffset: flags ? 0 : 8, + PositiveBucketCounts: buckets, + NegativeOffset: 0, + NegativeBucketCounts: [], + Min: 0, + Max: 0, + AggregationTemporality: 2, + }); + await client.insert({ + table: `${DATABASE}.${EXP_HIST_TABLE}`, + values: [ + expHistRow(0, []), + expHistRow(1, [10]), + expHistRow(2, [], NO_RECORDED_VALUE), + expHistRow(3, [10, 0, 0, 0, 0, 0, 0, 0, 5]), + ], + format: 'JSONEachRow', + }); + }); + + beforeEach(() => { + metadata = new Metadata(hdxClient, new MetadataCache()); + }); + + afterAll(async () => { + for (const table of [GAUGE_TABLE, SUM_TABLE, HIST_TABLE, EXP_HIST_TABLE]) { + await client.command({ + query: `DROP TABLE IF EXISTS ${DATABASE}.${table}`, + }); + } + await hdxClient.close(); + await client.close(); + }); + + it.each([ + ['min', 10], + ['avg', 10], + ['count', 1], + ])('leaves a gauge marker out of %s', async (aggFn, expected) => { + const values = await runConfig(MetricsDataType.Gauge, 'gauge.cpu', { + aggFn, + }); + expect(values.get(bucket(5))).toBe(expected); + }); + + it('keeps the gauge minute before the marker unchanged', async () => { + const counts = await runConfig(MetricsDataType.Gauge, 'gauge.cpu', { + aggFn: 'count', + }); + expect(counts.get(bucket(4))).toBe(2); + }); + + it('does not count a counter from zero after a marker', async () => { + const values = await runConfig(MetricsDataType.Sum, 'sum.requests', { + aggFn: 'increase', + }); + expect(values.get(bucket(4))).toBe(10); + expect([...values.values()].reduce((a, b) => a + b, 0)).toBe(30); + }); + + // The Sum query clamps a counter decrease to 0, so the restart minute reads + // 0 with and without the marker. + it('treats a counter restart behind a marker like any other counter reset', async () => { + const select = { aggFn: 'increase' }; + expect(await runConfig(MetricsDataType.Sum, 'sum.restart', select)).toEqual( + await runConfig(MetricsDataType.Sum, 'sum.restart.control', select), + ); + }); + + it('does not count a histogram from zero after a marker', async () => { + const counts = await runConfig(MetricsDataType.Histogram, 'hist.latency', { + aggFn: 'count', + }); + expect(counts.get(bucket(1))).toBe(10); + expect(counts.get(bucket(3))).toBe(5); + }); + + it('computes histogram quantiles from the observations after a marker only', async () => { + const p50 = await runConfig(MetricsDataType.Histogram, 'hist.latency', { + aggFn: 'quantile', + level: 0.5, + }); + expect(p50.get(bucket(3))).toBeGreaterThan(10); + expect(p50.get(bucket(3))).toBeLessThanOrEqual(30); + }); + + it('does not count an exponential histogram from zero after a marker', async () => { + const counts = await runConfig( + MetricsDataType.ExponentialHistogram, + 'exphist.latency', + { aggFn: 'count' }, + ); + expect(counts.get(bucket(1))).toBe(10); + expect(counts.get(bucket(3))).toBe(5); + }); + + it('computes exponential histogram quantiles at the series scale from the observations after a marker only', async () => { + const p50 = await runConfig( + MetricsDataType.ExponentialHistogram, + 'exphist.latency', + { aggFn: 'quantile', level: 0.5 }, + ); + expect(p50.get(bucket(3))).toBeGreaterThan(4); + expect(p50.get(bucket(3))).toBeLessThanOrEqual(2 ** (17 / 8)); + }); +}); diff --git a/packages/common-utils/src/__tests__/renderChartConfig.test.ts b/packages/common-utils/src/__tests__/renderChartConfig.test.ts index 0a6761b2c0..c025e687ee 100644 --- a/packages/common-utils/src/__tests__/renderChartConfig.test.ts +++ b/packages/common-utils/src/__tests__/renderChartConfig.test.ts @@ -927,6 +927,118 @@ describe('renderChartConfig', () => { ); }); + describe('metric points with no recorded value', () => { + const NO_RECORDED_VALUE_FILTER = 'bitAnd(Flags, 1) = 0'; + + const renderMetricSql = async ( + metricType: MetricsDataType, + select: { aggFn: 'avg' | 'count' | 'quantile'; level?: number }, + ) => { + const rendered = await renderChartConfig( + { + ...gaugeConfiguration, + select: [ + { + ...select, + aggCondition: '', + aggConditionLanguage: 'sql', + valueExpression: 'Value', + metricName: 'test.metric', + metricType, + }, + ], + } as ChartConfigWithOptDateRange, + mockMetadata, + querySettings, + ); + return parameterizedQueryToSql(rendered).replace(/\s+/g, ' '); + }; + + const cases = [ + [MetricsDataType.Gauge, { aggFn: 'avg' }], + [MetricsDataType.Sum, { aggFn: 'avg' }], + [MetricsDataType.Histogram, { aggFn: 'quantile', level: 0.5 }], + [MetricsDataType.Histogram, { aggFn: 'count' }], + [MetricsDataType.ExponentialHistogram, { aggFn: 'quantile', level: 0.5 }], + [MetricsDataType.ExponentialHistogram, { aggFn: 'count' }], + ] as const; + + it.each(cases)( + 'leaves flagged points out of the %s source scan (%o)', + async (metricType, select) => { + mockMetadata.getColumn = jest + .fn() + .mockImplementation(async ({ column }) => + column === 'Flags' ? { name: 'Flags', type: 'UInt32' } : undefined, + ); + const sql = await renderMetricSql(metricType, select); + expect(sql).toMatch( + /WHERE .*\(MetricName = 'test\.metric'\)\) AND bitAnd\(Flags, 1\) = 0/, + ); + expect(sql.split(NO_RECORDED_VALUE_FILTER)).toHaveLength(2); + }, + ); + + // Gauge and sum queries already read Flags, so only histogram tables can + // lack it. + it.each( + cases.filter( + ([metricType]) => + metricType === MetricsDataType.Histogram || + metricType === MetricsDataType.ExponentialHistogram, + ), + )( + 'reads a table without a Flags column as before (%s, %o)', + async (metricType, select) => { + const sql = await renderMetricSql(metricType, select); + expect(sql).not.toContain(NO_RECORDED_VALUE_FILTER); + }, + ); + + it('reads the table as before when the Flags lookup fails', async () => { + mockMetadata.getColumn = jest + .fn() + .mockImplementation(async ({ column }) => { + if (column === 'Flags') throw new Error('DESCRIBE failed'); + return undefined; + }); + const sql = await renderMetricSql(MetricsDataType.Gauge, { + aggFn: 'avg', + }); + expect(sql).not.toContain(NO_RECORDED_VALUE_FILTER); + }); + + it.each(['UInt8', 'Int64', 'LowCardinality(UInt32)'])( + 'leaves flagged points out when Flags is %s', + async flagsType => { + mockMetadata.getColumn = jest + .fn() + .mockImplementation(async ({ column }) => + column === 'Flags' ? { name: 'Flags', type: flagsType } : undefined, + ); + const sql = await renderMetricSql(MetricsDataType.Gauge, { + aggFn: 'avg', + }); + expect(sql).toContain(NO_RECORDED_VALUE_FILTER); + }, + ); + + it.each(['String', 'Nullable(UInt32)', 'LowCardinality(Nullable(UInt32))'])( + 'reads the table as before when Flags is %s', + async flagsType => { + mockMetadata.getColumn = jest + .fn() + .mockImplementation(async ({ column }) => + column === 'Flags' ? { name: 'Flags', type: flagsType } : undefined, + ); + const sql = await renderMetricSql(MetricsDataType.Gauge, { + aggFn: 'avg', + }); + expect(sql).not.toContain(NO_RECORDED_VALUE_FILTER); + }, + ); + }); + describe('histogram metric queries', () => { describe('quantile', () => { it('should generate a whole-range query without a time dimension', async () => { diff --git a/packages/common-utils/src/core/renderChartConfig.ts b/packages/common-utils/src/core/renderChartConfig.ts index 44181ced7b..a38d8e6567 100644 --- a/packages/common-utils/src/core/renderChartConfig.ts +++ b/packages/common-utils/src/core/renderChartConfig.ts @@ -2,7 +2,13 @@ import isPlainObject from 'lodash/isPlainObject'; import * as SQLParser from 'node-sql-parser'; import SqlString from 'sqlstring'; -import { ChSql, chSql, concatChSql, wrapChSqlIfNotEmpty } from '@/clickhouse'; +import { + ChSql, + chSql, + concatChSql, + isCHIntegerType, + wrapChSqlIfNotEmpty, +} from '@/clickhouse'; import { stripTypeWrappers } from '@/core/eventDeltas'; import { compileFormulaAst, FormulaAst, validateFormula } from '@/core/formula'; import { @@ -2011,6 +2017,44 @@ async function getComputedMetricColumns( }; } +/** + * WHERE for a metric source CTE. OTLP sets bit 0 of Flags + * (FLAG_NO_RECORDED_VALUE) on a point that has no value, such as a Prometheus + * staleness marker. The ClickHouse exporter stores that point as zeros, so it + * is left out before any aggregation or delta reads it. A series that comes + * back after a marker is then compared with its last real point, so a counter + * that restarted behind the marker is clamped like any other counter reset + * (see the Sum Source CTE). The gauge and sum CTEs already read Flags, but + * HyperDX does not require it on histogram tables, and none of them need it + * to be an integer. So a table without an integer Flags column (or a failed + * column lookup) is read as before. Nullable is excluded too: + * `bitAnd(NULL, 1) = 0` is NULL and would drop the row. + */ +async function renderMetricWhere( + chartConfig: BuilderChartConfigWithOptDateRangeEx, + metadata: Metadata, +): Promise { + const where = await renderWhere(chartConfig, metadata); + let hasFlags = false; + try { + const flags = await metadata.getColumn({ + databaseName: chartConfig.from.databaseName, + tableName: chartConfig.from.tableName, + column: 'Flags', + connectionId: chartConfig.connection, + }); + hasFlags = + flags != null && + isCHIntegerType(flags.type) && + !flags.type.includes('Nullable('); + } catch (e) { + console.warn('Failed to look up the Flags column of a metric table', e); + } + return hasFlags + ? concatChSql(' AND ', where, chSql`bitAnd(Flags, 1) = 0`) + : where; +} + async function translateMetricChartConfig( chartConfig: BuilderChartConfigWithOptDateRangeEx, metadata: Metadata, @@ -2070,7 +2114,7 @@ async function translateMetricChartConfig( minGranularitySeconds: chartConfig.minGranularitySeconds, }); - const where = await renderWhere( + const where = await renderMetricWhere( { ...chartConfig, from: { @@ -2176,7 +2220,7 @@ async function translateMetricChartConfig( // Render the where clause to limit data selection on the source CTE but also search forward/back one // bucket window to ensure that there is enough data to compute a reasonable value on the ends of the // series. - const where = await renderWhere( + const where = await renderMetricWhere( { ...chartConfig, from: { @@ -2472,7 +2516,7 @@ async function translateMetricChartConfig( minGranularitySeconds: cteChartConfig.minGranularitySeconds, }) : undefined; - const where = await renderWhere(cteChartConfig, metadata); + const where = await renderMetricWhere(cteChartConfig, metadata); // Time bucket grouping is being handled separately, so make sure to ignore the granularity // logic for histograms specifically.