participant-state-metrics: Add wrappers for v2. (#10404)

CHANGELOG_BEGIN
CHANGELOG_END

ledger/metrics/src/main/scala/com/daml/metrics/Metrics.scala

@@ -718,6 +718,7 @@ final class Metrics(val registry: MetricRegistry) {
         val allocateParty: Timer = registry.timer(Prefix :+ "allocate_party")
         val submitConfiguration: Timer = registry.timer(Prefix :+ "submit_configuration")
         val prune: Timer = registry.timer(Prefix :+ "prune")
+        val rejectSubmission: Timer = registry.timer(Prefix :+ "reject_submission")
       }
     }
 

ledger/participant-state-metrics/BUILD.bazel

@@ -23,6 +23,7 @@ da_scala_library(
         "//daml-lf/archive:daml_lf_dev_archive_proto_java",
         "//daml-lf/data",
         "//daml-lf/transaction",
+        "//ledger-api/grpc-definitions:ledger_api_proto_scala",
         "//ledger/ledger-api-health",
         "//ledger/ledger-configuration",
         "//ledger/ledger-offset",

ledger/participant-state-metrics/src/main/scala/com/daml/ledger/participant/state/v2/metrics/TimedReadService.scala

@@ -0,0 +1,27 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.participant.state.v2.metrics
+
+import akka.NotUsed
+import akka.stream.scaladsl.Source
+import com.daml.ledger.api.health.HealthStatus
+import com.daml.ledger.configuration.LedgerInitialConditions
+import com.daml.ledger.offset.Offset
+import com.daml.ledger.participant.state.v2.{ReadService, Update}
+import com.daml.metrics.{Metrics, Timed}
+
+final class TimedReadService(delegate: ReadService, metrics: Metrics) extends ReadService {
+
+  override def ledgerInitialConditions(): Source[LedgerInitialConditions, NotUsed] =
+    Timed.source(
+      metrics.daml.services.read.getLedgerInitialConditions,
+      delegate.ledgerInitialConditions(),
+    )
+
+  override def stateUpdates(beginAfter: Option[Offset]): Source[(Offset, Update), NotUsed] =
+    Timed.source(metrics.daml.services.read.stateUpdates, delegate.stateUpdates(beginAfter))
+
+  override def currentHealth(): HealthStatus =
+    delegate.currentHealth()
+}

ledger/participant-state-metrics/src/main/scala/com/daml/ledger/participant/state/v2/metrics/TimedWriteService.scala

@@ -0,0 +1,95 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.participant.state.v2.metrics
+
+import java.util.concurrent.CompletionStage
+
+import com.daml.daml_lf_dev.DamlLf
+import com.daml.ledger.api.health.HealthStatus
+import com.daml.ledger.configuration.Configuration
+import com.daml.ledger.offset.Offset
+import com.daml.ledger.participant.state.v2.{
+  PruningResult,
+  SubmissionResult,
+  SubmitterInfo,
+  TransactionMeta,
+  WriteService,
+}
+import com.daml.lf.data.{Ref, Time}
+import com.daml.lf.transaction.SubmittedTransaction
+import com.daml.metrics.{Metrics, Timed}
+import com.daml.telemetry.TelemetryContext
+import com.google.rpc.status.Status
+
+final class TimedWriteService(delegate: WriteService, metrics: Metrics) extends WriteService {
+
+  override def submitTransaction(
+      submitterInfo: SubmitterInfo,
+      transactionMeta: TransactionMeta,
+      transaction: SubmittedTransaction,
+      estimatedInterpretationCost: Long,
+  )(implicit telemetryContext: TelemetryContext): CompletionStage[SubmissionResult] =
+    Timed.timedAndTrackedCompletionStage(
+      metrics.daml.services.write.submitTransaction,
+      metrics.daml.services.write.submitTransactionRunning,
+      delegate.submitTransaction(
+        submitterInfo,
+        transactionMeta,
+        transaction,
+        estimatedInterpretationCost,
+      ),
+    )
+
+  override def uploadPackages(
+      submissionId: Ref.SubmissionId,
+      archives: List[DamlLf.Archive],
+      sourceDescription: Option[String],
+  )(implicit telemetryContext: TelemetryContext): CompletionStage[SubmissionResult] =
+    Timed.completionStage(
+      metrics.daml.services.write.uploadPackages,
+      delegate.uploadPackages(submissionId, archives, sourceDescription),
+    )
+
+  override def allocateParty(
+      hint: Option[Ref.Party],
+      displayName: Option[String],
+      submissionId: Ref.SubmissionId,
+  )(implicit telemetryContext: TelemetryContext): CompletionStage[SubmissionResult] =
+    Timed.completionStage(
+      metrics.daml.services.write.allocateParty,
+      delegate.allocateParty(hint, displayName, submissionId),
+    )
+
+  override def submitConfiguration(
+      maxRecordTime: Time.Timestamp,
+      submissionId: Ref.SubmissionId,
+      config: Configuration,
+  )(implicit telemetryContext: TelemetryContext): CompletionStage[SubmissionResult] =
+    Timed.completionStage(
+      metrics.daml.services.write.submitConfiguration,
+      delegate.submitConfiguration(maxRecordTime, submissionId, config),
+    )
+
+  override def prune(
+      pruneUpToInclusive: Offset,
+      submissionId: Ref.SubmissionId,
+  ): CompletionStage[PruningResult] =
+    Timed.completionStage(
+      metrics.daml.services.write.prune,
+      delegate.prune(pruneUpToInclusive, submissionId),
+    )
+
+  override def currentHealth(): HealthStatus =
+    delegate.currentHealth()
+
+  override def rejectSubmission(
+      submitterInfo: SubmitterInfo,
+      submissionTime: Time.Timestamp,
+      reason: Status,
+  )(implicit telemetryContext: TelemetryContext): CompletionStage[SubmissionResult] =
+    Timed.completionStage(
+      metrics.daml.services.write.rejectSubmission,
+      delegate.rejectSubmission(submitterInfo, submissionTime, reason),
+    )
+}