Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
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
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment