PPLT_3672: [Smart concurrency] Scaling up/down asset discovery queue

.github/workflows/test.yml

@@ -28,7 +28,7 @@ jobs:
             ${{ hashFiles('.github/.cache-key') }}/
       - run: yarn
       - run: yarn build
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
           name: dist
           path: packages/*/dist
@@ -56,6 +56,7 @@ jobs:
           - '@percy/cli-config'
           - '@percy/sdk-utils'
           - '@percy/webdriver-utils'
+          - '@percy/monitoring'
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v3
@@ -75,7 +76,7 @@ jobs:
           restore-keys: >
             ${{ runner.os }}/node-${{ matrix.node }}/
             ${{ hashFiles('.github/.cache-key') }}/
-      - uses: actions/download-artifact@v3
+      - uses: actions/download-artifact@v4
         with:
           name: dist
           path: packages

.github/workflows/windows.yml

@@ -28,7 +28,7 @@ jobs:
             ${{ hashFiles('.github/.cache-key') }}/
       - run: yarn
       - run: yarn build
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
           name: dist
           path: packages/*/dist
@@ -55,6 +55,7 @@ jobs:
           - '@percy/cli-config'
           - '@percy/sdk-utils'
           - '@percy/webdriver-utils'
+          - '@percy/monitoring'
     runs-on: windows-latest
     steps:
       - uses: actions/checkout@v3
@@ -74,7 +75,7 @@ jobs:
           restore-keys: >
             ${{ runner.os }}/node-14/
             ${{ hashFiles('.github/.cache-key') }}/
-      - uses: actions/download-artifact@v3
+      - uses: actions/download-artifact@v4
         with:
           name: dist
           path: packages

packages/cli-exec/test/exec.test.js

