在加特林场景中产生多个进一步动作的动作

Question

背景

我目前正在使用 gatling 开发压力测试工具的能力分析集。

其中一部分涉及使用滚动查询加载弹性搜索，然后调用更新 API。

我想达到的目标

第 1 步： 运行滚动启动器并保存 _scroll_id 以供进一步滚动查询使用

第 2 步： 运行重复滚动查询，作为每个滚动查询的一部分，对返回的每个命中进行修改并将其索引回 elasticsearch，有效从一个滚动查询操作中生成最多 1000 个操作，并对结果进行采样。

第 1 步很简单。第 2 步没那么多。

我试过的

我目前正在尝试通过解析 JSON 格式的结果的 ResponseTransformer 来实现这一点，对每个结果进行修改并为每个尝试另一个 exec(http(...).post(...) etc) 索引的线程触发一个线程更改回 elasticsearch。

基本上，我想我的处理方式是错误的。索引线程永远不会运行，更不用说 gatling 采样了。

这是我的滚动查询操作的主体：

  ...

  val pool = Executors.newFixedThreadPool(parallelism)

  val query = exec(http("Scroll Query")
    .get(s"/_search/scroll")
    .body(ElFileBody("queries/scrollquery.json")).asJSON // Do the scroll query
    .check(jsonPath("$._scroll_id").saveAs("scroll_id")) // Get the scroll ID from the response
    .transformResponse { case response if response.isReceived =>
      new ResponseWrapper(response) {
        val responseJson = JSON.parseFull(response.body.string)
        // Get the hits and
        val hits = responseJson.get.asInstanceOf[Map[String, Any]]("hits").asInstanceOf[Map[String,Any]]("hits").asInstanceOf[List[Map[String, Any]]]
        for (hit <- hits) {
          val id = hit.get("_id").get.asInstanceOf[String]
          val immutableSource = hit.get("_source").get.asInstanceOf[Map[String, Any]]
          val source = collection.mutable.Map(immutableSource.toSeq: _*) // Make the map mutable
          source("newfield") = "testvalue" // Make a modification
          Thread.sleep(pause) // Pause to simulate topology throughput
          pool.execute(new DocumentIndexer(index, doctype, id, source)) // Create a new thread that executes the index request
        }
      }
    }) // Make some mods and re-index into elasticsearch

  ...

DocumentIndexer 看起来像这样：

class DocumentIndexer(index: String, doctype: String, id: String, source: scala.collection.mutable.Map[String, Any]) extends Runnable {

  ...

  val httpConf = http
    .baseURL(s"http://$host:$port/${index}/${doctype}/${id}")
    .acceptHeader("application/json")
    .doNotTrackHeader("1")
    .disableWarmUp

  override def run() {

    val json = new ObjectMapper().writeValueAsString(source)

    exec(http(s"Index ${id}")
      .post("/_update")
      .body(StringBody(json)).asJSON)

  }

}

问题

这甚至可以使用加特林机吗？
我怎样才能达到我想达到的目的？

感谢任何help/suggestions！

Answer 1

可以通过使用 jsonPath 提取 JSON 命中数组并将元素保存到会话中，然后在操作链中使用 foreach 和 exec 来实现此目的- 在循环中执行索引任务，您可以相应地执行索引。

即：
滚动查询

...
  val query = exec(http("Scroll Query")
    .get(s"/_search/scroll")
    .body(ElFileBody("queries/scrollquery.json")).asJSON // Do the scroll query
    .check(jsonPath("$._scroll_id").saveAs("scroll_id")) // Get the scroll ID from the response
    .check(jsonPath("$.hits.hits[*]").ofType[Map[String,Any]].findAll.saveAs("hitsJson")) // Save a List of hit Maps into the session
  )
...

模拟

...
    val scrollQueries = scenario("Enrichment Topologies").exec(ScrollQueryInitiator.query, repeat(numberOfPagesToScrollThrough, "scrollQueryCounter"){
        exec(ScrollQuery.query, pause(10 seconds).foreach("${hitsJson}", "hit"){ exec(HitProcessor.query) })
    })
...

HitProcessor

...
  def getBody(session: Session): String = {
    val hit = session("hit").as[Map[String,Any]]
    val id = hit("_id").asInstanceOf[String]
    val source = mapAsScalaMap(hit("_source").asInstanceOf[java.util.LinkedHashMap[String,Any]])
    source.put("newfield", "testvalue")
    val sourceJson = new ObjectMapper().writeValueAsString(mapAsJavaMap(source))
    val json = s"""{"doc":${sourceJson}}"""
    json
  }

  def getId(session: Session): String = {
    val hit = session("hit").as[Map[String,Any]]
    val id = URLEncoder.encode(hit("_id").asInstanceOf[String], "UTF-8")
    val uri = s"/${index}/${doctype}/${id}/_update"
    uri
  }

  val query = exec(http(s"Index Item")
    .post(session => getId(session))
    .body(StringBody(session => getBody(session))).asJSON)
...

免责声明：此代码仍需优化！而且我实际上还没有学到很多 scala。欢迎提出更好的解决方案

完成这些之后，我现在真正想要实现的是并行化给定数量的索引任务。即：我得到 1000 次点击，我想为每个单独的点击执行一个索引任务，而不是只是迭代它们并一个接一个地执行它们，我想同时执行 10 个。

但是，我认为这是一个单独的问题，真的，所以我会这样提出。

在加特林场景中产生多个进一步动作的动作

Action to spawn multiple further actions in Gatling scenario

gatling

stress-testing

背景

我想达到的目标

我试过的

问题