[ftr/remote] stop WebDriver logs polling when its session no longer exists

packages/kbn-ftr-common-functional-ui-services/services/remote/poll_for_log_entry.ts

@@ -8,87 +8,63 @@
 
 import { WebDriver, logging } from 'selenium-webdriver';
 import * as Rx from 'rxjs';
-import { mergeMap, catchError, mergeMapTo, delay, first } from 'rxjs/operators';
+import { mergeMap, catchError, delay, repeat } from 'rxjs/operators';
+import { NoSuchSessionError, NoSuchWindowError } from 'selenium-webdriver/lib/error';
+
+export const FINAL_LOG_ENTRY_PREFIX = 'WEBDRIVER SESSION IS STOPPED';
 
 /**
  * Create an observable that emits log entries representing the calls to log messages
  * available for a specific logger.
  */
-export function pollForLogEntry$(
-  driver: WebDriver,
-  type: string,
-  ms: number,
-  stop$: Rx.Observable<void>
-) {
+export function pollForLogEntry$(driver: WebDriver, type: string, ms: number) {
   const logCtrl = driver.manage().logs();
-  const poll$ = new Rx.BehaviorSubject(undefined);
 
-  const FINAL_MSG = '@@final@@';
+  // setup log polling
+  return Rx.defer(async () => await logCtrl.get(type)).pipe(
+    // filter and flatten list of entries
+    mergeMap((entries) =>
+      entries.filter((entry) => {
+        // ignore react devtools
+        if (entry.message.includes('Download the React DevTools')) {
+          return false;
+        }
 
-  return new Rx.Observable<logging.Entry>((subscriber) => {
-    subscriber.add(
-      stop$.pipe(first()).subscribe(() => {
-        driver
-          .executeScript(
-            `
-              if (window.flushCoverageToLog) {
-                window.flushCoverageToLog();
-              }
+        // down-level inline script errors
+        if (entry.message.includes('Refused to execute inline script')) {
+          entry.level = logging.getLevel('INFO');
+        }
 
-              console.log(${JSON.stringify(FINAL_MSG)})
-            `
-          )
-          .catch((error) => subscriber.error(error));
+        return true;
       })
-    );
-
-    subscriber.add(
-      poll$
-        .pipe(
-          delay(ms),
-
-          mergeMap(async () => await logCtrl.get(type)),
-
-          // filter and flatten list of entries
-          mergeMap((entries) => {
-            const filtered = entries.filter((entry) => {
-              if (entry.message.includes(FINAL_MSG)) {
-                poll$.complete();
-                return false;
-              }
+    ),
 
-              // ignore react devtools
-              if (entry.message.includes('Download the React DevTools')) {
-                return false;
-              }
+    // resubscribe when parent completes with delay by `ms` milliseconds
+    repeat(),
+    delay(ms),
 
-              // down-level inline script errors
-              if (entry.message.includes('Refused to execute inline script')) {
-                entry.level = logging.getLevel('INFO');
-              }
+    catchError((error, resubscribe) => {
+      if (
+        error instanceof NoSuchSessionError || // session is invalid
+        error instanceof NoSuchWindowError || // browser window crashed
+        error?.message.startsWith('ECONNREFUSED') // webdriver server is not responding, often after one of previous errors
+      ) {
+        // WebDriver session is invalid, sending the last log message
+        return Rx.concat([
+          new logging.Entry('SEVERE', `${FINAL_LOG_ENTRY_PREFIX}: ${error.message}`),
+        ]);
+      } else {
+        return Rx.concat(
+          // log error as a log entry
+          [new logging.Entry('SEVERE', `ERROR FETCHING BROWSR LOGS: ${error.message}`)],
 
-              return true;
-            });
-
-            if (!poll$.isStopped) {
-              // schedule next poll
-              poll$.next(undefined);
-            }
-
-            return filtered;
-          }),
-
-          catchError((error, resubscribe) => {
-            return Rx.concat(
-              // log error as a log entry
-              [new logging.Entry('SEVERE', `ERROR FETCHING BROWSR LOGS: ${error.message}`)],
-
-              // pause 10 seconds then resubscribe
-              Rx.of(1).pipe(delay(10 * 1000), mergeMapTo(resubscribe))
-            );
-          })
-        )
-        .subscribe(subscriber)
-    );
-  });
+          // pause 10 seconds then resubscribe
+          Rx.of(1).pipe(
+            delay(10 * 1000),
+            mergeMap(() => resubscribe)
+          )
+        );
+      }
+    })
+  );
 }

packages/kbn-ftr-common-functional-ui-services/services/remote/webdriver.ts

@@ -10,11 +10,11 @@ import { resolve } from 'path';
 import Fs from 'fs';
 
 import * as Rx from 'rxjs';
-import { mergeMap, map, takeUntil, catchError, ignoreElements } from 'rxjs/operators';
+import { mergeMap, map, catchError, ignoreElements, takeWhile } from 'rxjs/operators';
 import { Lifecycle } from '@kbn/test';
 import { ToolingLog } from '@kbn/tooling-log';
 import chromeDriver from 'chromedriver';
-import { Builder, logging } from 'selenium-webdriver';
+import { Builder, logging, WebDriver } from 'selenium-webdriver';
 import chrome from 'selenium-webdriver/chrome';
 import firefox from 'selenium-webdriver/firefox';
 import edge from 'selenium-webdriver/edge';
@@ -25,7 +25,7 @@ import { getLogger } from 'selenium-webdriver/lib/logging';
 import { installDriver } from 'ms-chromium-edge-driver';
 
 import { REPO_ROOT } from '@kbn/repo-info';
-import { pollForLogEntry$ } from './poll_for_log_entry';
+import { FINAL_LOG_ENTRY_PREFIX, pollForLogEntry$ } from './poll_for_log_entry';
 import { createStdoutSocket } from './create_stdout_stream';
 import { preventParallelCalls } from './prevent_parallel_calls';
 
@@ -139,6 +139,18 @@ function initChromiumOptions(browserType: Browsers, acceptInsecureCerts: boolean
   return options;
 }
 
+function pollForChromiumLogs$(session: WebDriver, logPollingMs: number) {
+  return pollForLogEntry$(session, logging.Type.BROWSER, logPollingMs).pipe(
+    takeWhile(
+      (loggingEntry: logging.Entry) => !loggingEntry.message.startsWith(FINAL_LOG_ENTRY_PREFIX)
+    ),
+    map(({ message, level: { name: level } }) => ({
+      message: message.replace(/\\n/g, '\n'),
+      level,
+    }))
+  );
+}
+
 let attemptCounter = 0;
 let edgePaths: { driverPath: string | undefined; browserPath: string | undefined };
 async function attemptToCreateCommand(
@@ -175,18 +187,7 @@ async function attemptToCreateCommand(
 
         return {
           session,
-          consoleLog$: pollForLogEntry$(
-            session,
-            logging.Type.BROWSER,
-            config.logPollingMs,
-            lifecycle.cleanup.after$
-          ).pipe(
-            takeUntil(lifecycle.cleanup.after$),
-            map(({ message, level: { name: level } }) => ({
-              message: message.replace(/\\n/g, '\n'),
-              level,
-            }))
-          ),
+          consoleLog$: pollForChromiumLogs$(session, config.logPollingMs),
         };
       }
 
@@ -203,18 +204,7 @@ async function attemptToCreateCommand(
             .build();
           return {
             session,
-            consoleLog$: pollForLogEntry$(
-              session,
-              logging.Type.BROWSER,
-              config.logPollingMs,
-              lifecycle.cleanup.after$
-            ).pipe(
-              takeUntil(lifecycle.cleanup.after$),
-              map(({ message, level: { name: level } }) => ({
-                message: message.replace(/\\n/g, '\n'),
-                level,
-              }))
-            ),
+            consoleLog$: pollForChromiumLogs$(session, config.logPollingMs),
           };
         } else {
           throw new Error(