feat(prism-agent): Metrics for connection flow job

build.sbt

@@ -147,7 +147,7 @@ lazy val D = new {
 
 lazy val D_Shared = new {
   lazy val dependencies: Seq[ModuleID] =
-    Seq(D.typesafeConfig, D.scalaPbGrpc, D.testcontainersPostgres, D.testcontainersVault)
+    Seq(D.typesafeConfig, D.scalaPbGrpc, D.testcontainersPostgres, D.testcontainersVault, D.zio)
 }
 
 lazy val D_Connect = new {

connect/lib/core/src/main/scala/io/iohk/atala/connect/core/service/ConnectionServiceImpl.scala

@@ -136,7 +136,7 @@ private class ConnectionServiceImpl(
       record <- getRecordWithState(recordId, ProtocolState.InvitationReceived)
       request = ConnectionRequest
         .makeFromInvitation(record.invitation, pairwiseDid)
-        .copy(thid = Some(record.invitation.id)) //  This logic shound be move to the SQL when fetching the record
+        .copy(thid = Some(record.invitation.id)) //  This logic should be moved to the SQL when fetching the record
       count <- connectionRepository
         .updateWithConnectionRequest(recordId, request, ProtocolState.ConnectionRequestPending, maxRetries)
         .mapError(RepositoryError.apply)

prism-agent/service/server/src/main/scala/io/iohk/atala/agent/server/jobs/ConnectBackgroundJobs.scala

@@ -14,6 +14,7 @@ import io.iohk.atala.mercury.model.*
 import io.iohk.atala.mercury.model.error.*
 import io.iohk.atala.resolvers.DIDResolver
 import zio.*
+import zio.metrics.*
 object ConnectBackgroundJobs {
 
   val didCommExchanges = {
@@ -37,6 +38,64 @@ object ConnectBackgroundJobs {
   ): URIO[DidOps & DIDResolver & HttpClient & ConnectionService & ManagedDIDService, Unit] = {
     import ProtocolState.*
     import Role.*
+
+    val InviteeConnectionRequestMsgFailed =
+      Metric
+        .counterInt("connection_flow_invitee_connection_request_msg_failed_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val InviteeConnectionRequestMsgSuccess = Metric
+      .counterInt("connection_flow_invitee_connection_request_msg_success_count")
+      .fromConst(1)
+      .tagged("connectionId", record.id.toString)
+
+    val InviterConnectionResponseMsgFailed =
+      Metric
+        .counterInt("connection_flow_inviter_connection_response_msg_failed_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val InviterConnectionResponseMsgSuccess =
+      Metric
+        .counterInt("connection_flow_inviter_connection_response_msg_success_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviteePendingSuccess =
+      Metric
+        .counterInt("connection_flow_invitee_process_connection_record_success_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviteePendingFailed =
+      Metric
+        .counterInt("connection_flow_invitee_process_connection_record_failed_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviteePendingTotal = Metric
+      .counterInt("connection_flow_invitee_process_connection_record_total_count")
+      .fromConst(1)
+      .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviterPendingSuccess =
+      Metric
+        .counterInt("connection_flow_inviter_process_connection_record_success_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviterPendingFailed =
+      Metric
+        .counterInt("connection_flow_inviter_process_connection_record_failed_count")
+        .fromConst(1)
+        .tagged("connectionId", record.id.toString)
+
+    val ProcessConnectionRecordInviterPendingTotal = Metric
+      .counterInt("connection_flow_inviter_process_connection_record_total_count")
+      .fromConst(1)
+      .tagged("connectionId", record.id.toString)
+
     val exchange = record match {
       case ConnectionRecord(
             id,
@@ -53,19 +112,31 @@ object ConnectBackgroundJobs {
             _,
             _
           ) if metaRetries > 0 =>
-        val aux = for {
+        val inviteeProcessFlow = for {
 
           didCommAgent <- buildDIDCommAgent(request.from)
-          resp <- MessagingService.send(request.makeMessage).provideSomeLayer(didCommAgent)
+          resp <- MessagingService.send(request.makeMessage).provideSomeLayer(didCommAgent) @@ Metric
+            .gauge("connection_flow_invitee_send_connection_request_ms")
+            .tagged("connectionId", record.id.toString)
+            .trackDurationWith(_.toNanos.toDouble)
           connectionService <- ZIO.service[ConnectionService]
           _ <- {
-            if (resp.status >= 200 && resp.status < 300) connectionService.markConnectionRequestSent(id)
-            else ZIO.fail(ErrorResponseReceivedFromPeerAgent(resp))
+            if (resp.status >= 200 && resp.status < 300)
+              connectionService.markConnectionRequestSent(id) @@ InviteeConnectionRequestMsgSuccess
+            else ZIO.fail(ErrorResponseReceivedFromPeerAgent(resp)) @@ InviteeConnectionRequestMsgFailed
           }
         } yield ()
 
-        // aux // TODO decrete metaRetries if it has a error
-        aux
+        // inviteeProcessFlow // TODO decrease metaRetries if it has a error
+
+        inviteeProcessFlow
+          @@ ProcessConnectionRecordInviteePendingSuccess.trackSuccess
+          @@ ProcessConnectionRecordInviteePendingFailed.trackError
+          @@ ProcessConnectionRecordInviteePendingTotal
+          @@ Metric
+            .gauge("connection_flow_invitee_process_connection_record_ms")
+            .tagged("connectionId", record.id.toString)
+            .trackDurationWith(_.toNanos.toDouble)
 
       case ConnectionRecord(
             id,
@@ -82,16 +153,29 @@ object ConnectBackgroundJobs {
             _,
             _
           ) if metaRetries > 0 =>
-        for {
+        val inviterProcessFlow = for {
           didCommAgent <- buildDIDCommAgent(response.from)
-          resp <- MessagingService.send(response.makeMessage).provideSomeLayer(didCommAgent)
+          resp <- MessagingService.send(response.makeMessage).provideSomeLayer(didCommAgent) @@ Metric
+            .gauge("connection_flow_inviter_send_connection_response_ms")
+            .tagged("connectionId", record.id.toString)
+            .trackDurationWith(_.toNanos.toDouble)
           connectionService <- ZIO.service[ConnectionService]
           _ <- {
-            if (resp.status >= 200 && resp.status < 300) connectionService.markConnectionResponseSent(id)
-            else ZIO.fail(ErrorResponseReceivedFromPeerAgent(resp))
+            if (resp.status >= 200 && resp.status < 300)
+              connectionService.markConnectionResponseSent(id) @@ InviterConnectionResponseMsgSuccess
+            else ZIO.fail(ErrorResponseReceivedFromPeerAgent(resp)) @@ InviterConnectionResponseMsgFailed
           }
         } yield ()
 
+        inviterProcessFlow
+          @@ ProcessConnectionRecordInviterPendingSuccess.trackSuccess
+          @@ ProcessConnectionRecordInviterPendingFailed.trackError
+          @@ ProcessConnectionRecordInviterPendingTotal
+          @@ Metric
+            .gauge("connection_flow_inviter_process_connection_record_ms")
+            .tagged("connectionId", record.id.toString)
+            .trackDurationWith(_.toNanos.toDouble)
+
       case _ => ZIO.unit
     }
 

prism-agent/service/server/src/main/scala/io/iohk/atala/connect/controller/ConnectionControllerImpl.scala

@@ -47,11 +47,41 @@ class ConnectionControllerImpl(
     result.mapError(toHttpError)
   }
 
+  object CustomMetricsAspect {
+    import java.util.concurrent.TimeUnit
+    import zio.metrics.*
+    def attachDurationGaugeMetric(name: String): ZIOAspect[Nothing, Any, Nothing, Any, Nothing, Any] =
+      new ZIOAspect[Nothing, Any, Nothing, Any, Nothing, Any] {
+        override def apply[R, E, A](
+            zio: ZIO[R, E, A]
+        )(implicit trace: Trace): ZIO[R, E, A] =
+          def currTime = Clock.currentTime(TimeUnit.MILLISECONDS)
+
+          for {
+            timeBefore <- currTime
+            res <- zio
+            timeAfter <- currTime
+            _ <- ZIO.succeed((timeAfter - timeBefore).toDouble) @@ Metric.gauge(name)
+          } yield res
+      }
+  }
+
   override def getConnections(
       paginationInput: PaginationInput,
       thid: Option[String]
   )(implicit rc: RequestContext): IO[ErrorResponse, ConnectionsPage] = {
+
+    val deleyEffect = ZIO.sleep(Duration.fromMillis(2000)).map(_ => 5)
+    import zio.metrics.*
+
+    val successZIO = ZIO.succeed(1)
+    val errZIO = ZIO.fail("err")
+
+    val succCounter = Metric.counter("success_counter").fromConst(1)
+    val failCounter = Metric.counter("fail_counter").fromConst(1)
+
     val result = for {
+      _ <- successZIO @@ succCounter.trackError
       connections <- thid match
         case None       => service.getConnectionRecords()
         case Some(thid) => service.getConnectionRecordByThreadId(thid).map(_.toSeq)

shared/src/main/scala/io/iohk/atala/shared/utils/aspects/CustomMetricsAspect.scala

@@ -0,0 +1,22 @@
+package io.iohk.atala.shared.utils.aspects
+import java.util.concurrent.TimeUnit
+import zio.*
+import zio.metrics.*
+
+object CustomMetricsAspect {
+
+  def attachDurationGaugeMetric(name: String): ZIOAspect[Nothing, Any, Nothing, Any, Nothing, Any] =
+    new ZIOAspect[Nothing, Any, Nothing, Any, Nothing, Any] {
+      override def apply[R, E, A](
+          zio: ZIO[R, E, A]
+      )(implicit trace: Trace): ZIO[R, E, A] =
+        def currTime = Clock.currentTime(TimeUnit.MILLISECONDS)
+
+        for {
+          timeBefore <- currTime
+          res <- zio
+          timeAfter <- currTime
+          _ <- ZIO.succeed((timeAfter - timeBefore).toDouble) @@ Metric.gauge(name)
+        } yield res
+    }
+}