@@ -3,6 +3,9 @@ import exec from '@percy/cli-exec';
 describe('percy exec', () => {
   beforeEach(async () => {
     process.env.PERCY_TOKEN = '<<PERCY_TOKEN>>';
+    // due to lot of start calls, it slows spec to finish
+    // therefore disabling it
+    process.env.PERCY_DISABLE_SYSTEM_MONITORING = 'true';
     process.env.PERCY_FORCE_PKG_VALUE = JSON.stringify({ name: '@percy/client', version: '1.0.0' });
     jasmine.DEFAULT_TIMEOUT_INTERVAL = 25000;
     await setupTest();
@@ -24,6 +27,7 @@ describe('percy exec', () => {
     delete process.env.PERCY_PARALLEL_TOTAL;
     delete process.env.PERCY_PARTIAL_BUILD;
     delete process.env.PERCY_CLIENT_ERROR_LOGS;
+    delete process.env.PERCY_DISABLE_SYSTEM_MONITORING;
   });
 
   describe('projectType is app', () => {
@@ -277,7 +281,6 @@ describe('percy exec', () => {
       'setTimeout(() => process.exit(1), 5000)'
     )]);
 
-    // wait until the process starts
     await new Promise(r => setTimeout(r, 1000));
     expect(logger.stdout).toEqual(jasmine.arrayContaining([
       jasmine.stringContaining('[percy] Running "node --eval ')

packages/cli-exec/test/start.test.js

@@ -14,6 +14,9 @@ describe('percy exec:start', () => {
 
   beforeEach(async () => {
     process.env.PERCY_TOKEN = '<<PERCY_TOKEN>>';
+
+    // disabling as it increases spec time and logs system info
+    process.env.PERCY_DISABLE_SYSTEM_MONITORING = 'true';
     process.env.PERCY_FORCE_PKG_VALUE = JSON.stringify({ name: '@percy/client', version: '1.0.0' });
     await setupTest();
 
@@ -29,6 +32,7 @@ describe('percy exec:start', () => {
     delete process.env.PERCY_ENABLE;
     delete process.env.PERCY_PARALLEL_TOTAL;
     delete process.env.PERCY_PARTIAL_BUILD;
+    delete process.env.PERCY_DISABLE_SYSTEM_MONITORING;
 
     // it's important that percy is still running or we terminate the test process
     if (started) process.emit('SIGTERM');

packages/core/package.json

@@ -48,6 +48,7 @@
     "@percy/dom": "1.30.7-beta.2",
     "@percy/logger": "1.30.7-beta.2",
     "@percy/webdriver-utils": "1.30.7-beta.2",
+    "@percy/monitoring": "1.30.7-beta.1",
     "content-disposition": "^0.5.4",
     "cross-spawn": "^7.0.3",
     "extract-zip": "^2.0.1",

packages/core/src/discovery.js

@@ -355,7 +355,7 @@ async function* captureSnapshotResources(page, snapshot, options) {
 // one concurrently. When skipping asset discovery, the callback is called immediately for each
 // snapshot, also processing snapshot resources when not dry-running.
 export async function* discoverSnapshotResources(queue, options, callback) {
-  let { snapshots, skipDiscovery, dryRun } = options;
+  let { snapshots, skipDiscovery, dryRun, checkAndUpdateConcurrency } = options;
 
   yield* yieldAll(snapshots.reduce((all, snapshot) => {
     debugSnapshotOptions(snapshot);
@@ -368,6 +368,10 @@ export async function* discoverSnapshotResources(queue, options, callback) {
         callback(dryRun ? snap : processSnapshotResources(snap));
       }
     } else {
+      // update concurrency before pushing new job in discovery queue
+      // if case of monitoring is stopped due to in-activity,
+      // it can take upto 1 sec to execute this fun
+      checkAndUpdateConcurrency();
       all.push(queue.push(snapshot, callback));
     }
 

packages/core/src/percy.js

@@ -26,10 +26,16 @@ import {
   discoverSnapshotResources,
   createDiscoveryQueue
 } from './discovery.js';
+import Monitoring from '@percy/monitoring';
 import { WaitForJob } from './wait-for-job.js';
 
 const MAX_SUGGESTION_CALLS = 10;
 
+// If no activity is done for 5 mins, we will stop monitoring
+// system metric eg: (cpu load && memory usage)
+const MONITOR_ACTIVITY_TIMEOUT = 300000;
+const MONITORING_INTERVAL_MS = 5000; // 5 sec
+
 // A Percy instance will create a new build when started, handle snapshot creation, asset discovery,
 // and resource uploads, and will finalize the build when stopped. Snapshots are processed
 // concurrently and the build is not finalized until all snapshots have been handled.
@@ -107,8 +113,15 @@ export class Percy {
     this.browser = new Browser(this);
 
     this.#discovery = createDiscoveryQueue(this);
+    this.discoveryMaxConcurrency = this.#discovery.concurrency;
     this.#snapshots = createSnapshotsQueue(this);
 
+    this.monitoring = new Monitoring();
+    // used continue monitoring if there is activity going on
+    // if there is none, stop it
+    this.resetMonitoringId = null;
+    this.monitoringCheckLastExecutedAt = null;
+
     // generator methods are wrapped to autorun and return promises
     for (let m of ['start', 'stop', 'flush', 'idle', 'snapshot', 'upload']) {
       // the original generator can be referenced with percy.yield.<method>
@@ -117,6 +130,29 @@ export class Percy {
     }
   }
 
+  systemMonitoringEnabled() {
+    return (process.env.PERCY_DISABLE_SYSTEM_MONITORING !== 'true');
+  }
+
+  async configureSystemMonitor() {
+    await this.monitoring.startMonitoring({ interval: MONITORING_INTERVAL_MS });
+    this.resetSystemMonitor();
+  }
+
+  // Debouncing logic to only stop Monitoring system
+  // if there is no any activity for 5 mins
+  // means, no job is pushed in queue from 5 mins
+  resetSystemMonitor() {
+    if (this.resetMonitoringId) {
+      clearTimeout(this.resetMonitoringId);
+      this.resetMonitoringId = null;
+    }
+
+    this.resetMonitoringId = setTimeout(() => {
+      this.monitoring.stopMonitoring();
+    }, MONITOR_ACTIVITY_TIMEOUT);
+  }
+
   // Shortcut for controlling the global logger's log level.
   loglevel(level) {
     return logger.loglevel(level);
@@ -169,6 +205,13 @@ export class Percy {
     this.cliStartTime = new Date().toISOString();
 
     try {
+      // started monitoring system metrics
+
+      if (this.systemMonitoringEnabled()) {
+        await this.configureSystemMonitor();
+        await this.monitoring.logSystemInfo();
+      }
+
       if (process.env.PERCY_CLIENT_ERROR_LOGS !== 'false') {
         this.log.warn('Notice: Percy collects CI logs to improve service and enhance your experience. These logs help us debug issues and provide insights on your dashboards, making it easier to optimize the product experience. Logs are stored securely for 30 days. You can opt out anytime with export PERCY_CLIENT_ERROR_LOGS=false, but keeping this enabled helps us offer the best support and features.');
       }
@@ -298,13 +341,66 @@ export class Percy {
       this.log.error(err);
       throw err;
     } finally {
+      // stop monitoring system metric, if not already stopped
+      this.monitoring.stopMonitoring();
+      clearTimeout(this.resetMonitoringId);
+
       // This issue doesn't comes under regular error logs,
       // it's detected if we just and stop percy server
       await this.checkForNoSnapshotCommandError();
       await this.sendBuildLogs();
     }
   }
 
+  checkAndUpdateConcurrency() {
+    // early exit if monitoring is disabled
+    if (!this.systemMonitoringEnabled()) return;
+
+    // early exit if asset discovery concurrency change is disabled
+    // NOTE: system monitoring will still be running as only concurrency
+    // change is disabled
+    if (process.env.PERCY_DISABLE_CONCURRENCY_CHANGE === 'true') return;
+
+    // start system monitoring if not already doing...
+    // this doesn't handle cases where there is suggest cpu spikes
+    // in less 1 sec range and if monitoring is not in running state
+    if (this.monitoringCheckLastExecutedAt && Date.now() - this.monitoringCheckLastExecutedAt < MONITORING_INTERVAL_MS) return;
+
+    if (!this.monitoring.running) this.configureSystemMonitor();
+    else this.resetSystemMonitor();
+
+    // early return if last executed was less than 5 seconds
+    // as we will get the same cpu/mem info under 5 sec interval
+    const { cpuInfo, memoryUsageInfo } = this.monitoring.getMonitoringInfo();
+    this.log.debug(`cpuInfo: ${JSON.stringify(cpuInfo)}`);
+    this.log.debug(`memoryInfo: ${JSON.stringify(memoryUsageInfo)}`);
+
+    if (cpuInfo.currentUsagePercent >= 80 || memoryUsageInfo.currentUsagePercent >= 80) {
+      let currentConcurrent = this.#discovery.concurrency;
+
+      // concurrency must be betweeen [1, (default/user defined value)]
+      let newConcurrency = Math.max(1, parseInt(currentConcurrent / 2));
+      newConcurrency = Math.min(this.discoveryMaxConcurrency, newConcurrency);
+
+      this.log.debug(`Downscaling discovery browser concurrency from ${this.#discovery.concurrency} to ${newConcurrency}`);
+      this.#discovery.set({ concurrency: newConcurrency });
+    } else if (cpuInfo.currentUsagePercent <= 50 && memoryUsageInfo.currentUsagePercent <= 50) {
+      let currentConcurrent = this.#discovery.concurrency;
+      let newConcurrency = currentConcurrent + 2;
+
+      // concurrency must be betweeen [1, (default/user-defined value)]
+      newConcurrency = Math.min(this.discoveryMaxConcurrency, newConcurrency);
+      newConcurrency = Math.max(1, newConcurrency);
+
+      this.log.debug(`Upscaling discovery browser concurrency from ${this.#discovery.concurrency} to ${newConcurrency}`);
+      this.#discovery.set({ concurrency: newConcurrency });
+    }
+
+    // reset timeout to stop monitoring after no-activity of 5 mins
+    this.resetSystemMonitor();
+    this.monitoringCheckLastExecutedAt = Date.now();
+  }
+
   // Takes one or more snapshots of a page while discovering resources to upload with the resulting
   // snapshots. Once asset discovery has completed for the provided snapshots, the queued task will
   // resolve and an upload task will be queued separately.
@@ -351,7 +447,7 @@ export class Percy {
         yield* discoverSnapshotResources(this.#discovery, {
           skipDiscovery: this.skipDiscovery,
           dryRun: this.dryRun,
-
+          checkAndUpdateConcurrency: this.checkAndUpdateConcurrency.bind(this),
           snapshots: yield* gatherSnapshots(options, {
             meta: { build: this.build },
             config: this.config