[ML] Explain Log Rate Spikes: Fix stream flushing. (elastic#140506)

This is a temporary fix for response streaming. The current cloud environment buffers each stream with chunks up to 4KB. To force trigger flushing, we send along a 4KB dummy payload to trigger an update. This fixes a stale loading bar for Explain Log Rate Spikes. Once the cloud environment's proxy has been updated to support flushing below the 4KB threshold, we can remove this fix again.

x-pack/packages/ml/aiops_utils/src/stream_factory.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import crypto from 'crypto';
 import { Stream } from 'stream';
 import * as zlib from 'zlib';
 
@@ -42,7 +43,8 @@ interface StreamFactoryReturnType<T = unknown> {
  */
 export function streamFactory<T = string>(
   headers: Headers,
-  logger: Logger
+  logger: Logger,
+  flushFix?: boolean
 ): StreamFactoryReturnType<T>;
 /**
  * Sets up a response stream with support for gzip compression depending on provided
@@ -53,7 +55,8 @@ export function streamFactory<T = string>(
  */
 export function streamFactory<T = unknown>(
   headers: Headers,
-  logger: Logger
+  logger: Logger,
+  flushFix: boolean = false
 ): StreamFactoryReturnType<T> {
   let streamType: StreamType;
   const isCompressed = acceptCompression(headers);
@@ -82,7 +85,14 @@ export function streamFactory<T = unknown>(
     }
 
     try {
-      const line = typeof d !== 'string' ? `${JSON.stringify(d)}${DELIMITER}` : d;
+      const line =
+        streamType === 'ndjson'
+          ? `${JSON.stringify({
+              ...d,
+              // This is a temporary fix for response streaming with proxy configurations that buffer responses up to 4KB in size.
+              ...(flushFix ? { flushPayload: crypto.randomBytes(4096).toString('hex') } : {}),
+            })}${DELIMITER}`
+          : d;
       stream.write(line);
     } catch (e) {
       logger.error(`Could not serialize or stream data chunk: ${e.toString()}`);

x-pack/plugins/aiops/server/routes/explain_log_rate_spikes.ts

@@ -81,7 +81,8 @@ export const defineExplainLogRateSpikesRoute = (
 
       const { end, push, responseWithHeaders } = streamFactory<AiopsExplainLogRateSpikesApiAction>(
         request.headers,
-        logger
+        logger,
+        true
       );
 
       function endWithUpdatedLoadingState() {