feat: Expose log count metric from Runner (#684)

Creates a new `winston` transport method to count logs by level, and
exposes a new prometheus metric to record this value.

Additionally, metrics recorded on the main thread were not captured.
This was not an issue as the majority of metrics were done within the
worker. But since we also log on main thread, this PR updates metrics
aggregation to expose main thread metrics.

runner/package.json

@@ -70,6 +70,7 @@
     "redis": "^4.6.7",
     "verror": "^1.10.1",
     "vm2": "^3.9.19",
-    "winston": "^3.13.0"
+    "winston": "^3.13.0",
+    "winston-transport": "^4.7.0"
   }
 }

runner/src/logger.ts

@@ -1,14 +1,26 @@
 import winston from 'winston';
 import { LoggingWinston } from '@google-cloud/logging-winston';
+import Transport from 'winston-transport';
+
+import { METRICS } from './metrics';
 
 const { format, transports } = winston;
 
+class LogCounter extends Transport {
+  log (info: { level: string }, callback: () => void): void {
+    METRICS.LOGS_COUNT.labels({ level: info.level }).inc();
+
+    callback();
+  }
+}
+
 const logger = winston.createLogger({
   level: 'info',
   format: format.combine(
     format.timestamp(),
     format.errors({ stack: true }),
   ),
+  transports: [new LogCounter()],
 });
 
 if (process.env.GCP_LOGGING_ENABLED) {

runner/src/metrics.ts

@@ -1,5 +1,5 @@
 import express from 'express';
-import { Gauge, Histogram, Counter, AggregatorRegistry } from 'prom-client';
+import { Gauge, Histogram, Counter, AggregatorRegistry, register } from 'prom-client';
 
 import logger from './logger';
 
@@ -56,6 +56,12 @@ const EXECUTION_DURATION = new Histogram({
   labelNames: ['indexer'],
 });
 
+const LOGS_COUNT = new Counter({
+  name: 'queryapi_runner_logs_count',
+  help: 'Number of messages logged',
+  labelNames: ['level'],
+});
+
 export const METRICS = {
   HEAP_TOTAL_ALLOCATION,
   HEAP_USED,
@@ -66,6 +72,7 @@ export const METRICS = {
   UNPROCESSED_STREAM_MESSAGES,
   LAST_PROCESSED_BLOCK_HEIGHT,
   EXECUTION_DURATION,
+  LOGS_COUNT
 };
 
 const aggregatorRegistry = new AggregatorRegistry();
@@ -87,7 +94,9 @@ export const startServer = async (): Promise<void> => {
   app.get('/metrics', async (_req, res) => {
     res.set('Content-Type', aggregatorRegistry.contentType);
 
-    const metrics = await AggregatorRegistry.aggregate(Array.from(workerMetrics.values())).metrics();
+    const mainThreadMetrics = await register.getMetricsAsJSON();
+    const metrics = await AggregatorRegistry.aggregate([...Array.from(workerMetrics.values()), mainThreadMetrics]).metrics();
+
     res.send(metrics);
   });