Groovy breaklet to test how many workflows are needed to break AEM's Inbox page

breaklet-max-workflow-instances-touch-ui.groovy

import com.day.cq.workflow.model.WorkflowModel
import com.day.cq.workflow.WorkflowService
import com.day.cq.workflow.WorkflowSession
import com.day.cq.workflow.exec.WorkflowData
import com.day.cq.workflow.event.WorkflowEvent
import groovy.transform.InheritConstructors

//-------------------------------------
// *** Max running workflows
//-------------------------------------
// Breaklet Config
//-------------------------------------
def breakForDurationInSec = 60
def maxIterations = 100

def workflowsPerIteration = 500

def sessionUser = 'admin'
def sessionPassword = 'admin'

//-------------------------------------
// Breaklet
//-------------------------------------
new MyBreaklet(this, breakForDurationInSec, sessionUser, sessionPassword, maxIterations).run ({ breaklet, currentIt, itSession  ->
    breaklet.registerWorkflowStartedEventHandler()
    WorkflowSession wfSession = getService(WorkflowService.class).getWorkflowSession(session)
    for(int workflowNum = 1; workflowNum <= workflowsPerIteration; workflowNum++) {
        breaklet.startWorkflow(wfSession)
    }

    breaklet.waitForWorkflowEvents(workflowsPerIteration)

    def browserDurationInMillis = breaklet.runInBrowser("""

        let timeToLoadInboxMessagesStart = new Date().getTime();
        await page.goto(baseUrl + '/aem/inbox', { waitUntil: "networkidle2", timeout: ${breakForDurationInSec * 1000} });

        await waitForXPath(page, "//span[text() = 'Approve content']")
        let timeToLoadInboxMessages = new Date().getTime() - timeToLoadInboxMessagesStart;
        resultProps['timeToLoadInboxMessages'] = timeToLoadInboxMessages
      """)

    def browserResultProps = breaklet.browser.resultProps
    def timeToLoadInboxMessages = browserResultProps.getProperty('timeToLoadInboxMessages')
    breaklet.logOpFinished("Inbox messages were loaded", "", timeToLoadInboxMessages)

}, { breaklet ->
    breaklet.waitForWorkflowsVisibleForMBean(maxIterations * workflowsPerIteration)
    breaklet.purgeAllWorkflows()
})

@InheritConstructors
class MyBreaklet extends Breaklet {

  def workflowToStart = '/var/workflow/models/request_for_activation'
  def workflowPayload = '/content/we-retail/us/en/user/smartlist'

  def workflowStartedEventHandler = null
  def workflowStartedEventHandlerReg = null

  def startWorkflow = { wfSession ->
      WorkflowModel wfModel = wfSession.getModel(workflowToStart)
      WorkflowData wfData = wfSession.newWorkflowData('JCR_PATH', workflowPayload)
      wfSession.startWorkflow(wfModel, wfData)
  }

  def purgeAllWorkflows = {
    performPostRequest("/system/console/jmx/com.adobe.granite.workflow%3Atype%3DMaintenance/op/purgeCompleted/java.lang.String%2Cjava.lang.Integer%2Cjava.lang.Boolean", "model+%5Boptional%5D=&number+of+days+since+workflow+has+been+completed=0&dry+run=")
    performPostRequest("/system/console/jmx/com.adobe.granite.workflow%3Atype%3DMaintenance/op/purgeActive/java.lang.String%2Cjava.lang.Integer%2Cjava.lang.Boolean", "model+%5Boptional%5D=&number+of+days+since+workflow+started=0&dry+run=")
  }

  def registerWorkflowStartedEventHandler() {
      workflowStartedEventHandler = new EventHandler() {
          int eventsReceived = 0
          def allEvents = []
          public void handleEvent(final Event event) {
              eventsReceived++
              allEvents.addAll(event)
          }
          public void reset() {
              eventsReceived = 0
              allEvents = []
          }
      }
      workflowStartedEventHandlerReg = script.bundleContext.registerService("org.osgi.service.event.EventHandler", workflowStartedEventHandler,
      new Hashtable([
          (EventConstants.EVENT_TOPIC): WorkflowEvent.EVENT_TOPIC,
          (EventConstants.EVENT_FILTER): "(EventType=" + WorkflowEvent.WORKFLOW_STARTED_EVENT + ")",
          ]))
      return workflowStartedEventHandler
  }

