<commit_before>package main

import java.time.{Instant, Clock, ZonedDateTime}

import akka.actor.{ActorLogging, Props}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.metrics.StandardMetrics.{Cpu, HeapMemory}
import akka.cluster.metrics.{ClusterMetricsChanged, ClusterMetricsExtension}
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request, SubscriptionTimeoutExceeded}
import akka.util.ByteString

import scala.annotation.tailrec
import scala.collection.mutable

object ClusterMetrics {
  def props(cluster: Cluster) = Props(new ClusterMetrics(cluster))
}

class ClusterMetrics(cluster: Cluster) extends ActorPublisher[ByteString] with ActorLogging {
  val divider = 1024 * 1024
  val extension = ClusterMetricsExtension(context.system)

  private val queue = mutable.Queue[ByteString]()

  override def preStart() = extension.subscribe(self)

  override def postStop() = extension.unsubscribe(self)


  import spray.json._
  import DefaultJsonProtocol._

  import scala.collection.JavaConverters._

  override def receive = {
    case state: CurrentClusterState =>
      log.info(s"Leader Node: {}", state.getLeader)
    case ClusterMetricsChanged(clusterMetrics) =>
      clusterMetrics.foreach {
        /*val metrics = m.getMetrics.asScala.foldLeft(Map("node" -> m.address.toString, "ts" -> m.timestamp.toString)) { (acc, m) =>
          acc + (m.name -> m.value.toString)
        }
        queue.enqueue(ByteString(metrics.toJson.prettyPrint))*/
        case HeapMemory(address, timestamp, used, committed, max) =>
          //log.info("Used heap: {} mb", used.doubleValue / divider)
          //Timestamp
          //Instant.ofEpochMilli()
          //ZonedDateTime
          val metrics = Map("node" -> address.toString, "metric" -> "heap", "when" -> timestamp.toString,
            "used" -> (used.doubleValue / divider).toString, "committed" -> committed.toString, "max" -> max.toString)
          queue.enqueue(ByteString(metrics.toJson.prettyPrint))
        case Cpu(address, timestamp, Some(systemLoadAverage), cpuCombined, cpuStolen, processors) =>
          log.info("Load: {} ({} processors)", systemLoadAverage, processors)
          val metrics = Map("node" -> address.toString, "metric" -> "cpu",
            "when" -> timestamp.toString, "avr" -> systemLoadAverage.toString,
            "cpuCombined" -> cpuCombined.toString, "cpu-stolen" -> cpuStolen.toString, "processors" -> processors.toString)
          queue.enqueue(ByteString(metrics.toJson.prettyPrint))
        case other =>
          log.info("metric name: {}", other.getClass.getName)
      }
      tryToReply

    case req @ Request(n) ⇒
      tryToReply
    case SubscriptionTimeoutExceeded ⇒
      log.info("canceled")
      (context stop self)<commit_msg>a try to change port<commit_after>package main

import java.time.format.DateTimeFormatter
import java.time.{ZoneOffset, Instant, Clock, ZonedDateTime}
import java.util.Date

import akka.actor.{ActorLogging, Props}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.metrics.StandardMetrics.{Cpu, HeapMemory}
import akka.cluster.metrics.{ClusterMetricsChanged, ClusterMetricsExtension}
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.{Cancel, Request, SubscriptionTimeoutExceeded}
import akka.util.ByteString

import scala.annotation.tailrec
import scala.collection.mutable

object ClusterMetrics {
  def props(cluster: Cluster) = Props(new ClusterMetrics(cluster))
}

class ClusterMetrics(cluster: Cluster) extends ActorPublisher[ByteString] with ActorLogging {
  val divider = 1024 * 1024
  val extension = ClusterMetricsExtension(context.system)

  val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss Z")

  private val queue = mutable.Queue[ByteString]()

  override def preStart() = extension.subscribe(self)

  override def postStop() = extension.unsubscribe(self)

  import spray.json._
  import DefaultJsonProtocol._
  import scala.collection.JavaConverters._

  override def receive = {
    case state: CurrentClusterState =>
      log.info(s"Leader Node: {}", state.getLeader)
    case ClusterMetricsChanged(clusterMetrics) =>
      clusterMetrics.foreach {
        case HeapMemory(address, timestamp, used, committed, max) =>
          //JsNumber(timestamp),
          val json = JsObject(Map("node" -> JsString(address.toString),
              "metric" -> JsString("heap"),
              "when" -> JsString(formatter.format(ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneOffset.UTC))),
              "used" -> JsString((used.doubleValue / divider).toString + " mb"),
              "max" -> JsString((max.getOrElse(0l) / divider).toString + " mb"))).prettyPrint
          queue.enqueue(ByteString(json))
        case other =>
          log.info("metric name: {}", other.getClass.getName)
      }
      tryToReply

    case req @ Request(n) ⇒
      tryToReply
    case SubscriptionTimeoutExceeded ⇒
      log.info("canceled")
      (context stop self)
{'fraction_non_alphanumeric': 0.11623363938269193, 'fraction_numerical': 0.0033209611252197693, 'mean_word_length': 3.807511737089202, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2538334', 'n_tokens_mistral': 1525, 'n_tokens_neox': 1452, 'n_words': 308}