feat(phases): add the manual phase's group (HBM / ACME Container) to the grouped consumption

tests/unit/handlers/metrics.handlers.test.js

@@ -6072,6 +6072,109 @@ test('getConsumption - phase rejects groupBy, byMeter and racks, and needs a usa
   await t.exception(() => getConsumption(unconfigured, { query }), /ERR_PHASE_CONSUMPTION_REQUIRES_CENTRAL_DCS/)
 })
 
+const PHASE_DAY_MS = 24 * PHASE_HOUR_MS
+
+// Two stat-1d rows of miner telemetry for the grouped consumption tests.
+function groupedConsumptionRows (aggrField, groups) {
+  return [
+    { ts: PHASE_START, [aggrField]: groups },
+    { ts: PHASE_START + PHASE_DAY_MS, [aggrField]: groups }
+  ]
+}
+
+test('getConsumption - groupBy miner injects the manual phase group from its logged hours', async (t) => {
+  const reads = []
+  const ctx = phaseConsumptionCtx({
+    hashrate: groupedConsumptionRows('power_w_type_group_sum_aggr', { 'miner-wm-m63': 4000000 }),
+    // Day 1 has two logged hours; day 2 has none
+    manual: [
+      { ts: PHASE_START + 3 * PHASE_HOUR_MS, totalConsumptionMWh: 0.9 },
+      { ts: PHASE_START + 4 * PHASE_HOUR_MS, totalConsumptionMWh: 0.7 }
+    ],
+    onCall: (method, payload) => { if (method === 'getHistoricalLogs') reads.push(payload) }
+  })
+
+  const res = await getConsumption(ctx, {
+    query: { start: PHASE_START + 6 * PHASE_HOUR_MS, end: PHASE_START + 2 * PHASE_DAY_MS - 1, groupBy: 'miner' }
+  })
+
+  t.is(res.log[0].powerW['miner-wm-m63'], 4000000, 'telemetry groups untouched')
+  t.is(res.log[0].powerW.HBM, 800000, 'mean of the logged hours, as average W')
+  t.ok(Math.abs(res.log[0].consumptionMWh.HBM - 1.6) < 1e-9, 'the MWh actually logged, not 0.8 MW over 24 h')
+  t.is(res.log[0].consumptionMWh['miner-wm-m63'], (4000000 * 24) / 1000000, 'telemetry MWh untouched')
+  t.absent(res.log[1].powerW.HBM, 'a day without logged hours gets no key, not a zero')
+  t.is(res.summary.groupedBy.HBM.avgPowerW, 800000, 'the summary lists the phase group')
+  t.ok(Math.abs(res.summary.groupedBy.HBM.totalConsumptionMWh - 1.6) < 1e-9, 'its total is the logged MWh')
+  t.is(res.summary.groupedBy['miner-wm-m63'].totalConsumptionMWh, (2 * 4000000 * 24) / 1000000)
+  t.ok(Math.abs(res.summary.totalConsumptionMWh - (192 + 1.6)) < 1e-9, 'the site total adds the logged MWh')
+  t.is(reads.length, 1, 'one manual read for the whole range')
+  t.is(reads[0].start, PHASE_START, 'the first row is read from its UTC midnight, before start')
+  t.is(reads[0].end, PHASE_START + 2 * PHASE_DAY_MS - 1)
+})
+
+test('getConsumption - groupBy container injects the container group, phase=total too', async (t) => {
+  const ctx = phaseConsumptionCtx({
+    hashrate: groupedConsumptionRows('power_w_container_group_sum_aggr', { 'group-1': 500000 }),
+    manual: [{ ts: PHASE_START, totalConsumptionMWh: 1.2 }]
+  })
+  const query = { start: PHASE_START, end: PHASE_START + 2 * PHASE_DAY_MS - 1, groupBy: 'container' }
+
+  const res = await getConsumption(ctx, { query })
+  t.is(res.log[0].powerW['acme-container'], 1200000)
+  t.absent(res.log[0].powerW.HBM, 'only the key for this groupBy')
+
+  const total = await getConsumption(ctx, { query: { ...query, phase: 'total' } })
+  t.is(total.log[0].powerW['acme-container'], 1200000, 'Total tab lists the phase group too')
+})
+
+test('getConsumption - groupBy never overwrites a telemetry key with the phase group', async (t) => {
+  const ctx = phaseConsumptionCtx({
+    hashrate: groupedConsumptionRows('power_w_type_group_sum_aggr', { HBM: 3000000 }),
+    manual: [{ ts: PHASE_START, totalConsumptionMWh: 1 }]
+  })
+
+  const res = await getConsumption(ctx, {
+    query: { start: PHASE_START, end: PHASE_START + 2 * PHASE_DAY_MS - 1, groupBy: 'miner' }
+  })
+
+  t.is(res.log[0].powerW.HBM, 3000000)
+})
+
+test('getConsumption - groupBy is untouched without central DCS, phases, a manual phase or for racks', async (t) => {
+  const rows = groupedConsumptionRows('power_w_type_group_sum_aggr', { 'miner-wm-m63': 4000000 })
+  const query = { start: PHASE_START, end: PHASE_START + 2 * PHASE_DAY_MS - 1, groupBy: 'miner' }
+  let manualReads = 0
+  const jRequest = async (key, method) => {
+    if (method === 'getHistoricalLogs') {
+      manualReads++
+      return [{ ts: PHASE_START, totalConsumptionMWh: 1 }]
+    }
+    return method === 'tailLog' ? rows : []
+  }
+  const keysOf = (res) => Object.keys(res.log[0].powerW)
+
+  const noDcs = await getConsumption(phaseCtx(jRequest), { query })
+  t.alike(keysOf(noDcs), ['miner-wm-m63'], 'the manual store lives on the DCS worker')
+
+  const noPhases = await getConsumption(phaseCtx(jRequest, { centralDCSSetup: { enabled: true } }), { query })
+  t.alike(keysOf(noPhases), ['miner-wm-m63'])
+
+  const remainderOnly = {
+    ...PHASE_DCS_FEATURE,
+    phases: PHASE_DCS_FEATURE.phases.filter(p => p.id !== 'phase1_5')
+  }
+  const noManual = await getConsumption(phaseCtx(jRequest, remainderOnly), { query })
+  t.alike(keysOf(noManual), ['miner-wm-m63'], 'a remainder phase is inside the telemetry already')
+
+  const rackRows = groupedConsumptionRows('power_w_pdu_rack_group_sum_aggr', { r1: 1000000 })
+  const racks = await getConsumption(
+    phaseCtx(async (key, method) => (method === 'tailLog' ? rackRows : []), PHASE_DCS_FEATURE),
+    { query: { ...query, groupBy: 'rack' } }
+  )
+  t.alike(keysOf(racks), ['r1'])
+  t.is(manualReads, 0, 'the manual store is not read when nothing is injected')
+})
+
 test('getConsumption/getDowntime - total stays the site view, downtime still rejects a phase', async (t) => {
   const ctx = phaseCtx(async () => [])
   const query = { start: PHASE_START, end: PHASE_START + PHASE_HOUR_MS, phase: 'phase1_5' }

workers/lib/server/handlers/metrics.handlers.js

@@ -1020,6 +1020,63 @@ const CONSUMPTION_GROUP_FIELDS = {
   rack: { field: LOG_FIELDS.POWER_W_RACK_GROUP_SUM, aggrField: AGGR_FIELDS.POWER_W_RACK_GROUP_SUM }
 }
 
+/**
+ * Adds the configured synthetic group (e.g. the container's miner type) of every
+ * manually measured phase to the grouped consumption, so the Miner Type / Mining
+ * Unit views list the phase next to the telemetry-backed groups, as the grouped
+ * hashrate does. Its power is the mean of the hours logged in each day; a day
+ * without any logged hour gets no key rather than a zero. The grouped series is
+ * miner telemetry only, so the phase's load is not counted twice.
+ */
+async function injectSyntheticConsumptionGroups (ctx, req, log) {
+  if (!log.length) return
+  // `total` must list the phase groups exactly like a phase-less request.
+  if (lPhases.resolvePhase(ctx, req)) return
+
+  const groupBy = req.query.groupBy
+  if (groupBy !== 'miner' && groupBy !== 'container') return
+  if (!isCentralDCSEnabled(ctx)) return
+
+  const keys = (lPhases.getPhases(ctx) || [])
+    .filter((p) => p?.consumption?.source === lPhases.CONSUMPTION_SOURCES.MANUAL)
+    .map((p) => (groupBy === 'miner' ? p.groups?.minerType : p.groups?.container))
+    .filter(Boolean)
+  if (!keys.length) return
+
+  const dayMs = METRICS_TIME.ONE_DAY_MS
+  const firstDay = Math.min(...log.map((entry) => entry.ts))
+  const lastDay = Math.max(...log.map((entry) => entry.ts))
+  // Daily rows sit on UTC midnights that can precede `start`: read their whole days.
+  const manualByHour = await fetchManualConsumptionByHour(ctx, firstDay, lastDay + dayMs - 1)
+
+  for (const entry of log) {
+    let totalMWh = 0
+    let hours = 0
+    for (const [ts, mwh] of manualByHour) {
+      if (ts < entry.ts || ts >= entry.ts + dayMs) continue
+      totalMWh += mwh
+      hours++
+    }
+    if (!hours) continue
+
+    const powerW = (totalMWh / hours) * 1000000
+    // Copies: the maps can be shared with the worker response (or between rows)
+    entry.powerW = entry.powerW && typeof entry.powerW === 'object' ? { ...entry.powerW } : {}
+    entry.consumptionMWh = entry.consumptionMWh && typeof entry.consumptionMWh === 'object'
+      ? { ...entry.consumptionMWh }
+      : {}
+    for (const key of keys) {
+      // A synthetic key colliding with a real device-type key is a config error;
+      // losing telemetry to the overwrite would be worse than the group going missing.
+      if (entry.powerW[key] !== undefined) continue
+      entry.powerW[key] = powerW
+      // The energy actually logged, not the mean stretched over 24 h: a day with only
+      // some hours entered would otherwise report hours that were never logged.
+      entry.consumptionMWh[key] = totalMWh
+    }
+  }
+}
+
 async function getGroupedConsumption (ctx, req) {
   const { groupBy, start, end } = req.query
 
@@ -1054,6 +1111,8 @@ async function getGroupedConsumption (ctx, req) {
     return aggr
   }, [])
 
+  await injectSyntheticConsumptionGroups(ctx, req, log)
+
   const summary = calculateGroupedConsumptionSummary(log, groupBy)
 
   return { log, summary }
@@ -1069,6 +1128,9 @@ function calculateGroupedConsumptionSummary (log, groupBy) {
 
   const groupTotals = {}
   const groupCounts = {}
+  // MWh a row carries itself where it differs from its power over 24 h (a manual
+  // phase group with only some hours logged): the gap is added to the 24 h total.
+  const groupMWhAdjust = {}
 
   for (const entry of log) {
     const powerW = entry.powerW
@@ -1077,24 +1139,32 @@ function calculateGroupedConsumptionSummary (log, groupBy) {
         const v = Number(val) || 0
         groupTotals[name] = (groupTotals[name] || 0) + v
         groupCounts[name] = (groupCounts[name] || 0) + 1
+        const mwh = entry.consumptionMWh?.[name]
+        const fullDayMWh = (v * 24) / 1000000
+        if (Number.isFinite(mwh) && mwh !== fullDayMWh) {
+          groupMWhAdjust[name] = (groupMWhAdjust[name] || 0) + mwh - fullDayMWh
+        }
       }
     }
   }
 
   const byGroup = {}
   let siteTotal = 0
+  let siteMWhAdjust = 0
   for (const [name, total] of Object.entries(groupTotals)) {
     const avgPowerW = safeDiv(total, groupCounts[name])
+    const adjust = groupMWhAdjust[name] || 0
     byGroup[name] = {
       avgPowerW,
-      totalConsumptionMWh: (total * 24) / 1000000
+      totalConsumptionMWh: (total * 24) / 1000000 + adjust
     }
     siteTotal += total
+    siteMWhAdjust += adjust
   }
 
   return {
     avgPowerW: safeDiv(siteTotal, log.length),
-    totalConsumptionMWh: (siteTotal * 24) / 1000000,
+    totalConsumptionMWh: (siteTotal * 24) / 1000000 + siteMWhAdjust,
     groupedBy: byGroup
   }
 }