Add relevant scan connection metrics for validator [ci]

Signed-off-by: Julien Tinguely <julien.tinguely@digitalasset.com>

Author: julientinguely-da

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/metrics/ScanConnectionMetrics.scala

@@ -0,0 +1,57 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.metrics
+
+import com.daml.metrics.api.MetricHandle.{LabeledMetricsFactory, Meter, Timer}
+import com.daml.metrics.api.MetricQualification.{Latency, Traffic}
+import com.daml.metrics.api.{MetricInfo, MetricName, MetricsContext}
+import org.lfdecentralizedtrust.splice.environment.SpliceMetrics
+
+class ScanConnectionMetrics(metricsFactory: LabeledMetricsFactory) extends AutoCloseable {
+  private val prefix: MetricName = SpliceMetrics.MetricsPrefix :+ "validator" :+ "scan"
+
+  val latencyPerConnection: Timer =
+    metricsFactory.timer(
+      MetricInfo(
+        name = prefix :+ "per_connection_latency",
+        summary = "Latency per scan connection query",
+        qualification = Latency,
+        labelsWithDescription = Map(
+          "scan_connection" -> "The scan connection of the request",
+          "request" -> "Type of request",
+        ),
+      )
+    )(MetricsContext.Empty)
+
+  val failuresPerConnection: Meter = metricsFactory.meter(
+    MetricInfo(
+      name = prefix :+ "per_connection_errors",
+      summary = "Count of per connection errors",
+      qualification = Traffic,
+      labelsWithDescription = Map(
+        "scan_connection" -> "The scan connection of the request",
+        "request" -> "Type of request",
+      ),
+    )
+  )(MetricsContext.Empty)
+
+  val bftReadLatency: Timer =
+    metricsFactory.timer(
+      MetricInfo(
+        name = prefix :+ "bft_read_latency",
+        summary = "Total latency after BFT reads",
+        qualification = Latency,
+      )
+    )(MetricsContext.Empty)
+
+  val bftCallFailures: Meter = metricsFactory.meter(
+    MetricInfo(
+      name = prefix :+ "bft_errors",
+      summary = "Count of BFT errors",
+      qualification = Traffic,
+    )
+  )(MetricsContext.Empty)
+
+  override def close(): Unit = ???
+}

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala

@@ -91,6 +91,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   DsoRules_CloseVoteRequestResult,
   VoteRequest,
 }
