(superset) branch master updated: feat(websocket): improve operational logging and crash safety (#40868)

[ru...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rusackas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git


The following commit(s) were added to refs/heads/master by this push:
     new 62084f4015e feat(websocket): improve operational logging and crash safety (#40868)
62084f4015e is described below

commit 62084f4015ed0d6e97147c5da1559425a9c4e0f5
Author: Evan Rusackas <ev...@preset.io>
AuthorDate: Tue Jun 9 09:33:13 2026 -0700

    feat(websocket): improve operational logging and crash safety (#40868)
    
    Co-authored-by: Claude Code <no...@anthropic.com>
---
 superset-websocket/spec/index.test.ts |  4 ++-
 superset-websocket/src/index.ts       | 47 +++++++++++++++++++++++++++++------
 superset-websocket/src/logger.ts      |  1 +
 3 files changed, 43 insertions(+), 9 deletions(-)

diff --git a/superset-websocket/spec/index.test.ts b/superset-websocket/spec/index.test.ts
index ed306260974..bb7d3151175 100644
--- a/superset-websocket/spec/index.test.ts
+++ b/superset-websocket/spec/index.test.ts
@@ -43,7 +43,7 @@ const mockRedisXrange = jest.fn() as jest.MockedFunction<MockedRedisXrange>;
 jest.mock('ws');
 jest.mock('ioredis', () => {
   return jest.fn().mockImplementation(() => {
-    return { xrange: mockRedisXrange };
+    return { xrange: mockRedisXrange, on: jest.fn() };
   });
 });
 
@@ -526,6 +526,8 @@ describe('server', () => {
       server.httpUpgrade(request, socket, Buffer.alloc(5));
       expect(socketDestroySpy).toHaveBeenCalled();
       expect(wssUpgradeSpy).not.toHaveBeenCalled();
+      // rejected upgrades are counted for auditability
+      expect(statsdIncrementMock).toHaveBeenCalledWith('ws_upgrade_rejected');
     });
 
     test('valid JWT, no channel', async () => {
diff --git a/superset-websocket/src/index.ts b/superset-websocket/src/index.ts
index bebc89a20ec..38e09521b1d 100644
--- a/superset-websocket/src/index.ts
+++ b/superset-websocket/src/index.ts
@@ -18,6 +18,7 @@
  */
 import * as http from 'http';
 import * as net from 'net';
+import { inspect } from 'util';
 import WebSocket, { WebSocketServer } from 'ws';
 import { randomUUID } from 'crypto';
 import jwt, { Algorithm } from 'jsonwebtoken';
@@ -96,15 +97,15 @@ export const statsd = new StatsD({
 
 // enforce JWT secret length
 if (startServer && opts.jwtSecret.length < 32) {
-  console.error('ERROR: Please provide a JWT secret at least 32 bytes long');
+  logger.error('Please provide a JWT secret at least 32 bytes long');
   process.exit(1);
 }
 
 if (startServer && opts.jwtSecret.startsWith('CHANGE-ME')) {
-  console.warn(
-    'WARNING: it appears your secret in your config.json is insecure',
+  logger.warn(
+    'It appears your secret in your config.json is insecure. ' +
+      'DO NOT USE IN PRODUCTION',
   );
-  console.warn('DO NOT USE IN PRODUCTION');
 }
 
 export const buildRedisOpts = (baseConfig: RedisConfig) => {
@@ -140,6 +141,9 @@ export const buildRedisOpts = (baseConfig: RedisConfig) => {
 
 // initialize servers
 const redis = new Redis(buildRedisOpts(opts.redis));
+redis.on('error', (err: Error) => {
+  logger.error(`Redis connection error: ${err.message}`);
+});
 const httpServer = http.createServer();
 export const wss = new WebSocketServer({
   noServer: true,
@@ -267,7 +271,9 @@ export const subscribeToGlobalStream = async (
  * Callback function to process events received from a Redis Stream
  */
 export const processStreamResults = (results: StreamResult[]): void => {
-  logger.debug(`events received: ${results}`);
+  // Log only the batch size, not the raw payloads, which carry user and
+  // job identifiers.
+  logger.debug(`events received: count=${results.length}`);
   results.forEach(item => {
     try {
       const id = item[0];
@@ -437,8 +443,14 @@ export const httpUpgrade = (
   try {
     readChannelId(request);
   } catch (err) {
-    // JWT invalid, do not establish a WebSocket connection
-    logger.error(err);
+    // Token invalid/absent: do not establish a WebSocket connection. Record a
+    // structured warning (with the request's remote address) so rejected
+    // upgrade attempts are auditable, without logging the token itself.
+    statsd.increment('ws_upgrade_rejected');
+    logger.warn(
+      `Rejected WebSocket upgrade from ${request.socket.remoteAddress ?? 'unknown'}: ` +
+        `${(err as Error).message}`,
+    );
     socket.destroy();
     return;
   }
@@ -513,9 +525,28 @@ export const cleanChannel = (channel: string) => {
 // server startup
 
 if (startServer) {
+  // Last-resort handlers so an unhandled async error is recorded through the
+  // configured logger instead of printing a default trace (or, for an
+  // unhandled rejection, terminating the process on newer Node versions).
+  process.on('unhandledRejection', (reason: unknown) => {
+    // Normalize the reason defensively: a raw template interpolation throws on
+    // a Symbol (or other exotic value), which would crash this last-resort
+    // handler. `inspect` safely stringifies any value.
+    logger.error(`Unhandled promise rejection: ${inspect(reason)}`);
+  });
+  process.on('uncaughtException', (err: unknown) => {
+    // JavaScript can throw non-Error values (including null), so guard the
+    // shape before dereferencing instead of assuming an Error is present.
+    const detail =
+      err instanceof Error ? (err.stack ?? err.message) : inspect(err);
+    logger.error(`Uncaught exception: ${detail}`);
+  });
+
   // init server event listeners
   wss.on('connection', function (ws: WebSocket) {
-    ws.on('error', console.error);
+    ws.on('error', (err: Error) =>
+      logger.error(`socket error: ${err.message}`),
+    );
   });
   wss.on('connection', wsConnection);
   httpServer.on('request', httpRequest);
diff --git a/superset-websocket/src/logger.ts b/superset-websocket/src/logger.ts
index c6efc2a5043..96ce6dda3d3 100644
--- a/superset-websocket/src/logger.ts
+++ b/superset-websocket/src/logger.ts
@@ -43,6 +43,7 @@ export function createLogger(opts: LoggingOptionsType) {
     level: opts.logLevel,
     transports: logTransports,
     format: winston.format.combine(
+      winston.format.timestamp(),
       winston.format.errors({ stack: true }),
       winston.format.json(),
     ),