Make MetricsCollector communicate via messages only, not by logging

digital-asset · kamil-da · Nov 25, 2021 · Nov 16, 2021 · Nov 17, 2021 · Nov 17, 2021
commit d52644760a75c36bc6646e7c29f71e9cbe548584
diff --git a/ledger/ledger-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/Benchmark.scala b/ledger/ledger-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/Benchmark.scala
@@ -6,9 +6,9 @@ package com.daml.ledger.api.benchtool
 import com.daml.ledger.api.benchtool.config.WorkflowConfig.StreamConfig
 import com.daml.ledger.api.benchtool.metrics.{
   MetricRegistryOwner,
-  MetricsCollector,
   MetricsSet,
   StreamMetrics,
+  StreamResult,
 }
 import com.daml.ledger.api.benchtool.services.LedgerApiServices
 import com.daml.ledger.api.benchtool.util.TypedActorSystemResourceOwner
@@ -110,7 +110,7 @@ object Benchmark {
         }
         .transform {
           case Success(results) =>
-            if (results.contains(MetricsCollector.Message.MetricsResult.ObjectivesViolated))
+            if (results.contains(StreamResult.ObjectivesViolated))
               Failure(new RuntimeException("Metrics objectives not met."))
             else Success(())
           case Failure(ex) =>

diff --git a/...nch-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MeteredStreamObserver.scala b/...nch-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MeteredStreamObserver.scala
@@ -12,14 +12,15 @@ class MeteredStreamObserver[T](
     val streamName: String,
     logger: Logger,
     manager: MetricsManager[T],
-) extends ObserverWithResult[T, MetricsCollector.Message.MetricsResult](logger) {
+) extends ObserverWithResult[T, StreamResult](logger) {
 
   override def onNext(value: T): Unit = {
+    Thread.sleep(1000)
     manager.sendNewValue(value)
     super.onNext(value)
   }
 
-  override def completeWith(): Future[MetricsCollector.Message.MetricsResult] = {
+  override def completeWith(): Future[StreamResult] = {
     logger.debug(withStreamName(s"Asking for stream result..."))
     manager.result()
   }

diff --git a/...pi-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MetricsCollector.scala b/...pi-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MetricsCollector.scala
@@ -3,54 +3,50 @@
 
 package com.daml.ledger.api.benchtool.metrics
 
-import akka.actor.typed.scaladsl.{Behaviors, TimerScheduler}
+import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.{ActorRef, Behavior}
-import com.daml.ledger.api.benchtool.util.{MetricFormatter, TimeUtil}
+import com.daml.ledger.api.benchtool.metrics.objectives.ServiceLevelObjective
+import com.daml.ledger.api.benchtool.util.TimeUtil
 
 import java.time.Instant
-import scala.concurrent.duration._
 
 object MetricsCollector {
 
   sealed trait Message
   object Message {
-    sealed trait MetricsResult
-    object MetricsResult {
-      final case object Ok extends MetricsResult
-      final case object ObjectivesViolated extends MetricsResult
-    }
     final case class NewValue[T](value: T) extends Message
-    final case class PeriodicUpdateCommand() extends Message
-    final case class StreamCompleted(replyTo: ActorRef[MetricsResult]) extends Message
+    final case class PeriodicReportRequest(replyTo: ActorRef[Response.PeriodicReport])
+        extends Message
+    final case class FinalReportRequest(replyTo: ActorRef[Response.FinalReport]) extends Message
+  }
+
+  sealed trait Response
+  object Response {
+    final case class PeriodicReport(values: List[MetricValue]) extends Response
+    final case class MetricFinalReportData(
+        name: String,
+        value: MetricValue,
+        violatedObjective: Option[(ServiceLevelObjective[_], MetricValue)],
+    )
+    final case class FinalReport(metricsData: List[MetricFinalReportData]) extends Response
   }
 
   def apply[T](
-      streamName: String,
       metrics: List[Metric[T]],
-      logInterval: FiniteDuration,
-      reporter: MetricFormatter,
       exposedMetrics: Option[ExposedMetrics[T]] = None,
-  ): Behavior[Message] =
-    Behaviors.withTimers { timers =>
-      val startTime: Instant = Instant.now()
-      new MetricsCollector[T](timers, streamName, logInterval, reporter, startTime, exposedMetrics)
-        .handlingMessages(metrics, startTime)
-    }
-
+  ): Behavior[Message] = {
+    val startTime: Instant = Instant.now()
+    new MetricsCollector[T](startTime, exposedMetrics).handlingMessages(metrics, startTime)
+  }
 }
 
 class MetricsCollector[T](
-    timers: TimerScheduler[MetricsCollector.Message],
-    streamName: String,
-    logInterval: FiniteDuration,
-    reporter: MetricFormatter,
     startTime: Instant,
     exposedMetrics: Option[ExposedMetrics[T]],
 ) {
   import MetricsCollector._
   import MetricsCollector.Message._
-
-  timers.startTimerWithFixedDelay(PeriodicUpdateCommand(), logInterval)
+  import MetricsCollector.Response._
 
   def handlingMessages(metrics: List[Metric[T]], lastPeriodicCheck: Instant): Behavior[Message] = {
     Behaviors.receive { case (context, message) =>
@@ -59,39 +55,29 @@ class MetricsCollector[T](
           exposedMetrics.foreach(_.onNext(newValue.value))
           handlingMessages(metrics.map(_.onNext(newValue.value)), lastPeriodicCheck)
 
-        case _: PeriodicUpdateCommand =>
+        case request: PeriodicReportRequest =>
           val currentTime = Instant.now()
           val (newMetrics, values) = metrics
             .map(_.periodicValue(TimeUtil.durationBetween(lastPeriodicCheck, currentTime)))
             .unzip
-          context.log.info(namedMessage(reporter.formattedValues(values)))
+          context.log.info(s"RETURNING VALUES")
+          request.replyTo ! Response.PeriodicReport(values)
           handlingMessages(newMetrics, currentTime)
 
-        case message: StreamCompleted =>
-          context.log.info(
-            namedMessage(
-              reporter.finalReport(
-                streamName = streamName,
-                metrics = metrics,
-                duration = totalDuration,
+        case request: FinalReportRequest =>
+          val duration = TimeUtil.durationBetween(startTime, Instant.now())
+          val data: List[MetricFinalReportData] =
+            metrics.map { metric =>
+              MetricFinalReportData(
+                name = metric.name,
+                value = metric.finalValue(duration),
+                violatedObjective = metric.violatedObjective,
               )
-            )
-          )
-          message.replyTo ! result(metrics)
+            }
+          context.log.info(s"RETURNING FINAL DATA")
+          request.replyTo ! FinalReport(data)
           Behaviors.stopped
       }
     }
   }
-
-  private def result(metrics: List[Metric[T]]): MetricsResult = {
-    val atLeastOneObjectiveViolated = metrics.exists(_.violatedObjective.nonEmpty)
-
-    if (atLeastOneObjectiveViolated) MetricsResult.ObjectivesViolated
-    else MetricsResult.Ok
-  }
-
-  private def namedMessage(message: String) = s"[$streamName] $message"
-
-  private def totalDuration: java.time.Duration =
-    TimeUtil.durationBetween(startTime, Instant.now())
 }
diff --git a/...-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MetricsManager.scala b/...-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/MetricsManager.scala
@@ -7,12 +7,14 @@ import akka.actor.CoordinatedShutdown
 import akka.actor.typed.scaladsl.AskPattern._
 import akka.actor.typed.{ActorRef, ActorSystem, Props, SpawnProtocol}
 import akka.util.Timeout
-import com.daml.ledger.api.benchtool.util.MetricFormatter
 
 import scala.concurrent.duration._
 import scala.concurrent.{ExecutionContext, Future}
 
-case class MetricsManager[T](collector: ActorRef[MetricsCollector.Message])(implicit
+case class MetricsManager[T](
+    collector: ActorRef[MetricsCollector.Message],
+    logInterval: FiniteDuration,
+)(implicit
     system: ActorSystem[SpawnProtocol.Command]
 ) {
   CoordinatedShutdown(system).addTask(
@@ -23,12 +25,30 @@ case class MetricsManager[T](collector: ActorRef[MetricsCollector.Message])(impl
     result().map(_ => akka.Done)(system.executionContext)
   }
 
+  system.scheduler.scheduleWithFixedDelay(logInterval, logInterval)(() => {
+    implicit val timeout: Timeout = Timeout(logInterval)
+    collector
+      .ask(MetricsCollector.Message.PeriodicReportRequest)
+      .map { response =>
+        println(s"LOG NICE PERIODIC REPORT: ${response}")
+      }(system.executionContext)
+    ()
+  })(system.executionContext)
+
   def sendNewValue(value: T): Unit =
     collector ! MetricsCollector.Message.NewValue(value)
 
-  def result[Result](): Future[MetricsCollector.Message.MetricsResult] = {
+  def result[Result](): Future[StreamResult] = {
     implicit val timeout: Timeout = Timeout(3.seconds)
-    collector.ask(MetricsCollector.Message.StreamCompleted)
+    val result = collector.ask(MetricsCollector.Message.FinalReportRequest)
+
+    result.map { response: MetricsCollector.Response.FinalReport =>
+      println(s"PRINTLN NICE REPORT: ${response}")
+      if (response.metricsData.exists(_.violatedObjective.isDefined))
+        StreamResult.ObjectivesViolated
+      else
+        StreamResult.Ok
+    }(system.executionContext)
   }
 }
 
@@ -47,10 +67,7 @@ object MetricsManager {
     val collectorActor: Future[ActorRef[MetricsCollector.Message]] = system.ask(
       SpawnProtocol.Spawn(
         behavior = MetricsCollector(
-          streamName = streamName,
           metrics = metrics,
-          logInterval = logInterval,
-          reporter = MetricFormatter.Default,
           exposedMetrics = exposedMetrics,
         ),
         name = s"${streamName}-collector",
@@ -59,6 +76,11 @@ object MetricsManager {
       )
     )
 
-    collectorActor.map(MetricsManager[StreamElem](_))
+    collectorActor.map(collector =>
+      MetricsManager[StreamElem](
+        collector = collector,
+        logInterval = logInterval,
+      )
+    )
   }
 }
diff --git a/...er-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/StreamResult.scala b/...er-api-bench-tool/src/main/scala/com/daml/ledger/api/benchtool/metrics/StreamResult.scala
@@ -0,0 +1,11 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.benchtool.metrics
+
+sealed trait StreamResult extends Product with Serializable
+
+object StreamResult {
+  final case object Ok extends StreamResult
+  final case object ObjectivesViolated extends StreamResult
+}