AsyncQueryConstantAcs scenario (digital-asset#7054)

changelog_begin changelog_end
conr2d · Aug 7, 2020 · ba22f5b · ba22f5b
1 parent 8b089f5
commit ba22f5b
Show file tree

Hide file tree

Showing 5 changed files with 143 additions and 36 deletions.
diff --git a/...ice/http-json-perf/src/main/scala/com/daml/http/perf/scenario/AsyncQueryConstantAcs.scala b/...ice/http-json-perf/src/main/scala/com/daml/http/perf/scenario/AsyncQueryConstantAcs.scala
@@ -0,0 +1,60 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.http.perf.scenario
+
+import io.gatling.core.Predef._
+import io.gatling.http.Predef._
+import io.gatling.http.check.ws.WsTextFrameCheck
+
+import scala.concurrent.duration._
+
+@SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements"))
+class AsyncQueryConstantAcs
+    extends Simulation
+    with SimulationConfig
+    with HasRandomAmount
+    with HasCreateRequest {
+
+  private val wantedAcsSize = 5000
+
+  private val waitForResponse: FiniteDuration = 5.seconds
+
+  private val numberOfRuns = 100
+
+  private val queryRequest =
+    """{"templateIds": ["Iou:Iou"], "query": {"amount": {"%gt": 1.0}}}"""
+
+  val messageCheck: WsTextFrameCheck = ws
+    .checkTextMessage("messageCheck")
+    .check(jsonPath("$.offset").find.notExists)
+    .check(jsonPath("$.events[*].created").findAll)
+
+  private def query(runId: Int) = {
+    val wsName = s"websocket$runId"
+    ws("Connect websocket", wsName)
+      .connect("/v1/stream/query")
+      .subprotocol(s"jwt.token.$aliceJwt, daml.ws.auth")
+      .onConnected(
+        exec(
+          ws(s"Send Query Request and wait for response: $waitForResponse", wsName)
+            .sendText(queryRequest)
+            .await(waitForResponse)(Vector.fill(5)(messageCheck): _*)
+        )
+      )
+  }
+
+  private val asyncQueryScenario = scenario(s"AsyncQueryConstantAcs, numberOfRuns: $numberOfRuns")
+    .doWhile(_ => acsQueue.size() < wantedAcsSize) {
+      pause(1.second)
+    }
+    .exec(
+      1.to(numberOfRuns).map(runId => exec(query(runId)))
+    )
+
+  setUp(
+    fillAcsScenario(wantedAcsSize).inject(atOnceUsers(1)),
+    asyncQueryScenario.inject(atOnceUsers(1)),
+  ).protocols(httpProtocol)
+
+}
diff --git a/ledger-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/Checks.scala b/ledger-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/Checks.scala
@@ -0,0 +1,28 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.http.perf.scenario
+
+import io.gatling.commons.validation
+import io.gatling.core.Predef._
+import io.gatling.core.check.{Check, CheckResult}
+
+object Checks {
+
+  def printResponseCheck[R]: Check[R] = new PrintResponseCheck[R]
+
+  /**
+    * Useful for debugging.
+    *
+    * @tparam R response type
+    */
+  private class PrintResponseCheck[R] extends Check[R] {
+    override def check(
+        response: R,
+        session: Session,
+        preparedCache: java.util.Map[Any, Any]): validation.Validation[CheckResult] = {
+      println(s"Response: $response")
+      validation.Success(CheckResult(None, None))
+    }
+  }
+}
diff --git a/...-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/HasCreateRequest.scala b/...-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/HasCreateRequest.scala
@@ -0,0 +1,42 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.http.perf.scenario
+
+import java.util.concurrent.{BlockingQueue, LinkedBlockingQueue}
+
+import io.gatling.core.Predef._
+import io.gatling.core.structure.ScenarioBuilder
+import io.gatling.http.Predef._
+import io.gatling.http.request.builder.HttpRequestBuilder
+
+private[scenario] trait HasCreateRequest {
+  this: HasRandomAmount =>
+
+  lazy val acsQueue: BlockingQueue[String] = new LinkedBlockingQueue[String]()
+
+  lazy val createRequestAndCollectContractId: HttpRequestBuilder =
+    http("CreateCommand")
+      .post("/v1/create")
+      .body(StringBody("""{
+  "templateId": "Iou:Iou",
+  "payload": {
+    "issuer": "Alice",
+    "owner": "Alice",
+    "currency": "USD",
+    "amount": "${amount}",
+    "observers": []
+  }
+}"""))
+      .check(
+        status.is(200),
+        jsonPath("$.result.contractId").transform(x => acsQueue.put(x))
+      )
+
+  def fillAcsScenario(size: Int): ScenarioBuilder =
+    scenario(s"FillAcsScenario, size: $size")
+      .doWhile(_ => acsQueue.size() < size) {
+        feed(Iterator.continually(Map("amount" -> randomAmount())))
+          .exec(createRequestAndCollectContractId.silent)
+      }
+}
diff --git a/...-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/SimulationConfig.scala b/...-service/http-json-perf/src/main/scala/com/daml/http/perf/scenario/SimulationConfig.scala
@@ -8,20 +8,19 @@ import io.gatling.http.protocol.HttpProtocolBuilder
 
 private[scenario] trait SimulationConfig {
   lazy val httpProtocol: HttpProtocolBuilder = http
-    .baseUrl(baseUrl)
-    .warmUp(warmupUrl)
+    .baseUrl(s"http://$hostAndPort")
+    .wsBaseUrl(s"ws://$hostAndPort")
+    .warmUp(s"http://$hostAndPort/v1/query")
     .inferHtmlResources()
     .acceptHeader("*/*")
     .acceptEncodingHeader("gzip, deflate")
     .authorizationHeader(s"Bearer $aliceJwt")
     .contentTypeHeader("application/json")
 
-  private val baseUrl: String = "http://localhost:7575"
-
-  private val warmupUrl: String = "http://localhost:7575/v1/query"
+  private val hostAndPort = "localhost:7575"
 
   // {"https://daml.com/ledger-api": {"ledgerId": "MyLedger", "applicationId": "foobar", "actAs": ["Alice"]}}
-  private val aliceJwt: String =
+  val aliceJwt: String =
     "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJodHRwczovL2RhbWwuY29tL2xlZGdlci1hcGkiOnsibGVkZ2VySWQiOiJNeUxlZGdlciIsImFwcGxpY2F0aW9uSWQiOiJmb29iYXIiLCJhY3RBcyI6WyJBbGljZSJdfX0.VdDI96mw5hrfM5ZNxLyetSVwcD7XtLT4dIdHIOa9lcU"
 
 }
diff --git a/...vice/http-json-perf/src/main/scala/com/daml/http/perf/scenario/SyncQueryVariableAcs.scala b/...vice/http-json-perf/src/main/scala/com/daml/http/perf/scenario/SyncQueryVariableAcs.scala
@@ -1,36 +1,19 @@
 // Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
-package com.daml.http.perf.scenario
 
-import java.{util => jutil}
+package com.daml.http.perf.scenario
 
 import io.gatling.core.Predef._
 import io.gatling.http.Predef._
 
 import scala.concurrent.duration._
 
 @SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements"))
-class SyncQueryVariableAcs extends Simulation with SimulationConfig with HasRandomAmount {
-
-  private val acsQueue = new jutil.concurrent.LinkedBlockingQueue[String]()
-
-  private val createRequest =
-    http("CreateCommand")
-      .post("/v1/create")
-      .body(StringBody("""{
-  "templateId": "Iou:Iou",
-  "payload": {
-    "issuer": "Alice",
-    "owner": "Alice",
-    "currency": "USD",
-    "amount": "${amount}",
-    "observers": []
-  }
-}"""))
-      .check(
-        status.is(200),
-        jsonPath("$.result.contractId").transform(x => acsQueue.put(x))
-      )
+class SyncQueryVariableAcs
+    extends Simulation
+    with SimulationConfig
+    with HasRandomAmount
+    with HasCreateRequest {
 
   private val queryRequest =
     http("SyncQueryRequest")
@@ -54,11 +37,6 @@ class SyncQueryVariableAcs extends Simulation with SimulationConfig with HasRand
 
   private val numberOfRuns = 500
 
-  private val fillAcs = scenario(s"FillAcsScenario, size: $wantedAcsSize")
-    .doWhile(_ => acsQueue.size() < wantedAcsSize) {
-      feed(Iterator.continually(Map("amount" -> randomAmount()))).exec(createRequest.silent)
-    }
-
   private val syncQueryScn =
     scenario(s"SyncQueryWithVariableAcsScenario, numberOfRuns: $numberOfRuns")
       .doWhile(_ => acsQueue.size() < wantedAcsSize) {
@@ -76,13 +54,13 @@ class SyncQueryVariableAcs extends Simulation with SimulationConfig with HasRand
           // run query in parallel with archive and create
           queryRequest.notSilent.resources(
             archiveRequest.silent,
-            createRequest.silent
+            createRequestAndCollectContractId.silent
           )
         }
       }
 
   setUp(
-    fillAcs.inject(atOnceUsers(1)),
+    fillAcsScenario(wantedAcsSize).inject(atOnceUsers(1)),
     syncQueryScn.inject(atOnceUsers(1)),
   ).protocols(httpProtocol)
 }