ClusterMetricsElasticScaleListener.scala

class ClusterMetricsElasticScaleListener extends Actor with ActorLogging {
  val selfAddress = Cluster(context.system).selfAddress

  override def preStart(): Unit =
    Cluster(context.system).subscribe(self, classOf[ClusterMetricsChanged])

  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {
    case ClusterMetricsChanged(metrics) ⇒
      metrics filter (_.address == selfAddress) foreach { nodeMetrics ⇒
        elasticScaleForHeap(nodeMetrics)
        elasticScaleForCpu(nodeMetrics)
      }
    case state: CurrentClusterState ⇒ // ignore
  }

  def elasticScaleForHeap(metrics: NodeMetrics): Unit = metrics match {
    case HeapMemory(address, timestamp, used, committed, max) ⇒
      log.info("Used heap: {} MB", used.doubleValue / 1024 / 1024)
      // pipeTo elasticScale API
    case _ ⇒ // complete head data not available
  }

  def elasticScaleForCpu(metrics: NodeMetrics): Unit = metrics match {
    case Cpu(address, timestamp, Some(systemLoadAverage), cpuCombined, processors) ⇒
      log.info("Load: {} ({} processors)", systemLoadAverage, processors)
      // pipeTo elasticScale API
    case _ ⇒ // complete cpu data not available
  }
}