Graphite: Revised TcpSender #4202

vJoeyz · 2022-02-01T16:21:13Z

Contents
This pull request contains the implementation of a revised TcpSender used for sending statistics to Graphite. This revised version implements NACK-based write back-pressure with suspending, providing more reliability and data conservation when sending statistics.

Rationale
In the current version of the TcpSender, when a response isn't received within the WritePeriod (1 second by default) from Gatling's config (due to a slow db connection, resource problems etc.), the request is rejected, but the data is never buffered or resent. Furthermore, if this happens more than 5 times (as specified in the hardcoded maxRetries), the writing to Graphite halts completely, never to be resumed.

I have tested this implementation in my load tests and it fully resolves all issues I had with the current TcpSender when recording large amounts of data to a database.

Credits
This implementation was heavily inspired by Akka's sample implementation.

…4195 Motivation: In 3.7, we added some `Int => Expression[FiniteDuration]` implicits that in some cases take over the expected `Any => Expression[Any]`. Modification: Drop those implicits and add lots for overrides for during-like loops. Result: No more conflicts. Breaking binary change, can only be released in 3.8.

slandelle

Thanks fro contributing!

Could you please:

run compile to format your code and organize your imports
address comments
sign our CLA: https://docs.google.com/forms/d/1hbpxVqJ5hIYYJuOjOyOzub9fJFGFGvvfnB5w5LCWdUw/edit
rebase your branch on top of our main, so we don't have merge commits

Thanks!

slandelle · 2022-02-01T16:39:14Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+      case Event(GraphiteMetrics(bytes), data: ConnectedData) =>
+        buffer(bytes, data) match {
+          case Success(data) => stay() using data
+          case Failure(_) => goto(BufferOverflow) using NoData


slandelle · 2022-02-01T16:40:57Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+
+  private def writeFirst(data: ConnectedData): Unit = {
+    data.connection ! Write(data.storage(0), Ack(data.storageOffset))
+  }


single instruction => useless {}

slandelle · 2022-02-01T16:41:03Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+    for ((bytes, i) <- data.storage.zipWithIndex) {
+      data.connection ! Write(bytes, Ack(data.storageOffset + i))
+    }
+  }


single instruction => useless {}

