maiha / gatling-amqp Goto Github PK

View Code? Open in Web Editor NEW

14.0 8.0 26.0 160 KB

Gatling AMQP support

License: MIT License

Scala 97.23% Shell 2.77%

gatling-amqp's Introduction

Introduction

Gatling AMQP support

CAUTION: This is not official library!
- but using 'io.gatling.amqp' for FQCN to deal with 'private[gatling]', sorry.
inspired by https://github.com/fhalim/gatling-rabbitmq (thanks!)

Usage

handy cli (use AmqpProtocol as console utility) [0.6 feature]

scala> import io.gatling.amqp.Predef._
scala> amqp.declare(queue("q3", durable = true)).run

publish (normal)

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    // .vhost("/")
    .auth("guest", "guest")
    .poolSize(10)

  val req = PublishRequest("q1", payload = "{foo:1}")

  val scn = scenario("AMQP Publish").repeat(1000) {
    exec(amqp("Publish").publish(req))
  }

  setUp(scn.inject(rampUsers(10) over (1 seconds))).protocols(amqpProtocol)

publish (with persistent)

PublishRequest.persistent make request DeliveryMode(2)
known issue: persistent reset request's properties

  val req = PublishRequest("q1", payload = "{foo:1}").persistent

publish (with confirmation)

set "confirmMode()" in protocol that invokes "channel.confirmSelect()"

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .poolSize(10)
    .confirmMode()

  val req = PublishRequest("q1", payload = "{foo:1}")

  val scn = scenario("AMQP Publish(ack)").repeat(1000) {
    exec(amqp("Publish").publish(req))
  }

  setUp(scn.inject(rampUsers(10) over (1 seconds))).protocols(amqpProtocol)

declare queues

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .declare(queue("q1", durable = true, autoDelete = false))

declare exchange and binding

  val x = exchange("color", "direct", autoDelete = false)
  val q = queue("orange")
  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("localhost")
    .port(5672)
    .auth("guest", "guest")
    .declare(x)
    .declare(q)
    .bind(x, q, routingKey = "orange")

full code: src/test/scala/io/gatling/amqp/PublishingSimulation.scala

consume (auto acked)

  implicit val amqpProtocol: AmqpProtocol = amqp
    .host("amqp")
    .port(5672)
    .auth("guest", "guest")

  val scn = scenario("AMQP Publish(ack)").exec {
    amqp("Consume").consume("q1", autoAck = true)
  }

  setUp(scn.inject(atOnceUsers(1))).protocols(amqpProtocol)

full code: src/test/scala/io/gatling/amqp/ConsumingSimulation.scala

consume (manual acked)

not implemented yet

Run

sbt directly

% sbt
> testOnly io.gatling.amqp.PublishingSimulation

% sbt
> testOnly io.gatling.amqp.ConsumingSimulation

try sbt -J-Xmx8192m -J-XX:MaxPermSize=256m for publishing massive messages

shell script to store gatling stdout logs and simulation sources

% ./run p foo

stored in "stats/p/foo"

Benchmark

environments

server

cpu: Xeon X5687(3.60GHz)
mem: 24GB, limit(10GiB), watermark(5GiB)

rabbitmq

version: 3.5.2
total bytes = servers * payalod * messages * users
users = concurrency of AMQP connections

publish (persistent queue)

servers	payload	ack	repeat	users	total	sec	qps	spd
1	1 KB	o	100,000	10	1 GB	69	14326	14.3 MB/s
1	10 KB	o	10,000	10	1 GB	14	6778	67.8 MB/s
1	100 KB	o	1,000	10	1 GB	11	881	88.1 MB/s
1	1 MB	o	100	10	1 GB	10	97	97.8 MB/s
1	10 MB	o	100	1	1 GB	-	-	log error
1	10 KB	o	1,000	100	1 GB	17	5791	57.9 MB/s
4	1 KB	o	100,000	10	4 GB	298	13490	13.5 MB/s
4	10 KB	o	10,000	10	4 GB	56	7208	72.1 MB/s

log error: statsEngine stopped before working actors finished

publish (persistent queue, paging)

servers	payload	ack	repeat	users	total	sec	qps	spd
1	10 KB	o	100,000	10	10 GB	143	6983	69.8 MB/s
1	10 KB	o	200,000	10	20 GB	301	6632	66.3 MB/s

consume (persistent queue)

payload	message	users	total	sec	qps	spd
10 KB	100 k	1	1 GB	12	8436	84.4 MB/s
10 KB	2 m	1	20 GB	179	11161	111.6 MB/s

publish and consume (persistent queue)

payload	p-ack	repeat	publisher	total	qps	spd	consumer	ack	qps
10 KB	o	10,000	10	1 GB	6779	67.8 MB/s	1	auto	6233
10 KB	o	200,000	10	20 GB	7639	76.4 MB/s	1	auto	7622

