Monitoring (Akka) Streams

Ok, you got your Akka Streams pipeline finally deployed to production. But how are you monitoring it? When productionising Reactive Streams, the same backpressure that preserves the safety of your pipeline can get in the way of effectively monitoring its status. In this talk we’ll present a line of action to get a pulse on your Akka Streams app, to better understand its throughput and bottlenecks.

Transcript

1. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. Scala in the City - July 2018
2. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. MONITORING (AKKA) STREAMS
3. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. Stefano Bonetti Software Engineer @svezfaz @svez_faz
4. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. FIRST-TO-KNOW 4
5. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT BACKPRESSURE
6. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT BACKPRESSURE
7. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT (/s) 7 Web Service req res
8. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 8 CHALLENGE 1 - TOPOLOGY
9. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 9 CHALLENGE 1 - TOPOLOGY © Kevin Cappis 2004
10. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. CHALLENGE 2 - LOWER I/O CORRELATION 10 web service
11. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. CHALLENGE 2 - LOWER I/O CORRELATION 11 Processor[T, S]
12. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 12 © Kevin Cappis 2004
13. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. TRANSPARENT PROCESSORS 13 Processor[T, S] Processor[T, S] Publisher[T] Subscriber[S]
14. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. TRANSPARENT PROCESSORS 14 Processor[T, S] Processor[T, S] Publisher[T] Subscriber[S] c1 c2 c3 c4 Processor[T, T] Processor[S, S]
15. val pangram = Source(List("The", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")) val toChars = Flow[String].mapConcat(_.toList) val print = Sink.foreach(println) pangram .via(toChars) .runWith(print) AKKA STREAMS - THROUGHPUT
16. val pangram = Source(List("The", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")) val toChars = Flow[String].mapConcat(_.toList) val print = Sink.foreach(println) pangram .via(toChars) .runWith(print) AKKA STREAMS - THROUGHPUT
17. val pangram = Source(List("The", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")) val toChars = Flow[String].mapConcat(_.toList) val print = Sink.foreach(println) def meter[T](name: String): Flow[T, T, NotUsed] = { val msgCounter = Kamon.metrics.counter(name) Flow[T].map { x ⇒ msgCounter.increment(); x } } pangram .via(meter("produced")) .via(toChars) .via(meter("processed")) .runWith(print) AKKA STREAMS - THROUGHPUT
18. val pangram = Source(List("The", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")) val toChars = Flow[String].mapConcat(_.toList) val print = Sink.foreach(println) def meter[T](name: String): Flow[T, T, NotUsed] = { val msgCounter = Kamon.metrics.counter(name) Flow[T].map { x ⇒ msgCounter.increment(); x } } pangram .via(meter("produced")) .via(toChars) .via(meter("processed")) .runWith(print) AKKA STREAMS - THROUGHPUT
19. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT BACKPRESSURE
20. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT BACKPRESSURE
21. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. EXAMPLE 21 Validation/ Filtering Commit Kafka offset valid invalid Store via API
22. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. Store via API EXAMPLE 22 Validation/ Filtering Commit Kafka offset valid invalid 100/s 75/s 25/s 75/s
23. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. Store via API EXAMPLE 23 Validation/ Filtering Commit Kafka offset valid invalid 100/s 75/s 25/s 75/s 20000/s lag
24. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 24
25. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. REACTIVE STREAMS 101 25 Processor Publisher Subscriber request request emit emit
26. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. REACTIVE STREAMS 101 26 Processor Publisher Subscriber request request emit emit
27. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. REACTIVE STREAMS 101 27 Processor Publisher Subscriber request request emit emit
28. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. REACTIVE STREAMS 101 28 Processor Publisher Subscriber request request emit emit
29. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. REACTIVE STREAMS 101 29 Processor Publisher Subscriber request request emit emit
30. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. BACKPRESSURE 30 Processor Publisher Subscriber request ? request ? emit emit
31. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. NO BACKPRESSURE 31 Processor Publisher Subscriber request emit ? emit ? request
32. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. BACKPRESSURE 32 request emit request request emit Processor Publisher Subscriber request request emit emit
33. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 33 AKKA STREAMS - GRAPHSTAGE API
34. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
35. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
36. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
37. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
38. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
39. final case class Map[In, Out](f: In ⇒ Out) extends GraphStage[FlowShape[In, Out]] { val in = Inlet[In]("Map.in") val out = Outlet[Out]("Map.out") override val shape = FlowShape(in, out) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { override def onPush(): Unit = push(out, f(grab(in))) override def onPull(): Unit = pull(in) setHandlers(in, out, this) } } AKKA STREAMS - GRAPHSTAGE API
40. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 40 AKKA STREAMS - GRAPHSTAGE onPull onPush onPull onPull onPush
41. var lastPulled: Long = System.nanoTime() var lastPushed: Long = lastPulled private val backpressure = Kamon.histogram(label + "_backpressure") override def onPush(): Unit = { push(out, grab(in)) val now = System.nanoTime() backpressure.record((lastPulled - lastPushed) * 100 / now - lastPushed) lastPushed = now } override def onPull(): Unit = { pull(in) lastPulled = System.nanoTime() } setHandlers(in, out, this) AKKA STREAMS - BACKPRESSURE
42. var lastPulled: Long = System.nanoTime() var lastPushed: Long = lastPulled private val backpressure = Kamon.histogram(label + "_backpressure") override def onPush(): Unit = { push(out, grab(in)) val now = System.nanoTime() backpressure.record((lastPulled - lastPushed) * 100 / now - lastPushed) lastPushed = now } override def onPull(): Unit = { pull(in) lastPulled = System.nanoTime() } setHandlers(in, out, this) AKKA STREAMS - BACKPRESSURE
43. var lastPulled: Long = System.nanoTime() var lastPushed: Long = lastPulled private val backpressure = Kamon.histogram(label + "_backpressure") override def onPush(): Unit = { push(out, grab(in)) val now = System.nanoTime() backpressure.record((lastPulled - lastPushed) * 100 / now - lastPushed) lastPushed = now } override def onPull(): Unit = { pull(in) lastPulled = System.nanoTime() } setHandlers(in, out, this) AKKA STREAMS - BACKPRESSURE
44. var lastPulled: Long = System.nanoTime() var lastPushed: Long = lastPulled private val backpressure = Kamon.histogram(label + "_backpressure") override def onPush(): Unit = { push(out, grab(in)) val now = System.nanoTime() backpressure.record((lastPulled - lastPushed) * 100 / now - lastPushed) lastPushed = now } override def onPull(): Unit = { pull(in) lastPulled = System.nanoTime() } setHandlers(in, out, this) AKKA STREAMS - BACKPRESSURE
45. val pangram = Source(List("The", "quick", "brown", "fox", "jumps", "over", "the", "lazy", "dog")) val toChars = Flow[String].mapConcat(_.toList) val print = Sink.foreach(println) pangram .via(backpressureMeter("produced")) .via(toChars) .via(backpressureMeter("processed")) .runWith(print) AKKA STREAMS - BACKPRESSURE
46. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. THROUGHPUT BACKPRESSURE
47. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 47 svezfaz/akka-stream-checkpoint CURRENT FEATURES ・ throughput ・ backpressure ratio ・ push-pull latency ・ liveness checks FUTURE WORK ・ failures/completion
48. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. 48
49. © 2018 HOMEAWAY. ALL RIGHTS RESERVED. Stefano Bonetti Software Engineer @svezfaz @svez_faz Thank you.