  def waitForWorkflowEvents(long minimumExpectedChanges, long maxWaitInSec = breakForDurationInSec) {
      boolean timeout = false

      int waitCounter = 0
      while(workflowStartedEventHandler && workflowStartedEventHandler.eventsReceived < minimumExpectedChanges) {
          if((lastTimestamp + maxWaitInSec * 1000) < System.currentTimeMillis()) {
              timeout = true
              break
          }
          Thread.sleep(50)
      }

      String eventHandlerResult = workflowStartedEventHandler ? " eventsReceived="+workflowStartedEventHandler.eventsReceived : ""
      logOpFinished("Waiting for workflow events", (timeout?'Timed out after ' + maxWaitInSec + 'sec ':'') + eventHandlerResult)

      workflowStartedEventHandler?.reset()
  }

  def waitForWorkflowsVisibleForMBean(long expectedWorkflows, long maxWaitInSec = breakForDurationInSec) {
      while(!performPostRequest("/system/console/jmx/com.adobe.granite.workflow%3Atype%3DMaintenance/op/listRunningWorkflowsPerModel/").contains("<td>${expectedWorkflows}</td>")) {
          if((lastTimestamp + maxWaitInSec * 1000) < System.currentTimeMillis()) {
              break
          }
          Thread.sleep(50)
      }
  }

  def performPostRequest(String requestPath, String queryString = "", String reqUser=user, String reqPassword=password) {

      def baseUrl = new URL(script.getBaseUrl()+requestPath)
      def encoded = Base64.getEncoder().encodeToString((reqUser+":"+reqPassword).getBytes(StandardCharsets.UTF_8));
      def connection = baseUrl.openConnection()
      connection.setRequestProperty("Authorization", "Basic "+encoded);

      connection.with {
        doOutput = true
        requestMethod = 'POST'
        outputStream.withWriter { writer ->
          writer << queryString
        }
      }

      def result = IOUtils.toString(connection.getInputStream())
      return result
  }

  def run(toRun, toRunAfterAll) {
    try {
      super.run(toRun)
    } finally {
      if(workflowStartedEventHandlerReg) {
          workflowStartedEventHandlerReg.unregister()
      }
      toRunAfterAll(this)
    }
  }
}

// DO NOT CHANGE BEYOND THIS POINT - master is https://gist.github.com/ghenzler/58ce57794ba9b5a60f2a2e60160be747 here

// --------------------------------------------------------------------
// Breaklet Base Utilities (to be copied at bottom of other breaklets)
// --------------------------------------------------------------------
import org.apache.jackrabbit.commons.JcrUtils
import org.apache.commons.lang3.StringUtils
import java.util.concurrent.ThreadLocalRandom
import javax.jcr.query.*
import org.apache.sling.jcr.api.*
import org.apache.sling.api.resource.observation.*
import java.util.Properties
import java.io.ByteArrayInputStream
import java.nio.charset.StandardCharsets
import org.osgi.service.event.EventConstants
import org.osgi.service.event.EventHandler
import org.osgi.service.event.Event
import org.apache.commons.io.IOUtils
import org.apache.sling.hc.util.FormattingResultLog

class BreakletBrokeException extends RuntimeException {
    String opName
    String timeSpent
    BreakletBrokeException(opName, timeSpent) {
        this.opName = opName
        this.timeSpent = timeSpent
    }
}
class BreakletPanicException extends RuntimeException { }

class Breaklet {

    int iterations
    int breakForDurationInSec
    String user
    String password

    long lastTimestamp = System.currentTimeMillis()
    def script
    long currentIteration
    String currentItLogMarker

    Map<String,List> timings = new LinkedHashMap<String,List>()

    Browser browser

    def resourceChangeListener = null
    def resourceChangeListenerReg = null
    def resourceChangeListenerPath = null

    def resourceChangeEventHandler = null
    def resourceChangeEventHandlerReg = null
    def resourceChangeEventHandlerPath = null

    Breaklet(def script, int breakForDurationInSec, String user=null, String password=null, int iterations=Integer.MAX_VALUE) {
        this.script = script
        this.breakForDurationInSec = breakForDurationInSec
        this.user = user
        this.password = password
        this.iterations = iterations

        this.browser = new Browser(script, breakForDurationInSec, user, password)
    }

    def logOpFinished(String opName, String opDescription="", def timeSpentMillis=null, def neverBreak=false) {
        script.logMsg(StringUtils.leftPad(currentItLogMarker, 7)+" " + StringUtils.rightPad(getTimeSpent(opName, timeSpentMillis as Long, neverBreak) + ":", 8) + " "+opName + " "+opDescription)
        currentItLogMarker = ""
    }