Library

amqp-client-3.5.3
gatling-sbt-2.1.6 (to implement easily)
gatling-2.2.0-M3 (live with edge)

TODO

declare exchanges, queues and bindings in action builder context (to test declaration costs)
make AmqpProtocol immutable
make Builder mutable
mandatory
consume action (manual ack)

License

released under the MIT License.

gatling-amqp's People

Contributors

Stargazers

Watchers

Forkers

xuwei-k swachter frascuchon fred84 lubovarga elufimov maomaofeitui roarking jeantil mattsaxon sygour eric-gmendes ariden83 maxwo sandzii dieselr elenglart ajclaros82

gatling-amqp's Issues

Could you say how to integrate your plugin in

Could you say how to integrate your plugin "io.gatling.amqp.Predef._" in gatling, I need to do it in a dockerfile

see default dockerfile https://github.com/denvazh/gatling/edit/master/2.3.1/Dockerfile

thx

Passing custom headers

Hi, Can someone provide an example how can we pass custom headers using this plugin ?

I can't find a parameter to pass for this method PublishRequest

Simulation finished, gatling actor not stopped

Hi,

Very often I get Simulation finished but the gatling actor is not stopped. Below is one of the outputs

[INFO] [GatlingSystem-akka.actor.default-dispatcher-16] [akka://GatlingSystem/user/io.gatling.core.result.writer.FileDataWriter] Message [io.gatling.core.result.writer.UserMessage] from Actor[akka://GatlingSystem/deadLetters] to Actor[akka://GatlingSystem/user/io.gatling.core.result.writer.FileDataWriter#-1654396042] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

[GatlingSystem-akka.actor.default-dispatcher-20] [akka://GatlingSystem/user/throttler] Message [io.gatling.core.controller.throttle.OneSecondTick$] from Actor[akka://GatlingSystem/user/throttler#-346115863] to Actor[akka://GatlingSystem/user/throttler#-346115863] was not delivered. [8] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

An example of a scenario that leads to this problem.

 val messageExchange = exchange("EventExchange", "direct", autoDelete = false)
 val messageQueue = queue("message_queue", durable = true, autoDelete = false)

 implicit val amqpProtocol: AmqpProtocol = amqp
    .host(rabbitMqHost)
    .port(rabbitMqPort)
    .auth(rabbitMqUser, rabbitMqPassword)
    .declare(messageQueue)
    .bind(messageExchange, messageQueue, routingKey = "created")

  val scn = scenario("Deliver Message").exec(amqp("Consume").consume("message_queue", autoAck = true))
  setUp(scn.inject(rampUsers(50) over (5 seconds)).protocols(amqpProtocol)).maxDuration(30 seconds)

I run the test in a maven build using the gatling-maven-plugin. When running scenarios that use let's say http protocol it finishes successfully.

Do you have an idea why is this happening?

Thanks.

Open Source License

Could you please make your code available under an open source license (Hint: http://oss-watch.ac.uk/resources/opensourceyourcode)

Access session variables in AmqpRequestBuilder

E.g. in order to put something on the session in a "feeder" loop and access it when using amqp("..").publish():

val scn = scenario("Create via AMQP")
    .exec(
      amqp("Consume response").consume("response_queue", autoAck = true)
    )
    .repeat(noOfRepeats) {
      feed(feeder)
        .exec(session => {
          val command = commandProducer.getCommand().toJsonString().toCharArray.map(_.toByte)

          session.set("publishRequest",
            new PublishRequest(Exchange.apply("commands", "direct"), "create", basicProperties, command))
        })
        .exec(
          amqp("Create").publish("${publishRequest}") // somehow access session variable
        )
    }

How awesome would that be? :D

Is there a way to define just one consumer for a scenario which spans more than one user? I have a scenario which publishes messages concurently and would like to have just one consumer to see its throughput.

birynk

incompatibility with gatling-2.2.2

With latest version of gatling-2.2.2 Unable to use this plugin...
I have gatling-amqp_2.11:0.5 in project dependencies...

When run performance test getting the below error.
Missing or invalid dependency detected while loading class file 'AmqpProtocol.class'.
Could not access type Protocol in package io.gatling.core.config,
because it (or its dependencies) are missing. Check your build definition for
missing or conflicting dependencies. (Re-run with -Ylog-classpath to see the problematic classpath.)
A full rebuild may help if 'AmqpProtocol.class' was compiled against an incompatible version of io.gatling.core.config.
one error found

Can you help me on this one to work on gatling-2.2.2

Maven release

Your project looks very promising and we would like to use it in our performance tests, especially the part of consuming messages.

Is it possible to create a maven release in maven central?

Thx.