telemetry: slog sender with block chunking

packages/internal/src/node/fs-stream.js

@@ -47,14 +47,12 @@ const noPath = /** @type {import('fs').PathLike} */ (
 /** @typedef {NonNullable<Awaited<ReturnType<typeof makeFsStreamWriter>>>} FsStreamWriter */
 /** @param {string | undefined | null} filePath */
 export const makeFsStreamWriter = async filePath => {
-  if (!filePath) {
-    return undefined;
-  }
+  if (!filePath) return undefined;
 
-  const handle = await (filePath !== '-' ? open(filePath, 'a') : undefined);
+  const handle = await (filePath === '-' ? undefined : open(filePath, 'a'));
 
   const stream = handle
-    ? createWriteStream(noPath, { fd: handle.fd })
+    ? createWriteStream(noPath, { autoClose: false, fd: handle.fd })
     : process.stdout;
   await fsStreamReady(stream);
 
@@ -95,23 +93,27 @@ export const makeFsStreamWriter = async filePath => {
 
   const flush = async () => {
     await flushed;
-    await handle?.sync().catch(err => {
-      if (err.code === 'EINVAL') {
-        return;
-      }
-      throw err;
-    });
+    if (handle)
+      await handle.sync().catch(err => {
+        if (err.code === 'EINVAL') {
+          return;
+        }
+        throw err;
+      });
   };
 
   const close = async () => {
     // TODO: Consider creating a single Error here to use a write rejection
     closed = true;
     await flush();
-    // @ts-expect-error calling a possibly missing method
-    stream.close?.();
+
+    if (handle) {
+      await new Promise(resolve => stream.end(() => resolve(null)));
+      await handle.close();
+    }
   };
 
   stream.on('error', err => updateFlushed(Promise.reject(err)));
 
-  return harden({ write, flush, close });
+  return harden({ close, filePath, flush, write });
 };

packages/telemetry/src/block-slog.js

@@ -0,0 +1,135 @@
+/* eslint-env node */
+
+import { makeFsStreamWriter } from '@agoric/internal/src/node/fs-stream.js';
+import {
+  makeContextualSlogProcessor,
+  SLOG_TYPES,
+} from './context-aware-slog.js';
+import getContextFilePersistenceUtils, {
+  DEFAULT_CONTEXT_FILE,
+} from './context-aware-slog-persistent-util.js';
+import { serializeSlogObj } from './serialize-slog-obj.js';
+
+/**
+ * @typedef {import('./context-aware-slog.js').Slog} Slog
+ * @typedef {ReturnType<ReturnType<typeof makeContextualSlogProcessor>>} ContextSlog
+ */
+
+/**
+ * @param {import('./index.js').MakeSlogSenderOptions} options
+ */
+export const makeSlogSender = async options => {
+  const { CHAIN_ID, CONTEXTUAL_BLOCK_SLOGS } = options.env || {};
+  if (!(options.stateDir || CONTEXTUAL_BLOCK_SLOGS))
+    return console.error(
+      'Ignoring invocation of slogger "block-slog" without the presence of "stateDir" or "CONTEXTUAL_BLOCK_SLOGS"',
+    );
+
+  const persistenceUtils = getContextFilePersistenceUtils(
+    process.env.SLOG_CONTEXT_FILE_PATH ||
+      `${options.stateDir}/${DEFAULT_CONTEXT_FILE}`,
+  );
+
+  const contextualSlogProcessor = makeContextualSlogProcessor(
+    { 'chain-id': CHAIN_ID },
+    persistenceUtils,
+  );
+  /**
+   * @type {ReturnType<typeof createFileStream> | null}
+   */
+  let createFileStreamPromise = null;
+  /**
+   * @type {Awaited<ReturnType<typeof makeFsStreamWriter>> | null}
+   */
+  let currentStream = null;
+
+  /**
+   * Immediately frees the `currentStream` assignment and lazily closes the open file stream
+   */
+  const closeStream = () => {
+    if (currentStream) {
+      const streamClosePromise = currentStream.close();
+      currentStream = null;
+      return streamClosePromise;
+    } else console.error('No stream to close');
+  };
+
+  /**
+   * @param {string} fileName
+   */
+  const createFileStream = async fileName => {
+    if (currentStream)
+      throw Error(`Stream already open on file ${currentStream.filePath}`);
+
+    const filePath = `${options.stateDir || CONTEXTUAL_BLOCK_SLOGS}/slogfile_${fileName}.jsonl`;
+    currentStream = await makeFsStreamWriter(filePath);
+
+    if (!currentStream)
+      throw Error(`Couldn't create a write stream on file "${filePath}"`);
+  };
+
+  /**
+   * @param {import('./context-aware-slog.js').Slog} slog
+   */
+  const slogSender = async slog => {
+    await new Promise(resolve => resolve(null));
+
+    const { blockHeight, type: slogType } = slog;
+
+    switch (slogType) {
+      case SLOG_TYPES.KERNEL.INIT.START: {
+        createFileStreamPromise = createFileStream(
+          `init_${new Date().getTime()}`,
+        );
+        break;
+      }
+      case SLOG_TYPES.COSMIC_SWINGSET.BOOTSTRAP_BLOCK.START: {
+        createFileStreamPromise = createFileStream(
+          `bootstrap_${new Date().getTime()}`,
+        );
+        break;
+      }
+      case SLOG_TYPES.COSMIC_SWINGSET.BEGIN_BLOCK: {
+        createFileStreamPromise = createFileStream(
+          `block_${blockHeight}_${new Date().getTime()}`,
+        );
+        break;
+      }
+      default: {
+        break;
+      }
+    }
+
+    const contextualSlog = contextualSlogProcessor(slog);
+
+    if (createFileStreamPromise) await createFileStreamPromise;
+    createFileStreamPromise = null;
+
+    writeSlogToStream(contextualSlog)?.catch(console.error);
+
+    switch (slogType) {
+      case SLOG_TYPES.KERNEL.INIT.FINISH:
+      case SLOG_TYPES.COSMIC_SWINGSET.BOOTSTRAP_BLOCK.FINISH:
+      case SLOG_TYPES.COSMIC_SWINGSET.AFTER_COMMIT_STATS: {
+        closeStream()?.catch(console.error);
+        break;
+      }
+      default: {
+        break;
+      }
+    }
+  };
+
+  /**
+   * @param {ReturnType<contextualSlogProcessor>} slog
+   */
+  const writeSlogToStream = slog =>
+    !currentStream
+      ? console.error(`No stream available for slog type "${slog.body.type}"`) // eslint-disable-next-line prefer-template
+      : currentStream.write(serializeSlogObj(slog) + '\n');
+
+  return Object.assign(slogSender, {
+    forceFlush: () => currentStream?.flush(),
+    shutdown: closeStream,
+  });
+};

packages/telemetry/src/context-aware-slog-file.js

@@ -2,6 +2,9 @@
 
 import { makeFsStreamWriter } from '@agoric/internal/src/node/fs-stream.js';
 import { makeContextualSlogProcessor } from './context-aware-slog.js';
+import getContextFilePersistenceUtils, {
+  DEFAULT_CONTEXT_FILE,
+} from './context-aware-slog-persistent-util.js';
 import { serializeSlogObj } from './serialize-slog-obj.js';
 
 /**
@@ -21,9 +24,17 @@ export const makeSlogSender = async options => {
       `Couldn't create a write stream on file "${CONTEXTUAL_SLOGFILE}"`,
     );
 
-  const contextualSlogProcessor = makeContextualSlogProcessor({
-    'chain-id': CHAIN_ID,
-  });
+  const persistenceUtils = getContextFilePersistenceUtils(
+    process.env.SLOG_CONTEXT_FILE_PATH ||
+      `${options.stateDir}/${DEFAULT_CONTEXT_FILE}`,
+  );
+
+  const contextualSlogProcessor = makeContextualSlogProcessor(
+    {
+      'chain-id': CHAIN_ID,
+    },
+    persistenceUtils,
+  );
 
   /**
    * @param {import('./context-aware-slog.js').Slog} slog

packages/telemetry/src/context-aware-slog-persistent-util.js

@@ -0,0 +1,39 @@
+import { readFileSync, writeFileSync } from 'fs';
+import { serializeSlogObj } from './serialize-slog-obj.js';
+
+export const DEFAULT_CONTEXT_FILE = 'slog-context.json';
+const FILE_ENCODING = 'utf8';
+
+/**
+ * @param {string} filePath
+ */
+const getContextFilePersistenceUtils = filePath => {
+  console.warn(`Using file ${filePath} for slogger context`);
+
+  return {
+    /**
+     * @param {import('./context-aware-slog.js').Context} context
+     */
+    persistContext: context => {
+      try {
+        writeFileSync(filePath, serializeSlogObj(context), FILE_ENCODING);
+      } catch (err) {
+        console.error('Error writing context to file: ', err);
+      }
+    },
+
+    /**
+     * @returns {import('./context-aware-slog.js').Context | null}
+     */
+    restoreContext: () => {
+      try {
+        return JSON.parse(readFileSync(filePath, FILE_ENCODING));
+      } catch (parseErr) {
+        console.error('Error reading context from file: ', parseErr);
+        return null;
+      }
+    },
+  };
+};
+
+export default getContextFilePersistenceUtils;

packages/telemetry/src/context-aware-slog.js

@@ -49,7 +49,7 @@
  * }} Slog
  */
 
