Stresstest and memtest the Hive Logger (#1591)

Author: enisdenjo

.github/workflows/memtest.yml

@@ -25,6 +25,7 @@ jobs:
           - federation-subscriptions-passthrough
           - opentelemetry
           - programmatic-batching
+          - logger
         e2e_runner:
           - node
           # - bun TODO: get memory snaps and heap sampling for bun. is it even necessary?
@@ -65,4 +66,6 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: ${{matrix.test_name}}-heap-snapshots
-          path: e2e/${{matrix.test_name}}/*.heapsnapshot
+          path: |
+            e2e/${{matrix.test_name}}/*.heapsnapshot
+            packages/logger/tests/*.heapsnapshot

.gitignore

@@ -16,10 +16,7 @@ hive-gateway
 /examples/**/*/supergraph.graphql
 .helix/config.toml
 .helix/languages.toml
-/e2e/**/*/memtest-memory-usage_*.svg
-/e2e/**/*/*.heapsnapshot
-/e2e/**/*/*.heapprofile
-/e2e/**/*/*.out
-/internal/**/*/*.heapsnapshot
-/internal/**/*/*.heapprofile
-/internal/**/*/*.out
+memtest-*.svg
+*.heapsnapshot
+*.heapprofile
+*.out

internal/perf/src/loadtest-script.ts renamed to internal/perf/src/graphql-loadtest-script.ts

@@ -19,14 +19,20 @@ export interface LoadtestOptions extends ProcOptions {
   calmdown: number;
   /** How many times to run the loadtests? */
   runs: number;
-  /** The snapshotting window of the GraphQL server memory in milliseconds. */
+  /** The snapshotting window of the server memory in milliseconds. */
   memorySnapshotWindow: number;
-  /** The GraphQL server on which the loadtest is running. */
+  /** The server on which the loadtest is running. */
   server: Server;
   /**
    * The GraphQL query to execute for the loadtest.
+   * Either `query` or `pathname` must be provided.
    */
-  query: string;
+  query?: string;
+  /**
+   * The HTTP pathname to request for the loadtest.
+   * Either `query` or `pathname` must be provided.
+   */
+  pathname?: string;
   /**
    * Whether to take heap snapshots on the end of the `idle` phase and then at the end
    * of the `calmdown` {@link LoadtestPhase phase} in each of the {@link runs}.
@@ -104,6 +110,7 @@ export async function loadtest(opts: LoadtestOptions): Promise<{
     memorySnapshotWindow,
     server,
     query,
+    pathname,
     takeHeapSnapshots,
     performHeapSampling,
     allowFailingRequests,
@@ -120,36 +127,58 @@ export async function loadtest(opts: LoadtestOptions): Promise<{
     throw new Error(`At least one run is necessary, got "${runs}"`);
   }
 
+  if (!query && !pathname) {
+    throw new Error('Either "query" or "pathname" must be provided');
+  }
+
+  if (query && pathname) {
+    throw new Error('Cannot provide both "query" and "pathname"');
+  }
+
   const startTime = new Date();
 
   // we create a random id to make sure the heapsnapshot files are unique and easily distinguishable in the filesystem
   // when running multiple loadtests in parallel. see e2e/opentelemetry memtest as an example
   const id = Math.random().toString(36).slice(2, 6);
 
-  // make sure the query works before starting the loadtests
-  // the request here matches the request done in loadtest-script.ts
-  const res = await fetch(
-    `${server.protocol}://localhost:${server.port}/graphql`,
-    {
-      method: 'POST',
-      headers: {
-        'content-type': 'application/json',
+  // make sure the endpoint works before starting the loadtests
+  // the request here matches the request done in loadtest-script.ts or http-loadtest-script.ts
+  if (query) {
+    const res = await fetch(
+      `${server.protocol}://localhost:${server.port}/graphql`,
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+        },
+        body: JSON.stringify({ query }),
       },
-      body: JSON.stringify({ query }),
-    },
-  );
-  const text = await res.text();
-  if (!res.ok) {
-    const err = new Error(
-      `Status is not 200, got status ${res.status} ${res.statusText} and body:\n${text}`,
     );
-    err.name = 'ResponseError';
-    throw err;
-  }
-  if (!text.includes('"data":{')) {
-    const err = new Error(`Body does not contain "data":\n${text}`);
-    err.name = 'ResponseError';
-    throw err;
+    const text = await res.text();
+    if (!res.ok) {
+      const err = new Error(
+        `Status is not 200, got status ${res.status} ${res.statusText} and body:\n${text}`,
+      );
+      err.name = 'ResponseError';
+      throw err;
+    }
+    if (!text.includes('"data":{')) {
+      const err = new Error(`Body does not contain "data":\n${text}`);
+      err.name = 'ResponseError';
+      throw err;
+    }
+  } else if (pathname) {
+    const res = await fetch(
+      `${server.protocol}://localhost:${server.port}${pathname}`,
+    );
+    if (!res.ok) {
+      const text = await res.text();
+      const err = new Error(
+        `Status is not 200, got status ${res.status} ${res.statusText} and body:\n${text}`,
+      );
+      err.name = 'ResponseError';
+      throw err;
+    }
   }
 
   const ctrl = new AbortController();
@@ -244,9 +273,17 @@ export async function loadtest(opts: LoadtestOptions): Promise<{
       'run',
       `--vus=${vus}`,
       `--duration=${duration}ms`,
-      `--env=URL=${server.protocol}://localhost:${server.port}/graphql`,
-      `--env=QUERY=${query}`,
-      path.join(__dirname, 'loadtest-script.ts'),
+      // graphql
+      query &&
+        `--env=URL=${server.protocol}://localhost:${server.port}/graphql`,
+      query && `--env=QUERY=${query}`,
+      // pathname
+      pathname &&
+        `--env=URL=${server.protocol}://localhost:${server.port}${pathname}`,
+      path.join(
+        __dirname,
+        query ? 'graphql-loadtest-script.ts' : 'pathname-loadtest-script.ts',
+      ),
     );
     await Promise.race([waitForExit, serverThrowOnExit, memorySnapshotting]);
 

internal/perf/src/loadtest.ts

@@ -54,7 +54,19 @@ export interface MemtestOptions
     | 'calmdown'
     | 'runs'
     | 'server'
+    | 'query'
+    | 'pathname'
   > {
+  /**
+   * The GraphQL query to execute for the loadtest.
+   * Either `query` or `pathname` must be provided.
+   */
+  query?: string;
+  /**
+   * The HTTP pathname to request for the loadtest.
+   * Either `query` or `pathname` must be provided.
+   */
+  pathname?: string;
   /**
    * The snapshotting window of the GraphQL server memory in milliseconds.
    *

internal/perf/src/memtest.ts

@@ -0,0 +1,25 @@
+import { check } from 'k6';
+import { test } from 'k6/execution';
+import http from 'k6/http';
+
+export default function () {
+  const url = __ENV['URL'];
+  if (!url) {
+    return test.abort('Environment variable "URL" not provided');
+  }
+
+  const res = http.get(url);
+
+  if (__ENV['ALLOW_FAILING_REQUESTS']) {
+    check(res, {
+      'status is 200': (res) => res.status === 200,
+    });
+  } else {
+    const body = res.body?.toString() || '';
+    if (res.status !== 200) {
+      return test.abort(
+        `Status is not 200, got status ${res.status} and body:\n${body}`,
+      );
+    }
+  }
+}

internal/perf/src/pathname-loadtest-script.ts

@@ -90,6 +90,9 @@ export function spawn(
   let stdout = '';
   let stderr = '';
   let stdboth = '';
+  // Limit stored output to prevent memory issues with high-volume logging
+  // Keep last 10MB of output (approximately 10 million characters)
+  const MAX_OUTPUT_LENGTH = 10_000_000;
   const proc: Proc = {
     waitForExit,
     kill(signal) {
@@ -141,13 +144,25 @@ export function spawn(
   child.stdout.on('data', (x) => {
     const str = x.toString();
     stdout += str;
+    if (stdout.length > MAX_OUTPUT_LENGTH) {
+      stdout = stdout.slice(-MAX_OUTPUT_LENGTH);
+    }
     stdboth += str;
+    if (stdboth.length > MAX_OUTPUT_LENGTH) {
+      stdboth = stdboth.slice(-MAX_OUTPUT_LENGTH);
+    }
     pipeLog({ cwd, pipeLogs }, x);
   });
   child.stderr.on('data', (x) => {
     const str = replaceStderr(x.toString());
     stderr += str;
+    if (stderr.length > MAX_OUTPUT_LENGTH) {
+      stderr = stderr.slice(-MAX_OUTPUT_LENGTH);
+    }
     stdboth += str;
+    if (stdboth.length > MAX_OUTPUT_LENGTH) {
+      stdboth = stdboth.slice(-MAX_OUTPUT_LENGTH);
+    }
     pipeLog({ cwd, pipeLogs }, x);
   });
 

internal/proc/src/index.ts

@@ -0,0 +1,133 @@
+import { createServer } from 'node:http';
+import { Logger } from '../src/logger';
+
+const syncLogger = new Logger({
+  level: 'debug',
+  writers: [
+    {
+      write: () => {
+        // noop
+      },
+    },
+  ],
+});
+
+const asyncLogger = new Logger({
+  level: 'debug',
+  writers: [
+    {
+      // Simulate async I/O with minimal delay
+      write: () => new Promise((resolve) => setImmediate(resolve)),
+    },
+  ],
+});
+
+// Large object for attribute testing
+const largeObject = {
+  data: new Array(100).fill(null).map((_, j) => ({
+    index: j,
+    value: `data-value-${j}`,
+    nested: {
+      level1: { level2: { level3: `nested-${j}` } },
+    },
+  })),
+};
+
+// Circular reference object
+const circularObj: any = { id: 1, data: 'test' };
+circularObj.self = circularObj;
+circularObj.nested = { parent: circularObj };
+
+const server = createServer((req, res) => {
+  try {
+    const parts = req.url?.split('/') || [];
+    const name = parts[parts.length - 1];
+    switch (name) {
+      case 'sync-logging':
+        // High-volume synchronous logging
+        syncLogger.info(
+          { iteration: Date.now(), data: 'test data' },
+          'Sync message',
+        );
+        break;
+
+      case 'async-logging':
+        // Async logging with I/O simulation
+        asyncLogger.info(
+          { iteration: Date.now(), data: 'test data' },
+          'Async message',
+        );
+        break;
+
+      case 'child-loggers': {
+        // Create child loggers
+        const child1 = syncLogger.child({ parent: 'root' }, 'child1 ');
+        const child2 = child1.child({ parent: 'child1' }, 'child2 ');
+        child2.info('Deep child message');
+        break;
+      }
+
+      case 'child-loggers-large': {
+        // Create child loggers
+        const child1 = syncLogger.child(
+          { parent: 'root', l1: largeObject },
+          'child1 ',
+        );
+        const child2 = child1.child(
+          { parent: 'child1', l2: largeObject },
+          'child2 ',
+        );
+        child2.info(largeObject, 'Large object message');
+        break;
+      }
+
+      case 'large-attributes':
+        // Log large attribute objects
+        syncLogger.info(largeObject, 'Large object message');
+        break;
+
+      case 'circular-refs':
+        // Log circular references
+        syncLogger.info(circularObj, 'Circular message');
+        break;
+
+      case 'lazy-attributes':
+        // Log with lazy attributes
+        syncLogger.info(
+          () => ({
+            id: Date.now(),
+            data: new Array(10).fill(null).map((_, j) => `item-${j}`),
+            timestamp: Date.now(),
+          }),
+          'Lazy message',
+        );
+        break;
+
+      case 'mixed-levels':
+        // Log at different levels
+        syncLogger.trace('Trace message');
+        syncLogger.debug('Debug message');
+        syncLogger.info('Info message');
+        syncLogger.warn('Warn message');
+        syncLogger.error('Error message');
+        break;
+
+      case 'level-changes':
+        // Change log levels
+        syncLogger.setLevel('debug');
+        syncLogger.debug('Debug enabled');
+        syncLogger.setLevel('info');
+        syncLogger.info('Back to info');
+        break;
+
+      default:
+        res.writeHead(404).end();
+        return;
+    }
+    res.writeHead(200).end();
+  } catch (err) {
+    res.writeHead(500).end((err as Error).stack);
+  }
+});
+
+server.listen(parseInt(process.env['PORT']!));