WIP Support HttpClient based scenarios

CHANGELOG.md

@@ -23,6 +23,9 @@ Dropping a requirement of a major version of a dependency is a new contract.
 ## [Unreleased]
 [Unreleased]: https://github.com/atlassian/virtual-users/compare/release-3.10.0...master
 
+### Added
+- Support HttpClient based scenarios.
+
 ## [3.10.0] - 2019-08-02
 [3.10.0]: https://github.com/atlassian/virtual-users/compare/release-3.9.1...release-3.10.0
 

src/main/java/com/atlassian/performance/tools/virtualusers/lib/api/Scenario.java

@@ -0,0 +1,42 @@
+package com.atlassian.performance.tools.virtualusers.lib.api;
+
+import com.atlassian.performance.tools.jiraactions.api.action.Action;
+import com.atlassian.performance.tools.jiraactions.api.measure.ActionMeter;
+import com.atlassian.performance.tools.virtualusers.api.config.VirtualUserTarget;
+
+import java.util.List;
+
+/**
+ * IT doesn't need to be thread safe. Each VU will have own copy of the scenario and own Thread
+ */
+public abstract class Scenario {
+
+    /**
+     * vu will use this constructor to create the scenario
+     *
+     * @param virtualUserTarget
+     * @param meter
+     */
+    protected Scenario(VirtualUserTarget virtualUserTarget, ActionMeter meter) {
+    }
+
+    /**
+     * The method will be called before VU starts executing actions
+     */
+    public void before() {
+
+    }
+
+    /**
+     * The method will be called Once to setUp product instance
+     */
+    public void setup() {
+
+    }
+
+    public abstract List<Action> getActions();
+
+    public void cleanUp() {
+
+    }
+}

src/main/kotlin/com/atlassian/performance/tools/virtualusers/LoadTest.kt