    def getTimeSpent(String opName, Long timeSpentMillis=null, def neverBreak=false) {
        long delta = timeSpentMillis ?: System.currentTimeMillis()-lastTimestamp

        def timingSeriesForOp = timings.get(opName)
        if(timingSeriesForOp==null) { timings.put(opName, (timingSeriesForOp = new ArrayList<Long>())) }
        while(timingSeriesForOp.size() < currentIteration-1) { timingSeriesForOp.add(timingSeriesForOp.size(), '') }
        timingSeriesForOp.add((int)currentIteration-1, delta)

        String timeSpent = FormattingResultLog.msHumanReadable(delta)
        if((delta/1000) > breakForDurationInSec && !neverBreak) {
            throw new BreakletBrokeException(opName, timeSpent)
        }

        lastTimestamp = System.currentTimeMillis()
        return timeSpent
    }

    def run(toRun) {
        File currentWorkingDir = new File(System.getProperty('user.dir'))
        File panicFile = new File(currentWorkingDir, 'panic')

        try {
            (1..iterations).each { itNo ->
                currentItLogMarker = itNo+":"

                def sessionForIteration = null
                try {
                    if(StringUtils.isNotBlank(user)) {
                        sessionForIteration = getSessionFor(user, password)
                    }
                    currentIteration = itNo
                    def sessionForIt = sessionForIteration ?: script.session
                    toRun(this, currentIteration, sessionForIt)

                    if(panicFile.exists()) {
                        panicFile.renameTo(new File(currentWorkingDir, 'no-panic'))
                        throw new BreakletPanicException()
                    }

                } finally {
                    if(sessionForIteration!=null && sessionForIteration.isLive()) {
                        sessionForIteration.logout()
                    }
                }
            }
            script.logMsg("\n\nBreaklet survived - "+iterations+" iterations finished without exceeding the threshold\n")
        } catch(BreakletBrokeException e) {
            script.logMsg("\n\nBreaklet broke at iteration "+currentIteration+": Operation '"+e.opName+"' "+e.timeSpent + " >= "+breakForDurationInSec + "sec\n")
        } catch(BreakletPanicException e) {
            script.logMsg("\nBreaklet broke due to panic after iteration "+currentIteration+"\n")
        } catch(Exception e) {
            script.logMsg("\n\nBreaklet broke at iteration "+currentIteration+" with unexpected exception: "+e)
            script.log.info("Exception:"+e, e)
        } finally {
            String result = ""
            timings.keySet().each { key ->
                result += ""+key+","+ StringUtils.join(timings.get(key),',') + "\n"
            }
            new File(currentWorkingDir, 'breaklet-result-'+new Date().getTime()+'.csv').text = result

            if(resourceChangeListenerReg) {
                resourceChangeListenerReg.unregister()
            }
            if(resourceChangeEventHandlerReg) {
                resourceChangeEventHandlerReg.unregister()
            }
        }

    }

    def runInBrowser(String puppeteerScript) {
        return browser.run(puppeteerScript)
    }

    def getSessionFor(sessionUser, sessionPassword) {
      def repository = script.session.getRepository()
      def session = repository.login(new SimpleCredentials(sessionUser, sessionPassword.toCharArray()));
      return session
    }

    def registerResourceChangeListener(paths) {
        resourceChangeListenerPath = paths
        resourceChangeListener = new ResourceChangeListener() {
            int changesReceived = 0
            int callsReceived = 0
            def allChanges = []
            public void onChange(List<ResourceChange> changes) {
                changesReceived += changes.size()
                callsReceived++
                allChanges.addAll(changes)
            }
            public void reset() {
                changesReceived = 0
                callsReceived = 0
                allChanges = []
            }
        }
        resourceChangeListenerReg = script.bundleContext.registerService("org.apache.sling.api.resource.observation.ResourceChangeListener", resourceChangeListener, new Hashtable([(ResourceChangeListener.PATHS): paths]))
        return resourceChangeListener
    }

