Measure the distribution of payments across nodes (#1644)

We put nodes in buckets in order to build a distribution and monitor
incoming/outgoing payment count/volume by node id.

eclair-core/src/main/scala/fr/acinq/eclair/channel/Commitments.scala

@@ -384,6 +384,7 @@ object Commitments {
       case Some(htlc) if htlc.paymentHash == sha256(cmd.r) =>
         val fulfill = UpdateFulfillHtlc(commitments.channelId, cmd.id, cmd.r)
         val commitments1 = addLocalProposal(commitments, fulfill)
+        payment.Monitoring.Metrics.recordIncomingPaymentDistribution(commitments.remoteParams.nodeId, htlc.amountMsat)
         Right((commitments1, fulfill))
       case Some(_) => Left(InvalidHtlcPreimage(commitments.channelId, cmd.id))
       case None => Left(UnknownHtlcId(commitments.channelId, cmd.id))
@@ -392,7 +393,9 @@ object Commitments {
   def receiveFulfill(commitments: Commitments, fulfill: UpdateFulfillHtlc): Either[ChannelException, (Commitments, Origin, UpdateAddHtlc)] =
     commitments.getOutgoingHtlcCrossSigned(fulfill.id) match {
       case Some(htlc) if htlc.paymentHash == sha256(fulfill.paymentPreimage) => commitments.originChannels.get(fulfill.id) match {
-        case Some(origin) => Right(addRemoteProposal(commitments, fulfill), origin, htlc)
+        case Some(origin) =>
+          payment.Monitoring.Metrics.recordOutgoingPaymentDistribution(commitments.remoteParams.nodeId, htlc.amountMsat)
+          Right(addRemoteProposal(commitments, fulfill), origin, htlc)
         case None => Left(UnknownHtlcId(commitments.channelId, fulfill.id))
       }
       case Some(_) => Left(InvalidHtlcPreimage(commitments.channelId, fulfill.id))

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

@@ -16,6 +16,8 @@
 
 package fr.acinq.eclair.payment
 
+import fr.acinq.bitcoin.Crypto.PublicKey
+import fr.acinq.eclair.MilliSatoshi
 import fr.acinq.eclair.channel.CMD_FAIL_HTLC
 import kamon.Kamon
 
@@ -35,12 +37,34 @@ object Monitoring {
     // Once enough data has been collected, we will update the MultiPartPaymentLifecycle logic accordingly.
     val RetryFailedChannelsResult = Kamon.counter("payment.mpp.retry-failed-channels-result")
 
+    private val PaymentNodeInAmount = Kamon.histogram("payment.node.in.amount", "Distribution of incoming payments across nodes (satoshi)")
+    private val PaymentNodeIn = Kamon.histogram("payment.node.in", "Distribution of incoming payments across nodes (count)")
+    private val PaymentNodeOutAmount = Kamon.histogram("payment.node.out.amount", "Distribution of outgoing payments across nodes (satoshi)")
+    private val PaymentNodeOut = Kamon.histogram("payment.node.out", "Distribution of outgoing payments across nodes (count)")
+
     def recordPaymentRelayFailed(failureType: String, relayType: String): Unit =
       Metrics.PaymentFailed
         .withTag(Tags.Direction, Tags.Directions.Relayed)
         .withTag(Tags.Failure, failureType)
         .withTag(Tags.Relay, relayType)
         .increment()
+
+    /**
+     * Assign a bucket to a node id. There are 256 buckets.
+     */
+    def nodeIdBucket(nodeId: PublicKey): Short = nodeId.value.takeRight(1).toShort(signed = false) // we use short to not have negative values
+
+    def recordIncomingPaymentDistribution(nodeId: PublicKey, amount: MilliSatoshi): Unit = {
+      val bucket = nodeIdBucket(nodeId)
+      PaymentNodeInAmount.withoutTags().record(bucket, amount.truncateToSatoshi.toLong)
+      PaymentNodeIn.withoutTags().record(bucket)
+    }
+
+    def recordOutgoingPaymentDistribution(nodeId: PublicKey, amount: MilliSatoshi): Unit = {
+      val bucket = nodeIdBucket(nodeId)
+      PaymentNodeOutAmount.withoutTags().record(bucket, amount.truncateToSatoshi.toLong)
+      PaymentNodeOut.withoutTags().record(bucket)
+    }
   }
 
   object Tags {