-const SLOG_TYPES = {
+export const SLOG_TYPES = {
   CLIST: 'clist',
   CONSOLE: 'console',
   COSMIC_SWINGSET: {
@@ -60,8 +60,8 @@ const SLOG_TYPES = {
       START: 'cosmic-swingset-bootstrap-block-start',
     },
     COMMIT: {
-      FINISH: 'cosmic-swingset-commit-finish',
-      START: 'cosmic-swingset-commit-start',
+      FINISH: 'cosmic-swingset-commit-block-finish',
+      START: 'cosmic-swingset-commit-block-start',
     },
     END_BLOCK: {
       FINISH: 'cosmic-swingset-end-block-finish',

packages/telemetry/src/otel-context-aware-slog.js

@@ -1,51 +1,19 @@
 /* eslint-env node */
+
 import { logs, SeverityNumber } from '@opentelemetry/api-logs';
 import { OTLPLogExporter } from '@opentelemetry/exporter-logs-otlp-http';
 import { Resource } from '@opentelemetry/resources';
 import {
   LoggerProvider,
   SimpleLogRecordProcessor,
 } from '@opentelemetry/sdk-logs';
-import { readFileSync, writeFileSync } from 'fs';
 import { makeContextualSlogProcessor } from './context-aware-slog.js';
+import getContextFilePersistenceUtils, {
+  DEFAULT_CONTEXT_FILE,
+} from './context-aware-slog-persistent-util.js';
 import { getResourceAttributes } from './index.js';
 import { serializeSlogObj } from './serialize-slog-obj.js';
 
-const DEFAULT_CONTEXT_FILE = 'slog-context.json';
-const FILE_ENCODING = 'utf8';
-
-/**
- * @param {string} filePath
- */
-export const getContextFilePersistenceUtils = filePath => {
-  console.warn(`Using file ${filePath} for slogger context`);
-
-  return {
-    /**
-     * @param {import('./context-aware-slog.js').Context} context
-     */
-    persistContext: context => {
-      try {
-        writeFileSync(filePath, serializeSlogObj(context), FILE_ENCODING);
-      } catch (err) {
-        console.error('Error writing context to file: ', err);
-      }
-    },
-
-    /**
-     * @returns {import('./context-aware-slog.js').Context | null}
-     */
-    restoreContext: () => {
-      try {
-        return JSON.parse(readFileSync(filePath, FILE_ENCODING));
-      } catch (parseErr) {
-        console.error('Error reading context from file: ', parseErr);
-        return null;
-      }
-    },
-  };
-};
-
 /**
  * @param {import('./index.js').MakeSlogSenderOptions} options
  */