slandelle · 2022-02-01T16:41:24Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+
+  private def acknowledge(ack: Int, data: ConnectedData): ConnectedData = {
+    require(ack == data.storageOffset, s"Received wrong ack $ack at ${data.storageOffset}")
+    require(data.storage.nonEmpty, s"Storage was empty at ack $ack")


What happens if this throws?

the actor will be restarted

slandelle · 2022-02-01T16:43:56Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

@@ -47,7 +55,14 @@ private[graphite] class TcpSender(
      unstashAll()
      val connection = sender()
      connection ! Register(self)
-      goto(Running) using ConnectedData(connection, failures)
+      goto(Running) using ConnectedData(connection, failures, 0, Vector.empty[ByteString], 0L, false, 0)


Vector.empty[ByteString] => Nil

slandelle · 2022-02-01T16:44:50Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSenderFSM.scala


 private[sender] sealed trait TcpSenderData
 private[sender] case object NoData extends TcpSenderData
 private[sender] final case class DisconnectedData(retry: Retry) extends TcpSenderData
-private[sender] final case class ConnectedData(connection: ActorRef, retry: Retry) extends TcpSenderData
+private[sender] final case class ConnectedData(connection: ActorRef, retry: Retry, storageOffset: Int, storage: Vector[ByteString], stored: Long, suspended: Boolean, nack: Int) extends TcpSenderData


Why Vector?

slandelle · 2022-02-01T16:46:20Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+      data.connection ! Write(bytes, Ack(currentOffset(data)))
+      buffer(bytes, data) match {
+        case Success(data) => stay() using data
+        case Failure(_) => goto(BufferOverflow) using NoData


slandelle · 2022-02-01T16:49:41Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

+      logger.info(s"Sending metrics to Graphite server located at: $remote")
+      data.connection ! Write(bytes, Ack(currentOffset(data)))
+      buffer(bytes, data) match {
+        case Success(data) => stay() using data


Rename so you're not shadowing the data reference.

slandelle · 2022-02-01T16:53:03Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

-      }
+    // Connection actor failed to send metric, log it as a failure
+    case Event(CommandFailed(Write(_, Ack(ack))), data: ConnectedData) =>
+      logger.info(s"Failed to write to Graphite server located at: $remote, buffering...")


Shouldn't it be a warning?

slandelle · 2022-02-01T16:55:59Z

gatling-graphite/src/main/scala/io/gatling/graphite/sender/TcpSender.scala

-        goto(WaitingForConnection) using DisconnectedData(newFailures)
+  when(Buffering)(event => {
+    var toAck = 10
+    var peerClosed = false


Those should be in ConnectedData.

…-writeback-pressure

slandelle · 2022-02-04T15:05:12Z

@vJoeyz Have you seen all the ^M chars in your last commits? eg https://github.com/gatling/gatling/pull/4202/files#diff-696b3ab9a3d6b03bfb45bb037cc20cb0ac1e81f9070f05116d63a99c5e796419R23

vJoeyz · 2022-02-04T15:17:56Z

@vJoeyz Have you seen all the ^M chars in your last commits? eg https://github.com/gatling/gatling/pull/4202/files#diff-696b3ab9a3d6b03bfb45bb037cc20cb0ac1e81f9070f05116d63a99c5e796419R23

They appeared after running compile

slandelle · 2022-02-04T15:21:39Z

They appeared after running compile

???
What's your OS? Windows?

vJoeyz · 2022-02-04T15:23:46Z

They appeared after running compile

??? What's your OS? Windows?

Yes.

slandelle · 2022-02-04T17:14:07Z

Do you use something like git config core.autocrlf?

vJoeyz · 2022-02-05T15:46:57Z

Do you use something like git config core.autocrlf?

Not that I'm aware of, no. Anyways, I've removed them. :)

slandelle · 2022-02-07T08:24:36Z

Anyways, I've removed them. :)

Yeah, but your code is not properly formatted. It would if you were using compile to trigger scalafmt. As is, your PR won't pass our CI. We're going to investigate our settings for Windows but I'm not sure the issue is there.

Also, you commit history is broken because it doesn't only have your commits rebased on top of our main branch.

slandelle · 2022-02-08T17:44:34Z

@vJoeyz We'll release 3.7.5 next week. By then, do you thing you'll be able to fix your commit history and your formatting?

slandelle · 2022-02-11T13:37:47Z

I tried to fix formatting and wartremover on your branch, but still, TcpSenderSpec fails:

[info] - should go to the Running state and send metrics if it could connect without issues *** FAILED ***
[info]   java.lang.AssertionError: assertion failed: expected Write(ByteString(102, 111, 111, 32, 49, 32, 49, 10),NoAck(null)), found Write(ByteString(102, 111, 111, 32, 49, 32, 49, 10),Ack(0))
[info]   at scala.Predef$.assert(Predef.scala:279)
[info]   at akka.testkit.TestKitBase.expectMsg_internal(TestKit.scala:462)
[info]   at akka.testkit.TestKitBase.expectMsg(TestKit.scala:438)
[info]   at akka.testkit.TestKitBase.expectMsg$(TestKit.scala:438)
[info]   at akka.testkit.TestKit.expectMsg(TestKit.scala:973)
[info]   at io.gatling.graphite.sender.TcpSenderSpec.$anonfun$new$3(TcpSenderSpec.scala:64)
[info]   at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
[info]   at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
[info]   at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
[info]   at org.scalatest.Transformer.apply(Transformer.scala:22)
[info]   ...
14:30:06.218 [pool-1-thread-1-ScalaTest-running-TcpSenderSpec] WARN io.gatling.graphite.sender.TcpSenderSpec$TcpSenderNoIo - Disconnected from Graphite server located at: 0.0.0.0/0.0.0.0:9999, retrying...
[info] - should retry to connected until the retry limit has been exceeded to finally stop *** FAILED ***
[info]   FlushingBuffer was not equal to Running (TcpSenderSpec.scala:80)

biski · 2022-09-05T10:03:09Z

I've tested this fix and it works in my case. Before fix integration with graphite used to stop after ~24h. Please merge it to stable, thanks :)

slandelle · 2022-09-05T10:17:13Z

@biski There's no way to merge this work as is:

code formatting violation everywhere, while one just has to run compile/Test:compile to format automatically
merge conflict

Contributions welcome

vJoeyz · 2022-10-12T09:21:52Z

@biski There's no way to merge this work as is:

code formatting violation everywhere, while one just has to run compile/Test:compile to format automatically

merge conflict

Contributions welcome

Sorry for my late reply, the past half year my health got in the way. I'll try to fix it up this week. :)

slandelle · 2022-10-12T10:34:07Z

@vJoeyz Oh, sorry for you! I hope you're doing well now.
No worries, take you time. I haven't closed to PR for inactivity because I believe it has value. I've tried to rebase myself but had limited time for it and didn't manage to succeed.

slandelle · 2023-01-03T09:18:07Z

Closing as idle for too long and can't be merged.

vJoeyz and others added 6 commits December 15, 2021 12:43

Do not trim scenario name

59886d0

Merge branch 'gatling:main' into main

e0def71

Fetched upstream

8781fb0

Merge branch 'gatling-main' into main

ac00298

Implemented back-pressuring writes for Graphite

a9305e8

vJoeyz changed the title ~~Tcp writeback pressure~~ Graphite: Revised TcpSender Feb 1, 2022

slandelle requested changes Feb 1, 2022

View reviewed changes

vJoeyz added 8 commits February 4, 2022 15:23

Processed PR feedback

c4e8339

Fixed test

f994fee

Compile

c188750

Implemented back-pressuring writes for Graphite

e6c6df2

Processed PR feedback

0811f4e

Fixed test

55fe910

Compile

743735d

Merge remote-tracking branch 'origin/tcp-writeback-pressure' into tcp…

6291159

…-writeback-pressure

vJoeyz requested a review from slandelle February 4, 2022 14:51

Removed ^M

54f35ae

slandelle force-pushed the main branch from ef95f12 to 3d4c99c Compare May 30, 2022 16:30

slandelle force-pushed the main branch 2 times, most recently from 6b3802f to e6bb51e Compare June 8, 2022 19:44

slandelle force-pushed the main branch 4 times, most recently from 84fab42 to 772f5a0 Compare June 14, 2022 16:22

slandelle force-pushed the main branch from bcceba1 to c63d009 Compare July 10, 2022 13:57

slandelle force-pushed the main branch from 85ad6f6 to a1e5013 Compare August 5, 2022 11:47

slandelle force-pushed the main branch from e9e3e32 to 0679e8c Compare August 25, 2022 15:40

slandelle force-pushed the main branch from 936f318 to 9c809ee Compare September 12, 2022 07:02

slandelle force-pushed the main branch 2 times, most recently from ba5492e to 91ec84b Compare September 22, 2022 20:25

slandelle closed this Jan 3, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Graphite: Revised TcpSender #4202

Graphite: Revised TcpSender #4202

vJoeyz commented Feb 1, 2022

slandelle left a comment •

edited

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

vJoeyz Feb 4, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle Feb 1, 2022

slandelle commented Feb 4, 2022

vJoeyz commented Feb 4, 2022

slandelle commented Feb 4, 2022

vJoeyz commented Feb 4, 2022

slandelle commented Feb 4, 2022

vJoeyz commented Feb 5, 2022

slandelle commented Feb 7, 2022

slandelle commented Feb 8, 2022

slandelle commented Feb 11, 2022 •

edited

biski commented Sep 5, 2022

slandelle commented Sep 5, 2022

vJoeyz commented Oct 12, 2022 •

edited

slandelle commented Oct 12, 2022

slandelle commented Jan 3, 2023

Graphite: Revised TcpSender #4202

Graphite: Revised TcpSender #4202

Conversation

vJoeyz commented Feb 1, 2022

slandelle left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

slandelle commented Feb 4, 2022

vJoeyz commented Feb 4, 2022

slandelle commented Feb 4, 2022

vJoeyz commented Feb 4, 2022

slandelle commented Feb 4, 2022

vJoeyz commented Feb 5, 2022

slandelle commented Feb 7, 2022

slandelle commented Feb 8, 2022

slandelle commented Feb 11, 2022 • edited

biski commented Sep 5, 2022

slandelle commented Sep 5, 2022

vJoeyz commented Oct 12, 2022 • edited

slandelle commented Oct 12, 2022

slandelle commented Jan 3, 2023

slandelle left a comment •

edited

slandelle commented Feb 11, 2022 •

edited

vJoeyz commented Oct 12, 2022 •

edited