feat: add native duration-weighted distribution statistics

queries/cdmq/README.md

@@ -12,6 +12,10 @@ npm install
 
 The contents of this directory contain a collection of scripts in Javascript intended to be executed with [node.js](https://nodejs.org). These scripts get data from an OpenSearch instance. The data must be in Common Data Format, whose index mapping definitions are documented in [cdm.js](./cdm.js)'s `indexDefs` object. The scripts here are meant to help inspect, compare, and export data from benchmarks and performance & resource-utilization tools, in order to report and investigate performance.
 
+Native distribution statistics use OpenSearch Point-in-Time search with `search_after` for stable deep pagination. The Crucible controller currently ships OpenSearch 3.6.0, which is the tested compatibility target for this feature. Deployments without PIT support receive an explicit `NATIVE_STATS_PIT_UNSUPPORTED` error.
+
+Resource limits are configurable on the CDM server with positive-integer environment variables. Defaults are `CDM_NATIVE_STATS_MAX_DOCUMENTS=250000`, `CDM_NATIVE_STATS_MAX_INTERVALS=500000`, `CDM_NATIVE_STATS_MAX_RUNTIME_MS=300000`, and `CDM_NATIVE_STATS_PAGE_SIZE=1000`. Exceeding a limit returns an error; the server never silently falls back to statistics over resolution buckets.
+
 In order to generate this data, you must run a benchmark via automation framework which uses the Common Data Format and index that data into OpenSearch. One of those automation frameworks is the [crucible](https://github.com/perftool-incubator/crucible) project. A subproject of crucible, [crucible-examples](https://github.com/perftool-incubator/crucible-examples), includes scenarios to run some of these benchmarks.
 
 ## Terms
@@ -487,6 +491,15 @@ node ./get-metric-data.js --period <UUID> --source iostat --type kB-sec --breako
 
 So far all of the metrics have been represented as a single value for a specific time period. When `--period` is used, the script finds the begin and end times for this period, which in most cases, has a duration equal to the measurement time in the benchmark itself (around 90 seconds in these examples). One can also specify `--run`, `--begin`, and `--end` instead of `--period`, should they need to focus on a different period of time. However, for benchmark metrics (such as uperf), it is important to limit the begin and end to within the actual measurement period for that sample. Conversely, tool metrics can use a begin and end spanning any time period within the run, as the tool collection tends to run continuously for any particular run. Whatever time period is used, one can also use `--resolution` to divide this time period into multiple data-samples, in order to generate things like line graphs:
 
+The optional `--distribution-stats` option returns duration-weighted statistics over the native reconstructed timeline, independently of `--resolution`. For example:
+
+```bash
+node ./get-metric-data.js --period <UUID> --source uperf --type Gbps \
+  --distribution-stats min,max,mean,median,stddev,p95
+```
+
+This describes variation in the underlying metric over time rather than variation in caller-selected display buckets. The JSON response includes these statistics in `distributionStats`, keyed by breakout label, while the existing `values` response remains unchanged. Native statistics are opt-in because they stream all matching metric documents and are subject to resource limits. Statistics use population standard deviation, duration-weighted nearest-rank percentiles, and `median` is equivalent to `p50`; a single native interval has `stddev=0`.
+
     # node ./get-metric-data.js --period 4F1014D6-AD33-11EC-94E3-ADE96E3275F7 --source sar-net --type L2-Gbps --breakout csid=1,cstype=worker,type=physical,direction=tx,dev --filter gt:0.01 --resolution 10
     Checking for httpd...appears to be running
     Checking for OpenSearch...appears to be running

queries/cdmq/cdm.js

@@ -1,7 +1,20 @@
 //# vim: autoindent tabstop=2 shiftwidth=2 expandtab softtabstop=2 filetype=javascript
 var request = require('sync-request');
 var thenRequest = require('then-request');
+const { calculateDistributionStats, reconstructTimeline, resampleTimeline, validateRequestedStats } = require('./native-stats');
 var bigQuerySize = 262144;
+
+function nativeStatsLimit(name, fallback) {
+  const value = process.env[name];
+  if (typeof value === 'undefined') return fallback;
+  const limit = Number(value);
+  if (!Number.isSafeInteger(limit) || limit <= 0) {
+    const error = new Error(name + ' must be a positive integer');
+    error.code = 'NATIVE_STATS_CONFIG';
+    throw error;
+  }
+  return limit;
+}
 const docTypes = {
   v7dev: ['run', 'tag', 'iteration', 'param', 'sample', 'period', 'metric_desc', 'metric_data'],
   v8dev: ['run', 'tag', 'iteration', 'param', 'sample', 'period', 'metric_desc', 'metric_data'],
@@ -3699,6 +3712,44 @@ getMetricDataFromIdsSets = async function (instance, sets, metricGroupIdsByLabel
     // Each template has prefix/suffix pairs for the 4 query types,
     // with __IDS__ as placeholder for the metric UUID list.
     var defaultAggregation = sets[idx].defaultAggregation || 'sum';
+
+    // When native statistics are requested, use the reconstructed timeline for
+    // both the output series and the statistics. This avoids running the legacy
+    // resolution query and then fetching the same documents again for stats.
+    if (sets[idx].distributionStats) {
+      valueSets[idx] = { distributionStats: {} };
+      const nativeIndexName = getIndexName('metric_data', instance, yearDotMonth);
+      const sortedNativeLabels = Object.keys(metricGroupIdsByLabelSets[idx]).sort();
+      const nativeStatsConcurrency = 4;
+      for (let nativeStart = 0; nativeStart < sortedNativeLabels.length; nativeStart += nativeStatsConcurrency) {
+        const nativeLabels = sortedNativeLabels.slice(nativeStart, nativeStart + nativeStatsConcurrency);
+        const nativeResults = await Promise.all(
+          nativeLabels.map(async (label) => {
+            const native = await getNativeMetricStats(
+              instance,
+              metricGroupIdsByLabelSets[idx][label],
+              begin,
+              end,
+              defaultAggregation,
+              sets[idx].distributionStats,
+              yearDotMonth,
+              { includeTimeline: true, indexName: nativeIndexName }
+            );
+            return {
+              label: label,
+              values: resampleTimeline(native.timeline, begin, end, resolution, defaultAggregation),
+              stats: native.stats
+            };
+          })
+        );
+        nativeResults.forEach((result) => {
+          valueSets[idx][result.label] = result.values;
+          valueSets[idx].distributionStats[result.label] = result.stats;
+        });
+      }
+      continue;
+    }
+
     var timeRangeTemplates = [];
     var thisBegin = begin;
     var thisEnd = begin + duration;
@@ -3861,6 +3912,157 @@ getMetricDataFromIdsSets = async function (instance, sets, metricGroupIdsByLabel
 
 exports.getMetricDataFromIdsSets = getMetricDataFromIdsSets;
 
+// Stream the documents needed to reconstruct one native aggregate timeline.
+// This is deliberately separate from getMetricDataFromIdsSets(): resolution
+// bucketing can use aggregations, while native statistics must see every
+// boundary where any selected metric ID changes value.
+async function getNativeMetricStats(
+  instance,
+  metricIds,
+  begin,
+  end,
+  aggregation,
+  requestedStats,
+  yearDotMonth,
+  options = {}
+) {
+  validateRequestedStats(requestedStats);
+  const maxDocuments = options.maxDocuments || nativeStatsLimit('CDM_NATIVE_STATS_MAX_DOCUMENTS', 250000);
+  const maxIntervals = options.maxIntervals || nativeStatsLimit('CDM_NATIVE_STATS_MAX_INTERVALS', 500000);
+  const maxRuntimeMs = options.maxRuntimeMs || nativeStatsLimit('CDM_NATIVE_STATS_MAX_RUNTIME_MS', 300000);
+  const pageSize = options.pageSize || nativeStatsLimit('CDM_NATIVE_STATS_PAGE_SIZE', 1000);
+  const indexName = options.indexName || getIndexName('metric_data', instance, yearDotMonth);
+  const baseUrl = 'http://' + instance.host;
+  const headers = instance.header || { 'Content-Type': 'application/json' };
+  const fetchImpl = options.fetch || fetch;
+  let pitId;
+  const deadline = Date.now() + maxRuntimeMs;
+
+  async function send(method, url, body) {
+    const response = await fetchImpl(url, {
+      method: method,
+      headers: headers,
+      body: body === undefined ? undefined : JSON.stringify(body)
+    });
+    if (!response.ok) {
+      throw new Error('OpenSearch request failed with HTTP status ' + response.status);
+    }
+    return response.json();
+  }
+
+  try {
+    let pit;
+    try {
+      pit = await send('POST', baseUrl + '/' + indexName + '/_search/point_in_time?keep_alive=1m', {});
+    } catch (error) {
+      error.code = 'NATIVE_STATS_PIT_UNSUPPORTED';
+      error.message = 'native distribution statistics require OpenSearch Point-in-Time search support: ' + error.message;
+      throw error;
+    }
+    pitId = pit.pit_id || pit.id;
+    if (!pitId) {
+      const error = new Error('OpenSearch did not return a point-in-time ID');
+      error.code = 'NATIVE_STATS_PIT_UNSUPPORTED';
+      throw error;
+    }
+
+    const documentsById = {};
+    let searchAfter;
+    let documentCount = 0;
+    let firstPage = true;
+    while (true) {
+      if (Date.now() > deadline) {
+        const error = new Error('native distribution statistics processing time limit exceeded');
+        error.code = 'NATIVE_STATS_LIMIT';
+        throw error;
+      }
+      const query = {
+        size: pageSize,
+        track_total_hits: firstPage ? maxDocuments + 1 : false,
+        pit: { id: pitId, keep_alive: '1m' },
+        sort: [
+          { 'metric_data.begin': 'asc' },
+          { 'metric_data.end': 'asc' },
+          { 'metric_desc.metric_desc-uuid': 'asc' }
+        ],
+        docvalue_fields: [
+          { field: 'metric_desc.metric_desc-uuid' },
+          { field: 'metric_data.begin', format: 'epoch_millis' },
+          { field: 'metric_data.end', format: 'epoch_millis' },
+          { field: 'metric_data.value' }
+        ],
+        _source: false,
+        query: {
+          bool: {
+            filter: [
+              { range: { 'metric_data.end': { gte: begin } } },
+              { range: { 'metric_data.begin': { lte: end } } },
+              { terms: { 'metric_desc.metric_desc-uuid': metricIds } }
+            ]
+          }
+        }
+      };
+      if (searchAfter) query.search_after = searchAfter;
+
+      const response = await send('POST', baseUrl + '/_search', query);
+      const total = response.hits && response.hits.total;
+      if (firstPage && total && total.value > maxDocuments) {
+        const error = new Error('native distribution statistics document limit exceeded');
+        error.code = 'NATIVE_STATS_LIMIT';
+        throw error;
+      }
+
+      const hits = (response.hits && response.hits.hits) || [];
+      documentCount += hits.length;
+      if (documentCount > maxDocuments) {
+        const error = new Error('native distribution statistics document limit exceeded');
+        error.code = 'NATIVE_STATS_LIMIT';
+        throw error;
+      }
+      hits.forEach((hit) => {
+        const fields = hit.fields || {};
+        const valueOf = (name) => {
+          const values = fields[name];
+          return Array.isArray(values) ? values[0] : values;
+        };
+        const metricId = valueOf('metric_desc.metric_desc-uuid');
+        if (!documentsById[metricId]) documentsById[metricId] = [];
+        documentsById[metricId].push({
+          begin: Number(valueOf('metric_data.begin')),
+          end: Number(valueOf('metric_data.end')),
+          value: Number(valueOf('metric_data.value'))
+        });
+      });
+
+      if (hits.length < pageSize) break;
+      searchAfter = hits[hits.length - 1].sort;
+      if (!searchAfter) throw new Error('OpenSearch response did not include sort values for search_after');
+      firstPage = false;
+    }
+
+    metricIds.forEach((metricId) => {
+      if (!documentsById[metricId]) {
+        const error = new Error('native distribution statistics missing metric ID ' + metricId);
+        error.code = 'NATIVE_STATS_DATA_QUALITY';
+        throw error;
+      }
+    });
+    const timeline = reconstructTimeline(documentsById, Number(begin), Number(end), aggregation, { maxIntervals });
+    const stats = calculateDistributionStats(timeline, requestedStats);
+    return options.includeTimeline ? { timeline: timeline, stats: stats } : stats;
+  } finally {
+    if (pitId) {
+      try {
+        await send('DELETE', baseUrl + '/_search/point_in_time', { pit_id: pitId });
+      } catch (error) {
+        console.error('Failed to close OpenSearch point-in-time: ' + error.message);
+      }
+    }
+  }
+}
+
+exports.getNativeMetricStats = getNativeMetricStats;
+
 // --------------------------------------------------------------------------------------------------------------
 // Generates 1 or more values for 1 or more groups for a metric of a particular source
 // (tool or benchmark) and type (iops, l2-Gbps, ints/sec, etc).
@@ -4111,13 +4313,15 @@ getMetricDataSets = async function (instance, sets, yearDotMonth) {
 
   for (var i = 0; i < sets.length; i++) {
     // Rearrange the actual data into 'values' section
-    Object.keys(dataSets[i]).forEach((label) => {
+    Object.keys(dataSets[i])
+      .filter((label) => label !== 'distributionStats')
+      .forEach((label) => {
       if (isUndefined(dataSets[i].values)) {
         dataSets[i].values = {};
       }
       dataSets[i].values[label] = dataSets[i][label];
       delete dataSets[i][label];
-    });
+      });
     // Build the label-decoder and the remaining breakouts
     dataSets[i].usedBreakouts = sets[i].breakout;
     dataSets[i].valueSeriesLabelDecoder = '';
@@ -4148,6 +4352,7 @@ getMetricDataSets = async function (instance, sets, yearDotMonth) {
           )
         ) {
           delete dataSets[i].values[metric];
+          if (dataSets[i].distributionStats) delete dataSets[i].distributionStats[metric];
         }
       });
     }

queries/cdmq/get-metric-data.js

@@ -163,6 +163,11 @@ async function main() {
       '[optional] Filter out (do not output) metrics which do not pass the conditional.  gt=greater-than, ge=greater-than-or-equal, lt=less-than, le=less-than-or-equal'
     )
     .option('--aggregation <sum|avg|max|min>', '[optional] Override the default aggregation method for this query')
+    .option(
+      '--distribution-stats <stat1,stat2,...>',
+      '[optional] Return duration-weighted native-timeline statistics (min,max,mean,median,stddev,pNN)',
+      (value) => value.split(',').map((stat) => stat.trim()).filter(Boolean)
+    )
     .option(
       '--allow-incompatible-aggregation',
       '[optional] Allow an aggregation explicitly disallowed by the metric definition'
@@ -212,6 +217,7 @@ async function main() {
     breakout: program.breakout, // Send as array to preserve complex breakout syntax
     filter: program.filter,
     aggregation: program.aggregation,
+    'distribution-stats': program.distributionStats,
     'allow-incompatible-aggregation': program.allowIncompatibleAggregation,
     instances: program.instances.length > 0 ? program.instances : undefined
   };
@@ -424,6 +430,19 @@ async function main() {
     }
     console.log(line);
   }
+
+  if (metric_data.distributionStats && program.outputContent != 'headers') {
+    console.log('\nDistribution statistics (native timeline):');
+    Object.keys(metric_data.distributionStats)
+      .sort((a, b) => a.localeCompare(b, undefined, { numeric: true, sensitivity: 'base' }))
+      .forEach((label) => {
+        const stats = metric_data.distributionStats[label];
+        const formatted = Object.keys(stats)
+          .map((stat) => stat + '=' + Number(stats[stat]).toFixed(program.decimalPlaces))
+          .join(' ');
+        console.log('  ' + (label || '<all>') + ': ' + formatted);
+      });
+  }
 }
 
 main();