@@ -178,24 +178,23 @@ internal class LoadTest(
         userMemory: UserMemory,
         diagnostics: Diagnostics
     ): ExploratoryVirtualUser {
-        val scenarioAdapter = ScenarioAdapter(scenario)
         val maxOverallLoad = load.maxOverallLoad
         return ExploratoryVirtualUser(
             node = WebJiraNode(jira),
             nodeCounter = nodeCounter,
-            actions = scenarioAdapter.getActions(
-                jira = jira,
-                seededRandom = SeededRandom(random.random.nextLong()),
-                meter = meter
+            actions = scenario.getActions(
+                jira,
+                SeededRandom(random.random.nextLong()),
+                meter
             ),
-            setUpAction = scenarioAdapter.getSetupAction(
-                jira = jira,
-                meter = meter
+            setUpAction = scenario.getSetupAction(
+                jira,
+                meter
             ),
-            logInAction = scenarioAdapter.getLogInAction(
-                jira = jira,
-                meter = meter,
-                userMemory = userMemory
+            logInAction = scenario.getLogInAction(
+                jira,
+                meter,
+                userMemory
             ),
             maxLoad = maxOverallLoad / load.virtualUsers,
             diagnostics = diagnostics

src/main/kotlin/com/atlassian/performance/tools/virtualusers/NewExploratoryVirtualUser.kt

@@ -0,0 +1,58 @@
+package com.atlassian.performance.tools.virtualusers
+
+import com.atlassian.performance.tools.jiraactions.api.action.Action
+import com.atlassian.performance.tools.virtualusers.api.TemporalRate
+import com.atlassian.performance.tools.virtualusers.collections.CircularIterator
+import com.atlassian.performance.tools.virtualusers.measure.ApplicationNode
+import com.atlassian.performance.tools.virtualusers.measure.JiraNodeCounter
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import java.time.Duration
+import java.time.Instant.now
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * Applies load on a Jira via page objects. Explores the instance to learn about data and choose pages to visit.
+ * Wanders preset Jira pages with different proportions of each page. Their order is random.
+ */
+internal class NewExploratoryVirtualUser(
+    private val node: ApplicationNode,
+    private val nodeCounter: JiraNodeCounter,
+    private val actions: Iterable<Action>,
+    private val maxLoad: TemporalRate
+) {
+    private val logger: Logger = LogManager.getLogger(this::class.java)
+
+    /**
+     * Repeats [actions] until [done] is `true`.
+     */
+    fun applyLoad(
+        done: AtomicBoolean
+    ) {
+        logger.info("Applying load...")
+        nodeCounter.count(node)
+        val actionNames = actions.map { it.javaClass.simpleName }
+        logger.debug("Circling through $actionNames") // TODO Circling through [DiagnosableAction] :(
+        var actionsPerformed = 0.0
+        val start = now()
+        for (action in CircularIterator(actions)) {
+            if (done.get()) {
+                logger.info("Done applying load")
+                break
+            }
+            try {
+                action.run()
+                actionsPerformed++
+                val expectedTimeSoFar = maxLoad.scaleChange(actionsPerformed).time
+                val actualTimeSoFar = Duration.between(start, now())
+                val extraTime = expectedTimeSoFar - actualTimeSoFar
+                if (extraTime > Duration.ZERO) {
+                    Thread.sleep(extraTime.toMillis())
+                }
+            } catch (e: Exception) {
+                logger.error("Failed to run $action, but we keep running", e)
+            }
+        }
+    }
+
+}

src/main/kotlin/com/atlassian/performance/tools/virtualusers/NewLoadSegment.kt

@@ -0,0 +1,41 @@
+package com.atlassian.performance.tools.virtualusers
+
+import com.atlassian.performance.tools.virtualusers.lib.api.Scenario
+import org.apache.logging.log4j.LogManager
+import java.io.BufferedWriter
+import java.time.Duration
+import java.util.*
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+
+internal class NewLoadSegment(
+    val scenario: Scenario,
+    val output: BufferedWriter,
+    val done: AtomicBoolean,
+    val id: UUID,
+    val index: Int
+) : AutoCloseable {
+
+    override fun close() {
+        done.set(true)
+        output.close()
+        val executor = Executors.newSingleThreadExecutor {
+            Thread(it)
+                .apply { name = "close-driver" }
+                .apply { isDaemon = true }
+        }
+        try {
+            executor
+                .submit { scenario.cleanUp() }
+                .get(DRIVER_CLOSE_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)
+        } catch (e: Exception) {
+            LOGGER.warn("Failed to close WebDriver", e)
+        }
+    }
+
+    internal companion object {
+        private val LOGGER = LogManager.getLogger(this::class.java)
+        internal val DRIVER_CLOSE_TIMEOUT = Duration.ofSeconds(30)
+    }
+}

src/main/kotlin/com/atlassian/performance/tools/virtualusers/NewLoadTest.kt

@@ -0,0 +1,186 @@
+package com.atlassian.performance.tools.virtualusers
+
+import com.atlassian.performance.tools.io.api.ensureDirectory
+import com.atlassian.performance.tools.jiraactions.api.measure.ActionMeter
+import com.atlassian.performance.tools.jiraactions.api.measure.output.AppendableActionMetricOutput
+import com.atlassian.performance.tools.virtualusers.api.VirtualUserOptions
+import com.atlassian.performance.tools.virtualusers.api.config.VirtualUserTarget
+import com.atlassian.performance.tools.virtualusers.api.diagnostics.*
+import com.atlassian.performance.tools.virtualusers.lib.api.Scenario
+import com.atlassian.performance.tools.virtualusers.measure.ApplicationNode
+import com.atlassian.performance.tools.virtualusers.measure.JiraNodeCounter
+import com.google.common.util.concurrent.ThreadFactoryBuilder
+import org.apache.logging.log4j.CloseableThreadContext
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import org.openqa.selenium.WebDriver
+import org.openqa.selenium.remote.RemoteWebDriver
+import java.io.BufferedWriter
+import java.nio.file.Paths
+import java.time.Duration
+import java.util.*
+import java.util.concurrent.Executors
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * A [load test](https://en.wikipedia.org/wiki/Load_testing).
+ */
+internal class NewLoadTest(
+    private val options: VirtualUserOptions
+) {
+    private val logger: Logger = LogManager.getLogger(this::class.java)
+    private val behavior = options.behavior
+    private val target = options.target
+    private val workspace = Paths.get("test-results")
+    private val nodeCounter = JiraNodeCounter()
+    private val diagnosisPatience = DiagnosisPatience(Duration.ofSeconds(5))
+    private val diagnosisLimit = DiagnosisLimit(behavior.diagnosticsLimit)
+
+
+    private fun createScenario(virtualUserTarget: VirtualUserTarget, actionMeter: ActionMeter): Scenario {
+        return behavior
+            .scenario
+            .getConstructor(VirtualUserTarget::class.java, ActionMeter::class.java)
+            .newInstance(
+                virtualUserTarget,
+                actionMeter
+            ) as Scenario
+    }
+
+    private val load = behavior.load
+
+    fun run() {
+        logger.info("Holding for ${load.hold}.")
+        Thread.sleep(load.hold.toMillis())
+        workspace.toFile().ensureDirectory()
+        setUpJira()
+        applyLoad()
+        val nodesDump = workspace.resolve("nodes.csv")
+        nodesDump.toFile().bufferedWriter().use {
+            nodeCounter.dump(it)
+        }
+        logger.debug("Dumped node's counts to $nodesDump")
+    }
+
+    private fun setUpJira() {
+        CloseableThreadContext.push("setup").use {
+            val createScenario = createScenario(
+                virtualUserTarget = target,
+                actionMeter = ActionMeter(virtualUser = UUID.randomUUID())
+            )
+            createScenario.setup()
+            createScenario.cleanUp()
+        }
+    }
+
+    private fun applyLoad() {
+        val userCount = load.virtualUsers
+        val finish = load.ramp + load.flat
+        val loadPool = ThreadPoolExecutor(
+            userCount,
+            userCount,
+            0L,
+            TimeUnit.MILLISECONDS,
+            LinkedBlockingQueue<Runnable>(),
+            ThreadFactoryBuilder().setNameFormat("virtual-user-%d").setDaemon(true).build()
+        )
+        logger.info("Segmenting load across $userCount VUs")
+        val segments = (0..userCount).map { index ->
+            segmentLoad(index + 1)
+        }
+        logger.info("Load segmented")
+        segments.forEach { loadPool.submit { applyLoad(it) } }
+        Thread.sleep(finish.toMillis())
+        close(segments)
+    }
+
+    private fun segmentLoad(
+        index: Int
+    ): NewLoadSegment {
+        val uuid = UUID.randomUUID()
+        val output = output(uuid)
+        val scenario = createScenario(options.target,
+            ActionMeter(
+                virtualUser = uuid,
+                output = AppendableActionMetricOutput(output)
+            )
+        )
+
+        return NewLoadSegment(
+            scenario = scenario,
+            output = output,
+            done = AtomicBoolean(false),
+            id = uuid,
+            index = index
+        )
+    }
+
+    private fun output(uuid: UUID): BufferedWriter {
+        return workspace
+            .resolve(uuid.toString())
+            .toFile()
+            .ensureDirectory()
+            .resolve("action-metrics.jpt")
+            .bufferedWriter()
+    }
+
+    private fun applyLoad(
+        segment: NewLoadSegment
+    ) {
+        CloseableThreadContext.push("applying load #${segment.id}").use {
+            val rampUpWait = load.rampInterval.multipliedBy(segment.index.toLong())
+            logger.info("Waiting for $rampUpWait")
+            Thread.sleep(rampUpWait.toMillis())
+            val virtualUser = createVirtualUser(segment)
+            segment.scenario.before()
+            virtualUser.applyLoad(segment.done)
+        }
+    }
+
+    private fun createVirtualUser(
+        segment: NewLoadSegment
+    ): NewExploratoryVirtualUser {
+        val maxOverallLoad = load.maxOverallLoad
+        return NewExploratoryVirtualUser(
+            node = object : ApplicationNode {
+                override fun identify(): String = "todo??"
+            },
+            nodeCounter = nodeCounter,
+            actions = segment.scenario.actions,
+            maxLoad = maxOverallLoad / load.virtualUsers
+        )
+    }
+
+    private fun close(
+        segments: List<AutoCloseable>
+    ) {
+        logger.info("Closing segments")
+        val closePool = Executors.newCachedThreadPool { Thread(it, "close-segment") }
+        segments
+            .map { closePool.submit { it.close() } }
+            .forEach { it.get() }
+        logger.info("Segments closed")
+        closePool.shutdown()
+    }
+
+    private fun RemoteWebDriver.toDiagnosableDriver(): DiagnosableDriver {
+        return DiagnosableDriver(
+            this,
+            LimitedDiagnostics(
+                ImpatientDiagnostics(
+                    WebDriverDiagnostics(this),
+                    diagnosisPatience
+                ),
+                diagnosisLimit
+            )
+        )
+    }
+
+    internal data class DiagnosableDriver(
+        val driver: WebDriver,
+        val diagnostics: Diagnostics
+    )
+}