Monitor onion messages (#2877)

Improve metrics for for onion messages.
We count messages sent and received, throttled and that couldn't be relayed.

eclair-core/src/main/scala/fr/acinq/eclair/io/MessageRelay.scala

@@ -24,6 +24,7 @@ import akka.actor.{ActorRef, typed}
 import fr.acinq.bitcoin.scalacompat.ByteVector32
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.eclair.channel.Register
+import fr.acinq.eclair.io.Monitoring.{Metrics, Tags}
 import fr.acinq.eclair.io.Peer.{PeerInfo, PeerInfoResponse}
 import fr.acinq.eclair.io.Switchboard.GetPeerInfo
 import fr.acinq.eclair.message.OnionMessages
@@ -115,6 +116,7 @@ private class MessageRelay(nodeParams: NodeParams,
   private def waitForNextNodeId(msg: OnionMessage, channelId: ShortChannelId): Behavior[Command] = {
     Behaviors.receiveMessagePartial {
       case WrappedOptionalNodeId(None) =>
+        Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, Tags.Reasons.UnknownNextNodeId).increment()
         replyTo_opt.foreach(_ ! UnknownChannel(messageId, channelId))
         Behaviors.stopped
       case WrappedOptionalNodeId(Some(nextNodeId)) =>
@@ -126,6 +128,7 @@ private class MessageRelay(nodeParams: NodeParams,
     if (nextNodeId == nodeParams.nodeId) {
       OnionMessages.process(nodeParams.privateKey, msg) match {
         case OnionMessages.DropMessage(reason) =>
+          Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, reason.getClass.getSimpleName).increment()
           replyTo_opt.foreach(_ ! DroppedMessage(messageId, reason))
           Behaviors.stopped
         case OnionMessages.SendMessage(nextNode, nextMessage) =>
@@ -154,6 +157,7 @@ private class MessageRelay(nodeParams: NodeParams,
         switchboard ! GetPeerInfo(context.messageAdapter(WrappedPeerInfo), nextNodeId)
         waitForNextPeerForPolicyCheck(msg)
       case _ =>
+        Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, Tags.Reasons.NoChannelWithPreviousPeer).increment()
         replyTo_opt.foreach(_ ! AgainstPolicy(messageId, RelayChannelsOnly))
         Behaviors.stopped
     }
@@ -165,6 +169,7 @@ private class MessageRelay(nodeParams: NodeParams,
         peer ! Peer.RelayOnionMessage(messageId, msg, replyTo_opt)
         Behaviors.stopped
       case _ =>
+        Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, Tags.Reasons.NoChannelWithNextPeer).increment()
         replyTo_opt.foreach(_ ! AgainstPolicy(messageId, RelayChannelsOnly))
         Behaviors.stopped
     }
@@ -176,6 +181,7 @@ private class MessageRelay(nodeParams: NodeParams,
         r.peer ! Peer.RelayOnionMessage(messageId, msg, replyTo_opt)
         Behaviors.stopped
       case WrappedConnectionResult(f: PeerConnection.ConnectionResult.Failure) =>
+        Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, Tags.Reasons.ConnectionFailure).increment()
         replyTo_opt.foreach(_ ! ConnectionFailure(messageId, f))
         Behaviors.stopped
     }

eclair-core/src/main/scala/fr/acinq/eclair/io/Monitoring.scala

@@ -28,9 +28,9 @@ object Monitoring {
 
     val ReconnectionsAttempts = Kamon.counter("reconnections.attempts")
 
-    val OnionMessagesReceived = Kamon.counter("onionmessages.received")
+    val OnionMessagesProcessed = Kamon.counter("onionmessages.processed")
-    val OnionMessagesSent = Kamon.counter("onionmessages.sent")
     val OnionMessagesThrottled = Kamon.counter("onionmessages.throttled")
+    val OnionMessagesNotRelayed = Kamon.counter("onionmessages.not-relayed")
 
     val OpenChannelRequestsPending = Kamon.gauge("openchannelrequests.pending")
 
@@ -51,6 +51,19 @@ object Monitoring {
 
     val PublicPeers = "public"
 
+    val Direction = "direction"
+    object Directions {
+      val Incoming = "IN"
+      val Outgoing = "OUT"
+    }
+
+    val Reason = "reason"
+    object Reasons {
+      val UnknownNextNodeId = "UnknownNextNodeId"
+      val NoChannelWithPreviousPeer = "NoChannelWithPreviousPeer"
+      val NoChannelWithNextPeer = "NoChannelWithNextPeer"
+      val ConnectionFailure = "ConnectionFailure"
+    }
   }
 
 }

eclair-core/src/main/scala/fr/acinq/eclair/io/PeerConnection.scala

@@ -182,19 +182,19 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
           d.transport ! TransportHandler.ReadAck(msg)
           if (incomingRateLimiter.tryAcquire()) {
             d.peer ! msg
-            Metrics.OnionMessagesReceived.withoutTags().increment()
+            Metrics.OnionMessagesProcessed.withTag(Tags.Direction, Tags.Directions.Incoming).increment()
           } else {
-            Metrics.OnionMessagesThrottled.withoutTags().increment()
+            Metrics.OnionMessagesThrottled.withTag(Tags.Direction, Tags.Directions.Incoming).increment()
           }
         } else {
           if (outgoingRateLimiter.tryAcquire()) {
             d.transport forward msg
-            Metrics.OnionMessagesSent.withoutTags().increment()
+            Metrics.OnionMessagesProcessed.withTag(Tags.Direction, Tags.Directions.Outgoing).increment()
             if (!d.isPersistent) {
               startSingleTimer(KILL_IDLE_TIMER, KillIdle, conf.killIdleDelay)
             }
           } else {
-            Metrics.OnionMessagesThrottled.withoutTags().increment()
+            Metrics.OnionMessagesThrottled.withTag(Tags.Direction, Tags.Directions.Outgoing).increment()
           }
         }
         stay()