use kamon for move latency monitoring

conf/base.conf

@@ -332,7 +332,6 @@ monitor {
   actor.name = monitor-actor
   socket.name = monitor-socket
   socket.uid.ttl = ${site.socket.uid.ttl}
-  channel.move_lat = monitor-move-lat
 }
 lobby {
   message.ttl = 30 seconds
@@ -421,6 +420,7 @@ round {
     timeout = 30 seconds
   }
   actor.map.name = round-map
+  channel.move_time.name = round-move-time
   collection {
     note = game_note
     history = round_history
@@ -651,7 +651,7 @@ hub {
     hub = ${socket.hub.name}
   }
   channel {
-    move_lat = ${monitor.channel.move_lat}
+    move_time = ${round.channel.move_time.name}
   }
 }
 
@@ -672,3 +672,28 @@ akka {
   log-config-on-start = off
   log-dead-letters-during-shutdown = off
 }
+
+kamon {
+  modules {
+    kamon-scala {
+      auto-start = no
+      requires-aspectj = no
+      extension-id = none
+    }
+    kamon-akka {
+      auto-start = no
+      requires-aspectj = no
+      extension-id = "kamon.akka.Akka"
+    }
+    kamon-play {
+      auto-start = no
+      requires-aspectj = no
+      extension-id = "kamon.play.Play"
+    }
+    kamon-statsd {
+      auto-start = no
+      requires-aspectj = no
+      extension-id = "kamon.statsd.StatsD"
+    }
+  }
+}

modules/hub/src/main/Env.scala

@@ -34,7 +34,7 @@ final class Env(config: Config, system: ActorSystem) {
   }
 
   object channel {
-    val moveLat = select("channel.move_lat")
+    val roundMoveTime = select("round.channel.move_time")
   }
 
   object socket {

modules/monitor/src/main/Env.scala

@@ -19,7 +19,6 @@ final class Env(
   private val ActorName = config getString "actor.name"
   private val SocketName = config getString "socket.name"
   private val SocketUidTtl = config duration "socket.uid.ttl"
-  private val ChannelMoveLat = config getString "channel.move_lat"
 
   private lazy val socket = system.actorOf(
     Props(new Socket(timeout = SocketUidTtl)), name = SocketName)
@@ -35,24 +34,12 @@ final class Env(
       hub = hub
     )), name = ActorName)
 
-  private val moveLatChannel = system.actorOf(Props(new Channel), name = ChannelMoveLat)
-
   {
     import scala.concurrent.duration._
 
     scheduler.message(1 seconds) {
       reporting -> lila.hub.actorApi.monitor.Update
     }
-
-    scheduler.effect(1 seconds, "move latency channel") {
-      import akka.pattern.ask
-      import makeTimeout.short
-      reporting ? actorApi.GetMoveLatency foreach {
-        case lat: Int => moveLatChannel ! Channel.Publish(
-          lila.socket.Socket.makeMessage("mlat", lat)
-        )
-      }
-    }
   }
 
   // requests per second

modules/round/src/main/Env.scala

@@ -49,9 +49,14 @@ final class Env(
     val CollectionNote = config getString "collection.note"
     val CollectionHistory = config getString "collection.history"
     val CollectionForecast = config getString "collection.forecast"
+    val ChannelMoveTime = config getString "channel.move_time.name "
   }
   import settings._
 
+  private val moveTimeChannel = system.actorOf(Props(classOf[lila.socket.Channel]), name = ChannelMoveTime)
+
+  private val moveMonitor = new MoveMonitor(system, moveTimeChannel)
+
   lazy val eventHistory = History(db(CollectionHistory)) _
 
   val roundMap = system.actorOf(Props(new lila.hub.ActorMap {
@@ -65,7 +70,7 @@ final class Env(
       drawer = drawer,
       forecastApi = forecastApi,
       socketHub = socketHub,
-      monitorMove = (ms: Option[Int]) => hub.actor.monitor ! lila.hub.actorApi.monitor.Move(ms),
+      monitorMove = moveMonitor.record,
       moretimeDuration = Moretime,
       activeTtl = ActiveTtl)
     def receive: Receive = ({

modules/round/src/main/MoveMonitor.scala

@@ -0,0 +1,33 @@
+package lila.round
+
+import akka.actor._
+import kamon._
+import metric.SubscriptionsDispatcher.TickMetricSnapshot
+
+private final class MoveMonitor(
+    system: ActorSystem,
+    channel: ActorRef) {
+
+  private val timeName = "round.move.time"
+  private val countName = "round.move.count"
+
+  private val time = Kamon.metrics.histogram(timeName)
+  private val count = Kamon.metrics.counter(countName)
+
+  def record(ms: Option[Int]) = {
+    ms foreach { x => time.record(x) }
+    count.increment()
+  }
+
+  Kamon.metrics.subscribe("histogram", timeName, system.actorOf(Props(new Actor {
+    def receive = {
+      case tick: TickMetricSnapshot => tick.metrics.collectFirst {
+        case (entity, snapshot) if entity.category == "histogram" => snapshot
+      } flatMap (_ histogram "histogram") foreach { h =>
+        if (!h.isEmpty) channel ! lila.socket.Channel.Publish(
+          lila.socket.Socket.makeMessage("mlat", (h.sum / h.numberOfMeasurements).toInt)
+        )
+      }
+    }
+  })))
+}

modules/socket/src/main/Handler.scala

@@ -36,7 +36,7 @@ object Handler {
       case ("startWatching", o) => o str "d" foreach { ids =>
         hub.actor.moveBroadcast ! StartWatching(uid, member, ids.split(' ').toSet)
       }
-      case ("moveLat", o) => hub.channel.moveLat ! (~(o boolean "d")).fold(
+      case ("moveLat", o) => hub.channel.roundMoveTime ! (~(o boolean "d")).fold(
         Channel.Sub(member),
         Channel.UnSub(member))
       case ("anaMove", o) => AnaRateLimit(uid) {

project/Build.scala

@@ -186,8 +186,9 @@ object ApplicationBuild extends Build {
   )
 
   lazy val round = project("round", Seq(
-    common, db, memo, hub, socket, chess, game, user, i18n, ai, pref, chat, history, playban)).settings(
-    libraryDependencies ++= provided(play.api, RM, PRM, hasher)
+    common, db, memo, hub, socket, chess, game, user,
+    i18n, ai, pref, chat, history, playban)).settings(
+    libraryDependencies ++= provided(play.api, RM, PRM, hasher, kamon.core)
   )
 
   lazy val lobby = project("lobby", Seq(