ktg-plugin-marketplace/plugins/voyage/lib/exporters/otlp-format.mjs

// lib/exporters/otlp-format.mjs
// Pure transform: voyage JSONL stats records → OTLP/JSON v1.0 metrics payload.
//
// Per OpenTelemetry Protocol § metrics.proto + research/01 dim 4 (CRITICAL):
//   AggregationTemporality enum values are INTEGERS in JSON, NOT strings.
//   "CUMULATIVE" → 2  (not the string)
//   "DELTA"      → 1
//   timeUnixNano is uint64 over the wire — emit as decimal STRING in JSON to
//   avoid JS Number precision loss (per research/01 + risk-assessor CRITICAL 2).
//
// Output contract:
//   {
//     resourceMetrics: [{
//       resource: { attributes: [...] },
//       scopeMetrics: [{
//         scope: { name: 'voyage', version: '...' },
//         metrics: [{
//           name: 'voyage.<metric>',
//           description: '...',
//           unit: '1' | 'ms' | ...,
//           sum: { dataPoints: [{ ... aggregationTemporality: <int> ...}] }
//             | gauge: { dataPoints: [...] }
//         }]
//       }]
//     }]
//   }

// ---- Inline integer enum constants (CRITICAL: integers, NOT strings) -------

const AGG_TEMPORALITY_UNSPECIFIED = 0;
const AGG_TEMPORALITY_DELTA = 1;
const AGG_TEMPORALITY_CUMULATIVE = 2;

const DATA_POINT_FLAGS_NONE = 0;
const DATA_POINT_FLAGS_NO_RECORDED_VALUE_MASK = 1;

const VOYAGE_SCOPE_NAME = 'voyage';
const VOYAGE_SCOPE_VERSION = '4.1.0';

// ---- Helpers ---------------------------------------------------------------

/**
 * Convert ISO-8601 timestamp to OTLP timeUnixNano (uint64 as decimal STRING).
 * Avoids Number precision loss for nanosecond-scale values.
 */
function toUnixNanoString(iso) {
  const ms = Date.parse(iso);
  if (Number.isNaN(ms)) return '0';
  // ms × 1e6 = nanoseconds; use BigInt for precision
  return (BigInt(ms) * 1000000n).toString();
}

/**
 * Build OTLP attribute object: {key, value: {stringValue: "..."}} or numeric variants.
 */
function attribute(key, value) {
  if (typeof value === 'string') return { key, value: { stringValue: value } };
  if (typeof value === 'boolean') return { key, value: { boolValue: value } };
  if (typeof value === 'number' && Number.isInteger(value)) return { key, value: { intValue: String(value) } };
  if (typeof value === 'number') return { key, value: { doubleValue: value } };
  return { key, value: { stringValue: String(value) } };
}

/**
 * Partition record into numeric metrics and string/bool labels.
 * (Same convention as textfile-format.mjs.)
 */
function partitionRecord(record) {
  const labels = {};
  const metrics = {};
  for (const [k, v] of Object.entries(record)) {
    if (k === 'ts' || k === '_schema_id') continue;
    if (typeof v === 'number') metrics[k] = v;
    else if (typeof v === 'boolean') metrics[k] = v ? 1 : 0;
    else if (typeof v === 'string') labels[k] = v;
  }
  return { labels, metrics };
}

/**
 * Build OTLP DataPoint object for a numeric value.
 */
function dataPoint(value, ts, labels) {
  const tsNano = toUnixNanoString(ts);
  return {
    attributes: Object.entries(labels).map(([k, v]) => attribute(k, v)),
    startTimeUnixNano: tsNano,
    timeUnixNano: tsNano,
    asDouble: Number.isInteger(value) ? undefined : value,
    asInt: Number.isInteger(value) ? String(value) : undefined,
    flags: DATA_POINT_FLAGS_NONE,
  };
}

/**
 * Decide metric "kind": sum (for *_total/*_count/*_passed/*_failed) or gauge.
 * Sum metrics get aggregationTemporality + isMonotonic; gauges get neither.
 */
function isSumMetric(name) {
  return /_total$|_count$|_passed$|_failed$|_skipped$/.test(name);
}

/**
 * Transform JSONL records → OTLP/JSON metrics payload. Pure function.
 *
 * @param {Array<object>} records  Allowlist-redacted records (caller responsibility).
 * @param {{help?: object}} [opts]
 * @returns {object}                OTLP-shaped payload (POST body for /v1/metrics).
 */
export function transformToOtlpJson(records, opts = {}) {
  const helpMap = opts.help || {};

  if (!Array.isArray(records) || records.length === 0) {
    return {
      resourceMetrics: [{
        resource: { attributes: [attribute('service.name', VOYAGE_SCOPE_NAME)] },
        scopeMetrics: [{
          scope: { name: VOYAGE_SCOPE_NAME, version: VOYAGE_SCOPE_VERSION },
          metrics: [],
        }],
      }],
    };
  }

  // Group all data points by metric name (schema_id_field).
  const metricsMap = new Map();

  for (const record of records) {
    const schemaId = (record && typeof record._schema_id === 'string') ? record._schema_id : 'unknown';
    const ts = record.ts || new Date().toISOString();
    const { labels, metrics } = partitionRecord(record);
    const allLabels = { ...labels, _schema_id: schemaId };

    for (const [field, value] of Object.entries(metrics)) {
      const name = `${VOYAGE_SCOPE_NAME}.${schemaId}.${field}`;
      if (!metricsMap.has(name)) {
        metricsMap.set(name, {
          name,
          description: helpMap[name] || `voyage stats — ${schemaId}.${field}`,
          unit: /_ms$|_duration/.test(field) ? 'ms' : (/_seconds$/.test(field) ? 's' : '1'),
          dataPoints: [],
          isSum: isSumMetric(name),
        });
      }
      metricsMap.get(name).dataPoints.push(dataPoint(value, ts, allLabels));
    }
  }

  // Sort metrics for deterministic output
  const sortedNames = [...metricsMap.keys()].sort();
  const otlpMetrics = sortedNames.map(name => {
    const m = metricsMap.get(name);
    if (m.isSum) {
      return {
        name: m.name,
        description: m.description,
        unit: m.unit,
        sum: {
          dataPoints: m.dataPoints,
          aggregationTemporality: AGG_TEMPORALITY_CUMULATIVE, // INTEGER 2
          isMonotonic: true,
        },
      };
    }
    return {
      name: m.name,
      description: m.description,
      unit: m.unit,
      gauge: {
        dataPoints: m.dataPoints,
      },
    };
  });

  return {
    resourceMetrics: [{
      resource: {
        attributes: [
          attribute('service.name', VOYAGE_SCOPE_NAME),
          attribute('service.version', VOYAGE_SCOPE_VERSION),
        ],
      },
      scopeMetrics: [{
        scope: { name: VOYAGE_SCOPE_NAME, version: VOYAGE_SCOPE_VERSION },
        metrics: otlpMetrics,
      }],
    }],
  };
}

export {
  AGG_TEMPORALITY_UNSPECIFIED,
  AGG_TEMPORALITY_DELTA,
  AGG_TEMPORALITY_CUMULATIVE,
  DATA_POINT_FLAGS_NONE,
  DATA_POINT_FLAGS_NO_RECORDED_VALUE_MASK,
};