    def registerResourceChangeEventHandler(path) {
        resourceChangeEventHandlerPath = path
        resourceChangeEventHandler = new EventHandler() {
            int eventsReceived = 0
            def allEvents = []
            public void handleEvent(final Event event) {
                eventsReceived++
                allEvents.addAll(event)
            }
            public void reset() {
                eventsReceived = 0
                allEvents = []
            }
        }
        resourceChangeEventHandlerReg = script.bundleContext.registerService("org.osgi.service.event.EventHandler", resourceChangeEventHandler,
        new Hashtable([
            (EventConstants.EVENT_TOPIC): "org/apache/sling/api/resource/Resource/*",
            (EventConstants.EVENT_FILTER): "("+SlingConstants.PROPERTY_PATH+"="+path+"*)",
            ]))
        return resourceChangeEventHandler
    }

    def waitForResourceChangedEvents(long minimumExpectedChanges, long maxWaitInSec=300, String opName=null) {
        boolean timeout = false

        int waitCounter = 0
        while((resourceChangeListener && resourceChangeListener.changesReceived < minimumExpectedChanges)
              || (resourceChangeEventHandler && resourceChangeEventHandler.eventsReceived < minimumExpectedChanges)
            ) {
            if((lastTimestamp + maxWaitInSec*1000) < System.currentTimeMillis()) {
                timeout = true
                break
            }
            Thread.sleep(50)
        }

        String resourceChangeListenerResult = resourceChangeListener ? " changesReceived="+resourceChangeListener.changesReceived+" (in "+resourceChangeListener.callsReceived +" calls) " : ""
        String resourceChangeEventHandlerResult = resourceChangeEventHandler ? " eventsReceived="+resourceChangeEventHandler.eventsReceived : ""
        opName = opName ?: "Wait for resource changed events ("+(resourceChangeListenerPath?:resourceChangeEventHandlerPath)+")"
        logOpFinished(opName, (timeout?'Timed out after '+maxWaitInSec+'sec ':'') + resourceChangeListenerResult + resourceChangeEventHandlerResult)

        resourceChangeListener?.reset()
        resourceChangeEventHandler?.reset()
    }

    def performRequest(String requestPath, String reqUser=user, String reqPassword=password) {
        def con = new URL(script.getBaseUrl()+requestPath).openConnection();
        String encoded = Base64.getEncoder().encodeToString((reqUser+":"+reqPassword).getBytes(StandardCharsets.UTF_8));
        con.setRequestProperty("Authorization", "Basic "+encoded);
        def result = IOUtils.toString(con.getInputStream());
        return result
    }

}

class Browser {

    // config
    def script
    String user
    String password
    int breakForDurationInSec

    // state
    File browserDir
    String testJsTemplate
    Properties resultProps

    Browser(def script, int breakForDurationInSec, String user, String password) {
        this.script = script
        this.breakForDurationInSec = breakForDurationInSec
        this.user = user
        this.password = password
    }