+import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
 import org.lfdecentralizedtrust.tokenstandard.{
   allocation,
   allocationinstruction,
@@ -114,6 +115,7 @@ class BftScanConnection(
     protected val clock: Clock,
     val retryProvider: RetryProvider,
     val loggerFactory: NamedLoggerFactory,
+    val connectionMetrics: Option[ScanConnectionMetrics] = None,
 )(implicit protected val ec: ExecutionContextExecutor, protected val mat: Materializer)
     extends FlagCloseableAsync
     with NamedLogging
@@ -633,6 +635,10 @@ class BftScanConnection(
       LoggerUtil.logThrowableAtLevel(consensusFailureLogLevel, msg, exception)
       Future.failed(exception)
     } else {
+      val timer = connectionMetrics match {
+        case Some(metrics) => Some(metrics.bftReadLatency.startAsync())
+        case None => None
+      }
       retryProvider
         .retryForClientCalls(
           "bft_call",
@@ -647,12 +653,23 @@ class BftScanConnection(
           logger,
           (_: String) => ConsensusNotReachedRetryable,
         )
+        .andThen { case _ =>
+          timer match {
+            case Some(t) => t.stop()
+            case None =>
+          }
+        }
         .recoverWith { case c: ConsensusNotReached =>
           val httpError = HttpErrorWithHttpCode(
             StatusCodes.BadGateway,
             s"Failed to reach consensus from ${callConfig.requestsToDo} Scan nodes, requiring ${callConfig.targetSuccess} matching responses.",
           )
           LoggerUtil.logThrowableAtLevel(consensusFailureLogLevel, s"Consensus not reached.", c)
+          connectionMetrics match {
+            case Some(metrics) =>
+              metrics.bftCallFailures.mark()
+            case None =>
+          }
           Future.failed(httpError)
         }
     }
@@ -1251,6 +1268,7 @@ object BftScanConnection {
       clock: Clock,
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
+      connectionMetrics: Option[ScanConnectionMetrics] = None,
       lastPersistedScanUrlList: () => Future[Option[List[(String, String)]]] = () =>
         Future.successful(None),
       persistScanUrlsCallback: Seq[(String, String)] => Future[Unit] = _ => Future.unit,
@@ -1262,7 +1280,8 @@ object BftScanConnection {
       templateDecoder: TemplateJsonDecoder,
   ): Future[BftScanConnection] = {
 
-    val builder = buildScanConnection(upgradesConfig, clock, retryProvider, loggerFactory)
+    val builder =
+      buildScanConnection(upgradesConfig, clock, retryProvider, loggerFactory, connectionMetrics)
     val logger = loggerFactory.getTracedLogger(getClass)
 
     config match {
@@ -1277,6 +1296,7 @@ object BftScanConnection {
           clock,
           retryProvider,
           loggerFactory,
+          connectionMetrics,
         )
 
       case ts @ BftScanClientConfig.BftCustom(_, _, _, _, _, _) =>
@@ -1457,7 +1477,8 @@ object BftScanConnection {
       httpClient: HttpClient,
       templateDecoder: TemplateJsonDecoder,
   ): Future[BftScanConnection] = {
-    val builder = buildScanConnection(upgradesConfig, clock, retryProvider, loggerFactory)
+    val builder =
+      buildScanConnection(upgradesConfig, clock, retryProvider, loggerFactory, None)
 
     for {
       scans <- retryProvider.retry(
@@ -1512,6 +1533,7 @@ object BftScanConnection {
       clock: Clock,
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
+      connectionMetrics: Option[ScanConnectionMetrics],
   )(implicit
       ec: ExecutionContextExecutor,
       tc: TraceContext,
@@ -1535,6 +1557,7 @@ object BftScanConnection {
           // We only need f+1 Scans to be available, so as long as those are connected we don't need to slow init down.
           // Furthermore, the refresh (either on init, or periodically) will retry anyway.
           retryConnectionOnInitialFailure = false,
+          connectionMetrics,
         )
 
   sealed trait BftScanClientConfig {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala

@@ -35,14 +35,15 @@ import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.lifecycle.FlagCloseableAsync
 import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger}
 import com.digitalasset.canton.time.Clock
-import com.digitalasset.canton.topology.{SynchronizerId, PartyId}
+import com.digitalasset.canton.topology.{PartyId, SynchronizerId}
 import com.digitalasset.canton.tracing.TraceContext
 import io.grpc.Status
 import org.apache.pekko.stream.Materializer
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   DsoRules_CloseVoteRequestResult,
   VoteRequest,
 }
+import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
 
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 import scala.jdk.OptionConverters.*
@@ -268,6 +269,7 @@ object ScanConnection {
       clock: Clock,
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
+      connectionMetrics: Option[ScanConnectionMetrics] = None,
       retryConnectionOnInitialFailure: Boolean = true,
   )(implicit
       ec: ExecutionContextExecutor,
@@ -284,6 +286,7 @@ object ScanConnection {
         clock,
         retryProvider,
         loggerFactory,
+        connectionMetrics,
       ),
       retryConnectionOnInitialFailure,
     )
@@ -295,6 +298,7 @@ object ScanConnection {
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
       retryConnectionOnInitialFailure: Boolean,
+      connectionMetrics: Option[ScanConnectionMetrics] = None,
   )(implicit
       ec: ExecutionContextExecutor,
       tc: TraceContext,
@@ -303,7 +307,14 @@ object ScanConnection {
       templateDecoder: TemplateJsonDecoder,
   ): Future[SingleScanConnection] =
     HttpAppConnection.checkVersionOrClose(
-      new SingleScanConnection(config, upgradesConfig, clock, retryProvider, loggerFactory),
+      new SingleScanConnection(
+        config,
+        upgradesConfig,
+        clock,
+        retryProvider,
+        loggerFactory,
+        connectionMetrics,
+      ),
       retryConnectionOnInitialFailure,
     )
 
@@ -313,14 +324,22 @@ object ScanConnection {
       clock: Clock,
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
+      connectionMetrics: Option[ScanConnectionMetrics] = None,
   )(implicit
       ec: ExecutionContextExecutor,
       tc: TraceContext,
       mat: Materializer,
       httpClient: HttpClient,
       templateDecoder: TemplateJsonDecoder,
   ): SingleScanConnection =
-    new SingleScanConnection(config, upgradesConfig, clock, retryProvider, loggerFactory)
+    new SingleScanConnection(
+      config,
+      upgradesConfig,
+      clock,
+      retryProvider,
+      loggerFactory,
+      connectionMetrics,
+    )
 
   private[client] case class CachedAmuletRules(
       cacheValidUntil: CantonTimestamp,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala

@@ -5,6 +5,8 @@ package org.lfdecentralizedtrust.splice.scan.admin.api.client
 
 import cats.data.OptionT
 import cats.syntax.either.*
+import com.daml.metrics.api.MetricsContext
+import com.daml.metrics.api.MetricsContext.Implicits.empty
 import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.FeaturedAppRight
 import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{
   AmuletRules,
@@ -31,7 +33,7 @@ import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   LookupTransferCommandStatusResponse,
   MigrationSchedule,
 }
-import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.{HttpScanAppClient}
+import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient
 import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig
 import org.lfdecentralizedtrust.splice.scan.store.db.ScanAggregator
 import org.lfdecentralizedtrust.splice.store.HistoryBackfilling.SourceMigrationInfo
@@ -67,13 +69,18 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   VoteRequest,
 }
 import io.grpc.Status
+import org.apache.pekko.http.scaladsl.model.{HttpHeader, Uri}
+import org.lfdecentralizedtrust.splice.admin.api.client.commands.HttpCommand
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.transferinstructionv1
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.transferinstructionv1.TransferInstruction
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationv1.Allocation
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationinstructionv1
+import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
 import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient.BftSequencer
 import org.lfdecentralizedtrust.tokenstandard.transferinstruction.v1.definitions.TransferFactoryWithChoiceContext
 
+import scala.util.{Failure, Success}
+
 /** Connection to the admin API of CC Scan. This is used by other apps
   * to query for the DSO party id.
   */
@@ -83,6 +90,7 @@ class SingleScanConnection private[client] (
     protected val clock: Clock,
     retryProvider: RetryProvider,
     outerLoggerFactory: NamedLoggerFactory,
+    connectionMetrics: Option[ScanConnectionMetrics],
 )(implicit
     protected val ec: ExecutionContextExecutor,
     tc: TraceContext,
@@ -101,6 +109,38 @@ class SingleScanConnection private[client] (
     with HasUrl {
   import ScanRoundAggregatesDecoder.*
 
+  override def runHttpCmd[Res, Result](
+      url: Uri,
+      command: HttpCommand[Res, Result],
+      headers: List[HttpHeader] = List.empty[HttpHeader],
+  )(implicit
+      templateDecoder: TemplateJsonDecoder,
+      httpClient: HttpClient,
+      tc: TraceContext,
+      ec: ExecutionContext,
+      mat: Materializer,
+  ): Future[Result] = {
+    connectionMetrics match {
+      case Some(metrics) =>
+        MetricsContext.withExtraMetricLabels(
+          ("scan_connection", url.scheme),
+          ("request", command.fullName),
+        ) { m =>
+          val timer = metrics.latencyPerConnection.startAsync()(m)
+          super
+            .runHttpCmd(url, command, headers)
+            .andThen {
+              case Failure(_) =>
+                metrics.failuresPerConnection.mark()(m)
+                timer.stop()(m)
+              case Success(_) =>
+                timer.stop()(m)
+            }
+        }
+      case None => super.runHttpCmd(url, command, headers)
+    }
+  }
+
   def url = config.adminApi.url
 
   // cached DSO reference. Never changes.
@@ -743,13 +783,21 @@ class CachedScanConnection private[client] (
     clock: Clock,
     retryProvider: RetryProvider,
     outerLoggerFactory: NamedLoggerFactory,
+    connectionMetrics: Option[ScanConnectionMetrics],
 )(implicit
     ec: ExecutionContextExecutor,
     tc: TraceContext,
     mat: Materializer,
     httpClient: HttpClient,
     templateDecoder: TemplateJsonDecoder,
-) extends SingleScanConnection(config, upgradesConfig, clock, retryProvider, outerLoggerFactory)
+) extends SingleScanConnection(
+      config,
+      upgradesConfig,
+      clock,
+      retryProvider,
+      outerLoggerFactory,
+      connectionMetrics,
+    )
     with CachingScanConnection {
 
   override protected val amuletRulesCacheTimeToLive: NonNegativeFiniteDuration =

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala

@@ -219,6 +219,7 @@ class ValidatorApp(
                 clock,
                 retryProvider,
                 loggerFactory,
+                Some(metrics.scanConnections),
                 ValidatorScanConnection.getPersistedScanList(configProvider),
                 ValidatorScanConnection.persistScanUrlListBuilder(configProvider),
               )
@@ -750,6 +751,7 @@ class ValidatorApp(
           clock,
           retryProvider,
           loggerFactory,
+          Some(metrics.scanConnections),
           ValidatorScanConnection.getPersistedScanList(configProvider),
           ValidatorScanConnection.persistScanUrlListBuilder(configProvider),
         )

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/metrics/ValidatorAppMetrics.scala

@@ -6,10 +6,13 @@ package org.lfdecentralizedtrust.splice.validator.metrics
 import com.daml.metrics.api.MetricHandle.LabeledMetricsFactory
 import com.digitalasset.canton.metrics.DbStorageHistograms
 import org.lfdecentralizedtrust.splice.BaseSpliceMetrics
+import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
 
 /** Modelled after [[com.digitalasset.canton.synchronizer.metrics.DomainMetrics]].
   */
 class ValidatorAppMetrics(
     metricsFactory: LabeledMetricsFactory,
     storageHistograms: DbStorageHistograms,
-) extends BaseSpliceMetrics("validator", metricsFactory, storageHistograms) {}
+) extends BaseSpliceMetrics("validator", metricsFactory, storageHistograms) {
+  val scanConnections = new ScanConnectionMetrics(metricsFactory)
+}