Add merge module metrics. #92 (#94)

Author: juliangruber

README.md

@@ -79,17 +79,31 @@ the configuration options described in
 - `$MAX_DISK_SPACE`_(number; optional)_: Maximum disk space (in bytes) to use in
   `$CACHE_ROOT` and `$STATE_ROOT` combined.
 
-### `$ station metrics`
+### `$ station metrics <module>`
 
-Get Station metrics.
+Get combined metrics from all Station Modules:
 
 ```bash
 $ station metrics
 {
 	"totalJobsCompleted": 123,
 	"totalEarnings": "0"
 }
+```
+
+Get metrics from a specific Station Module:
 
+```bash
+$ station metrics saturn-l2-node
+{
+	"totalJobsCompleted": 123,
+	"totalEarnings": "0"
+}
+```
+
+Follow metrics:
+
+```bash
 $ station metrics --follow
 {
 	"totalJobsCompleted": 123,
@@ -205,11 +219,11 @@ $ station --help
 Usage: station <command> [options]
 
 Commands:
-  station                Start Station                                 [default]
-  station metrics        Show metrics
-  station activity       Show activity log
-  station logs [module]  Show module logs
-  station events         Events stream
+  station                   Start Station                              [default]
+  station metrics [module]  Show metrics
+  station activity          Show activity log
+  station logs [module]     Show module logs
+  station events            Events stream
 
 Options:
   -v, --version  Show version number                                   [boolean]

bin/station.js

@@ -23,13 +23,15 @@ Sentry.init({
 await fs.mkdir(join(paths.moduleCache, 'saturn-L2-node'), { recursive: true })
 await fs.mkdir(join(paths.moduleState, 'saturn-L2-node'), { recursive: true })
 await fs.mkdir(paths.moduleLogs, { recursive: true })
+await fs.mkdir(paths.metrics, { recursive: true })
 await maybeCreateMetricsFile()
+await maybeCreateMetricsFile('saturn-L2-node')
 await maybeCreateActivityFile()
 
 yargs(hideBin(process.argv))
   .usage('Usage: $0 <command> [options]')
   .command('$0', 'Start Station', () => {}, commands.station)
-  .command('metrics', 'Show metrics', () => {}, commands.metrics)
+  .command('metrics [module]', 'Show metrics', () => {}, commands.metrics)
   .commands(
     'activity',
     'Show activity log',

commands/metrics.js

@@ -1,11 +1,11 @@
 import { followMetrics, getLatestMetrics } from '../lib/metrics.js'
 
-export const metrics = async ({ follow }) => {
+export const metrics = async ({ follow, module }) => {
   if (follow) {
-    for await (const obj of followMetrics()) {
+    for await (const obj of followMetrics(module)) {
       console.log(JSON.stringify(obj, 0, 2))
     }
   } else {
-    console.log(JSON.stringify(await getLatestMetrics(), 0, 2))
+    console.log(JSON.stringify(await getLatestMetrics(module), 0, 2))
   }
 }

commands/station.js

@@ -32,7 +32,7 @@ export const station = async () => {
     MAX_DISK_SPACE,
     storagePath: join(paths.moduleCache, 'saturn-L2-node'),
     binariesPath: paths.moduleBinaries,
-    metricsStream: createMetricsStream(paths.metrics),
+    metricsStream: await createMetricsStream('saturn-L2-node'),
     activityStream: createActivityStream('Saturn'),
     logStream: createLogStream(join(paths.moduleLogs, 'saturn-L2-node.log'))
   })

lib/metrics.js

@@ -1,5 +1,5 @@
 import { createLogStream, SerializeStream, parseLog, formatLog } from './log.js'
-import { Transform } from 'node:stream'
+import { Transform, PassThrough } from 'node:stream'
 import { writeClient } from './telemetry.js'
 import { Point } from '@influxdata/influxdb-client'
 import fs from 'node:fs/promises'
@@ -8,38 +8,59 @@ import { paths } from './paths.js'
 import { Tail } from 'tail'
 import { maybeCreateFile } from './util.js'
 import { platform } from 'node:os'
+import assert from 'node:assert'
+import { join } from 'node:path'
 
 // Metrics stream
 // - Filters duplicate entries
 // - Writes `jobs-completed` to InfluxDB
 // - Serializes JSON
-export const createMetricsStream = metricsPath => {
+// - Persists to
+//   - module specific metrics file
+//   - merged metrics file
+export const createMetricsStream = async moduleName => {
+  assert(moduleName, 'moduleName required')
   const deduplicateStream = new DeduplicateStream()
+  const splitStream = new PassThrough({ objectMode: true })
   deduplicateStream
-    .pipe(new TelemetryStream())
+    .pipe(new TelemetryStream(moduleName))
+    .pipe(splitStream)
+  splitStream
     .pipe(new SerializeStream())
-    .pipe(createLogStream(metricsPath))
+    .pipe(createLogStream(join(paths.metrics, `${moduleName}.log`)))
+  splitStream
+    .pipe(new MergeMetricsStream(...await Promise.all([
+      getLatestMetrics(),
+      getLatestMetrics(moduleName)
+    ])))
+    .pipe(new SerializeStream())
+    .pipe(createLogStream(paths.allMetrics))
   return deduplicateStream
 }
 
-export const maybeCreateMetricsFile = async () => {
+export const maybeCreateMetricsFile = async (moduleName) => {
   await maybeCreateFile(
-    paths.metrics,
+    getMetricsFilePath(moduleName),
     formatLog(
       JSON.stringify({ totalJobsCompleted: 0, totalEarnings: '0' }) + '\n'
     )
   )
 }
 
 const metricsLogLineToObject = metrics => JSON.parse(parseLog(metrics).text)
+const getMetricsFilePath = moduleName => {
+  return moduleName
+    ? join(paths.metrics, `${moduleName}.log`)
+    : paths.allMetrics
+}
 
-export const getLatestMetrics = async () => {
-  const metrics = await fs.readFile(paths.metrics, 'utf-8')
+export const getLatestMetrics = async (moduleName) => {
+  const metrics = await fs.readFile(getMetricsFilePath(moduleName), 'utf-8')
   return metricsLogLineToObject(metrics.trim().split('\n').pop())
 }
 
-export const followMetrics = async function * ({ signal } = {}) {
-  const tail = new Tail(paths.metrics, {
+export const followMetrics = async function * ({ moduleName, signal } = {}) {
+  const tail = new Tail(getMetricsFilePath(moduleName), {
     nLines: 1,
     useWatchFile: platform() === 'win32'
   })
@@ -69,16 +90,18 @@ class DeduplicateStream extends Transform {
 }
 
 class TelemetryStream extends Transform {
-  constructor () {
+  constructor (moduleName) {
+    assert(moduleName, 'moduleName required')
     super({ objectMode: true })
     this.lastMetrics = null
+    this.moduleName = moduleName
   }
 
   _transform (metrics, _, callback) {
     if (typeof this.lastMetrics?.totalJobsCompleted === 'number') {
       writeClient.writePoint(
         new Point('jobs-completed')
-          .stringField('module', 'saturn')
+          .stringField('module', this.moduleName)
           .intField(
             'value',
             metrics.totalJobsCompleted - this.lastMetrics.totalJobsCompleted
@@ -89,3 +112,28 @@ class TelemetryStream extends Transform {
     callback(null, metrics)
   }
 }
+
+class MergeMetricsStream extends Transform {
+  constructor (allMetrics, moduleMetrics) {
+    assert.strictEqual(
+      typeof allMetrics?.totalJobsCompleted,
+      'number',
+      'allMetrics.totalJobsCompleted required'
+    )
+    assert.strictEqual(
+      typeof moduleMetrics?.totalJobsCompleted,
+      'number',
+      'moduleMetrics.totalJobsCompleted required'
+    )
+    super({ objectMode: true })
+    this.allMetrics = allMetrics
+    this.moduleMetrics = moduleMetrics
+  }
+
+  _transform (metrics, _, callback) {
+    const diff = metrics.totalJobsCompleted - this.moduleMetrics.totalJobsCompleted
+    this.allMetrics.totalJobsCompleted += diff
+    this.moduleMetrics.totalJobsCompleted = metrics.totalJobsCompleted
+    callback(null, this.allMetrics)
+  }
+}

lib/paths.js

@@ -16,7 +16,8 @@ const repoRoot = join(dirname(fileURLToPath(import.meta.url)), '..')
 
 export const getPaths = (cacheRoot, stateRoot) => ({
   repoRoot,
-  metrics: join(stateRoot, 'logs', 'metrics.log'),
+  metrics: join(stateRoot, 'logs', 'metrics'),
+  allMetrics: join(stateRoot, 'logs', 'metrics', 'all.log'),
   activity: join(stateRoot, 'logs', 'activity.log'),
   moduleCache: join(cacheRoot, 'modules'),
   moduleState: join(stateRoot, 'modules'),

test/test.js

@@ -107,11 +107,11 @@ describe('Metrics', () => {
     const CACHE_ROOT = join(tmpdir(), randomUUID())
     const STATE_ROOT = join(tmpdir(), randomUUID())
     await fs.mkdir(
-      dirname(getPaths(CACHE_ROOT, STATE_ROOT).metrics),
+      dirname(getPaths(CACHE_ROOT, STATE_ROOT).allMetrics),
       { recursive: true }
     )
     await fs.writeFile(
-      getPaths(CACHE_ROOT, STATE_ROOT).metrics,
+      getPaths(CACHE_ROOT, STATE_ROOT).allMetrics,
       '[date] {"totalJobsCompleted":1,"totalEarnings":"2"}\n'
     )
     const { stdout } = await execa(
@@ -124,6 +124,27 @@ describe('Metrics', () => {
       JSON.stringify({ totalJobsCompleted: 1, totalEarnings: '2' }, 0, 2)
     )
   })
+  it('outputs module metrics', async () => {
+    const CACHE_ROOT = join(tmpdir(), randomUUID())
+    const STATE_ROOT = join(tmpdir(), randomUUID())
+    await fs.mkdir(
+      getPaths(CACHE_ROOT, STATE_ROOT).metrics,
+      { recursive: true }
+    )
+    await fs.writeFile(
+      join(getPaths(CACHE_ROOT, STATE_ROOT).metrics, 'saturn-l2-node.log'),
+      '[date] {"totalJobsCompleted":1,"totalEarnings":"2"}\n'
+    )
+    const { stdout } = await execa(
+      station,
+      ['metrics', 'saturn-l2-node'],
+      { env: { CACHE_ROOT, STATE_ROOT } }
+    )
+    assert.deepStrictEqual(
+      stdout,
+      JSON.stringify({ totalJobsCompleted: 1, totalEarnings: '2' }, 0, 2)
+    )
+  })
 
   describe('Follow', async () => {
     for (const flag of ['-f', '--follow']) {