    def init() {
        browserDir = new File(System.getProperty('user.dir')+'/puppeteer')
        if(!browserDir.exists()) {
            script.logMsg("init "+ browserDir)
            browserDir.mkdir()
            new File(browserDir, 'package.json').text = """
                {
                  "name": "puppeteer-breaklet",
                  "version": "1.0.0",
                  "scripts": {
                    "test": "node test.js"
                  },
                  "dependencies": {
                    "puppeteer": "1.7.0"
                  }
                }
                """.stripIndent()
            def result = script.exec("npm install", browserDir)
            script.logMsg("npm install result: "+ result)
        }
        testJsTemplate = '''
            const puppeteer = require('puppeteer');

            let baseUrl = process.argv[2]
            let user = process.argv[3]
            let password = process.argv[4]
            let startTime = process.argv[5]
            let defaultTimeout = 120*1000
            let screenshotCreated = false

            let loginPage = "/libs/granite/core/content/login.html"

            function sleep(ms) {
              return new Promise(resolve => setTimeout(resolve, ms));
            }

            function getCrxdeXPath(nodeName) {
                return "//*[@id='repository']//a[span/text()='"+nodeName+"']"
            }

            async function waitForXPath(page, xpath) {
                await page.waitForXPath(xpath, { timeout: defaultTimeout})
                let nodeElements = await page.$x(xpath)
                return nodeElements[0]
            }

            async function waitForCrxdeNodeWithName(page, nodeName) {
                let nodeElXPath = getCrxdeXPath(nodeName)
                return await waitForXPath(page, nodeElXPath)
            }

            async function waitForAndClick(page, selector, retry=true) {

                await page.waitForSelector(selector, { timeout: defaultTimeout, visible: true})

                if (await page.$(selector) === null) {
                    throw new Error("Could not find " + selector + " even after waiting for " + defaultTimeout);
                }

                try {
                    await page.click(selector)
                } catch(err) {
                    if(retry) {
                        console.log("Error " +err+" while clicking on selector "+selector+", retrying...")
                        await sleep(200)
                        waitForAndClick(page, selector, false);
                    } else {
                        throw err
                    }
                }
            }

            async function clickOnIconOfCrxdeNode(page, nodeHandle) {
                await page.evaluate(node => node.previousSibling.previousSibling.click(), nodeHandle);
            }

            async function createScreenshot(page, name=null) {
                if(page) {
                    await page.screenshot({path: 'run-'+startMillis+(name ? '-'+name : '')+'.png'});
                    screenshotCreated = true
                }
            }

            async function run() {
             var resultProps = {};
             const browser = await puppeteer.launch({headless: true, dumpio: true, args: ['--window-size=1440,900', "--no-sandbox"]});
             let page = null
             try {
                //const context = await browser.createIncognitoBrowserContext();
                page = await browser.newPage(); //  context or browser
                await page.setViewport({width: 1440, height: 900});

                await page.goto(baseUrl + loginPage);
                await page.$eval('#username', (el, pUsername) => el.value = pUsername, user);
                await page.$eval('#password', (el, pPassword) => el.value = pPassword, password);
                await sleep(500);
                await page.$eval('#login', form => form.submit());
                await page.waitForNavigation({ waitUntil: 'networkidle0'})
                if(new RegExp('.*?'+loginPage+'.*').test(await page.url())) {
                    throw new Error("Could not login!");
                } else {
                    console.log("Logged in user "+user)
                }

                startMillis = new Date().getTime()
                console.log("Starting user script at start milliseconds =", startMillis)

                %SCRIPT%

              } catch(err) {
                console.log("Error during execution: ", err)
              } finally {
                if(!screenshotCreated) {
                    await createScreenshot(page)
                }
                browser.close();
                let delta = new Date().getTime()-startMillis
                console.log("----------------")
                for (let key in resultProps) {
                    console.log(key+"="+resultProps[key])
                }
                console.log("browserDurationInMillis="+delta)
              }
            }
            run();
            '''.stripIndent()

    }

    def run(String scriptToRun) {
        init()
        new File(browserDir, 'test.js').text = testJsTemplate.replace('%SCRIPT%', scriptToRun)
        def baseUrl = script.getBaseUrl()
        def startTime = System.currentTimeMillis()
        def cmd = "npm test $baseUrl $user $password $startTime"
        def scriptResult = script.exec(cmd, browserDir, breakForDurationInSec)
        new File(browserDir, 'run-'+new Date().getTime()+".log").text = scriptResult
        if(scriptResult.contains("Error during execution:")) {
            throw new IllegalStateException("Browser execution failed")
        }
        def propsString = StringUtils.substringAfter(scriptResult, "----------------\n")
        resultProps = new Properties()
        resultProps.load(new ByteArrayInputStream(propsString.getBytes(StandardCharsets.UTF_8)))
        long browserDurationInMillis = resultProps.containsKey('breakForDurationInSec') ? Long.parseLong(resultProps.getProperty('breakForDurationInSec')) : (System.currentTimeMillis()-startTime)
        return browserDurationInMillis
    }

}

def logMsg(String msg) {
    log.info(msg)
    println(msg)
}

def getBaseUrl() {
    def aemPort = System.getProperty('sun.java.command').replaceFirst(".*?-p ([0-9]+) .*", "\$1")
    return "http://localhost:"+aemPort
}

import org.apache.commons.exec.CommandLine
import org.apache.commons.exec.DefaultExecutor
import org.apache.commons.exec.PumpStreamHandler
import org.apache.commons.exec.ExecuteWatchdog

def exec(String command, File dir, int breakForDurationInSec=600) {
    CommandLine cmdLine = CommandLine.parse(command);
    DefaultExecutor executor = new DefaultExecutor();
    executor.setWorkingDirectory(dir)

    ExecuteWatchdog watchdog = new ExecuteWatchdog(breakForDurationInSec * 1000);
    executor.setWatchdog(watchdog);

    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream);
    executor.setStreamHandler(streamHandler);
    executor.setExitValues([0,1,143] as int[])
    int exitValue = executor.execute(cmdLine);
    String scriptOutput = outputStream.toString()
    if(exitValue == 1) {
        logMsg("Script '"+command+"' returned exit code 1, output:\n "+ scriptOutput)
    }
    return scriptOutput
}