<commit_before>package io.vamp.pulse

import java.time.OffsetDateTime
import java.time.format.DateTimeFormatter

import io.vamp.common.{ ClassMapper, Config, Namespace }
import io.vamp.common.http.OffsetEnvelope
import io.vamp.common.json.{ OffsetDateTimeSerializer, SerializationFormat }
import io.vamp.common.util.HashUtil
import io.vamp.common.vitals.{ InfoRequest, StatsRequest }
import io.vamp.model.event.Aggregator.AggregatorType
import io.vamp.model.event._
import io.vamp.model.resolver.NamespaceValueResolver
import io.vamp.pulse.Percolator.{ GetPercolator, RegisterPercolator, UnregisterPercolator }
import io.vamp.pulse.notification._
import org.json4s.ext.EnumNameSerializer
import org.json4s.native.Serialization.{ read, write }
import org.json4s.{ DefaultFormats, Extraction }

import scala.concurrent.Future

class ElasticsearchPulseActorMapper extends ClassMapper {
  val name = "elasticsearch"
  val clazz = classOf[ElasticsearchPulseActor]
}

object ElasticsearchPulseActor {

  val config = PulseActor.config

  val elasticsearchUrl = Config.string(s"$config.elasticsearch.url")

  val indexName = Config.string(s"$config.elasticsearch.index.name")

  def indexTimeFormat()(implicit namespace: Namespace): Map[String, String] = {
    Config.entries(s"$config.elasticsearch.index.time-format")().map { case (key, value) ⇒ key → value.toString }
  }
}

class ElasticsearchPulseActor extends ElasticsearchPulseEvent with NamespaceValueResolver with PulseStats with PulseActor {

  import ElasticsearchClient._
  import PulseActor._

  lazy val url = ElasticsearchPulseActor.elasticsearchUrl()

  lazy val indexName = resolveWithNamespace(ElasticsearchPulseActor.indexName(), lookup = true)

  lazy val indexTimeFormat = ElasticsearchPulseActor.indexTimeFormat()

  private lazy val es = new ElasticsearchClient(url)

  def receive = {

    case InfoRequest ⇒ reply(info)

    case StatsRequest ⇒ reply(stats)

    case Publish(event, publishEventValue) ⇒ reply((validateEvent andThen publish(publishEventValue) andThen broadcast(publishEventValue))(Event.expandTags(event)), classOf[EventIndexError])

    case Query(envelope) ⇒ reply((validateEventQuery andThen eventQuery(envelope.page, envelope.perPage))(envelope.request), classOf[EventQueryError])

    case GetPercolator(name) ⇒ reply(Future.successful(getPercolator(name)))

    case RegisterPercolator(name, tags, kind, message) ⇒ registerPercolator(name, tags, kind, message)

    case UnregisterPercolator(name) ⇒ unregisterPercolator(name)

    case any ⇒ unsupported(UnsupportedPulseRequest(any))
  }

  private def info = es.health map { health ⇒ Map[String, Any]("type" → "elasticsearch", "elasticsearch" → health) }

  private def publish(publishEventValue: Boolean)(event: Event): Future[Any] = {
    implicit val formats = SerializationFormat(OffsetDateTimeSerializer, new EnumNameSerializer(Aggregator))
    val (indexName, typeName) = indexTypeName(event.`type`)
    log.debug(s"Pulse publish an event to index '$indexName/$typeName': ${event.tags}")

    val attachment = (publishEventValue, event.value) match {
      case (true, str: String) ⇒ Map(typeName → str)
      case (true, any)         ⇒ Map("value" → write(any)(DefaultFormats), typeName → (if (typeName == Event.defaultType) "" else any))
      case (false, _)          ⇒ Map("value" → "")
    }

    val data = Extraction.decompose(if (publishEventValue) event else event.copy(value = None)) merge Extraction.decompose(attachment)

    es.index[ElasticsearchIndexResponse](indexName, typeName, data) map {
      case r: ElasticsearchIndexResponse ⇒ event.copy(id = Option(convertId(r._id)))
      case other ⇒
        log.error(s"Unexpected index result: ${other.toString}.")
        other
    }
  }

  private def broadcast(publishEventValue: Boolean): (Future[Any] ⇒ Future[Any]) = _.map {
    case event: Event ⇒ percolate(publishEventValue)(event)
    case other        ⇒ other
  }

  protected def eventQuery(page: Int, perPage: Int)(query: EventQuery): Future[Any] = {
    log.debug(s"Pulse query: $query")
    query.aggregator match {
      case None                                    ⇒ getEvents(query, page, perPage)
      case Some(Aggregator(Aggregator.`count`, _)) ⇒ countEvents(query)
      case Some(Aggregator(aggregator, field))     ⇒ aggregateEvents(query, aggregator, field)
      case _                                       ⇒ throw new UnsupportedOperationException
    }
  }

  protected def getEvents(query: EventQuery, page: Int, perPage: Int) = {
    implicit val formats = SerializationFormat(OffsetDateTimeSerializer, new EnumNameSerializer(Aggregator))
    val (p, pp) = OffsetEnvelope.normalize(page, perPage, EventRequestEnvelope.maxPerPage)

    es.search[ElasticsearchSearchResponse](indexName, constructSearch(query, p, pp)) map {
      case ElasticsearchSearchResponse(hits) ⇒
        EventResponseEnvelope(hits.hits.flatMap(hit ⇒ Option(read[Event](write(hit._source)).copy(id = Option(convertId(hit._id))))), hits.total, p, pp)
      case other ⇒ reportException(EventQueryError(other))
    }
  }

  protected def countEvents(eventQuery: EventQuery): Future[Any] = es.count(indexName, constructQuery(eventQuery)) map {
    case ElasticsearchCountResponse(count) ⇒ LongValueAggregationResult(count)
    case other                             ⇒ reportException(EventQueryError(other))
  }

  private def constructSearch(eventQuery: EventQuery, page: Int, perPage: Int): Map[Any, Any] = {
    constructQuery(eventQuery) +
      ("from" → ((page - 1) * perPage)) +
      ("size" → perPage) +
      ("sort" → Map("timestamp" → Map("order" → "desc")))
  }

  private def constructQuery(eventQuery: EventQuery): Map[Any, Any] = {
    Map("query" →
      Map("filtered" →
        Map(
          "query" → Map("match_all" → Map()),
          "filter" → Map("bool" →
            Map("must" → List(constructTagQuery(eventQuery.tags), constructTypeQuery(eventQuery.`type`), constructTimeRange(eventQuery.timestamp)).filter(_.isDefined).map(_.get)))
        )))
  }

  private def constructTagQuery(tags: Set[String]): Option[List[Map[String, Any]]] = {
    if (tags.nonEmpty) Option((for (tag ← tags) yield Map("term" → Map("tags" → tag))).toList) else None
  }

  private def constructTypeQuery(`type`: Option[String]): Option[Map[String, Any]] = {
    if (`type`.nonEmpty) Option(Map("term" → Map("type" → `type`.get))) else None
  }

  private def constructTimeRange(timeRange: Option[TimeRange]): Option[Map[Any, Any]] = timeRange match {
    case Some(tr) ⇒
      val query = Map(
        "lt" → tr.lt,
        "lte" → tr.lte,
        "gt" → tr.gt,
        "gte" → tr.gte
      ).filter(_._2.isDefined).map { case (k, v) ⇒ k → v.get }
      if (query.isEmpty) None else Some(Map("range" → Map("timestamp" → query)))
<commit_msg>Upgrade elasticsearch client to 5.0<commit_after>package io.vamp.pulse

import java.time.OffsetDateTime
import java.time.format.DateTimeFormatter

import io.vamp.common.{ ClassMapper, Config, Namespace }
import io.vamp.common.http.OffsetEnvelope
import io.vamp.common.json.{ OffsetDateTimeSerializer, SerializationFormat }
import io.vamp.common.util.HashUtil
import io.vamp.common.vitals.{ InfoRequest, StatsRequest }
import io.vamp.model.event.Aggregator.AggregatorType
import io.vamp.model.event._
import io.vamp.model.resolver.NamespaceValueResolver
import io.vamp.pulse.Percolator.{ GetPercolator, RegisterPercolator, UnregisterPercolator }
import io.vamp.pulse.notification._
import org.json4s.ext.EnumNameSerializer
import org.json4s.native.Serialization.{ read, write }
import org.json4s.{ DefaultFormats, Extraction }

import scala.concurrent.Future

class ElasticsearchPulseActorMapper extends ClassMapper {
  val name = "elasticsearch"
  val clazz = classOf[ElasticsearchPulseActor]
}

object ElasticsearchPulseActor {

  val config = PulseActor.config

  val elasticsearchUrl = Config.string(s"$config.elasticsearch.url")

  val indexName = Config.string(s"$config.elasticsearch.index.name")

  def indexTimeFormat()(implicit namespace: Namespace): Map[String, String] = {
    Config.entries(s"$config.elasticsearch.index.time-format")().map { case (key, value) ⇒ key → value.toString }
  }
}

trait ElasticsearchPulseActor extends ElasticsearchPulseEvent with NamespaceValueResolver with PulseStats with PulseActor {

  import ElasticsearchClient._
  import PulseActor._

  lazy val url = ElasticsearchPulseActor.elasticsearchUrl()

  lazy val indexName = resolveWithNamespace(ElasticsearchPulseActor.indexName(), lookup = true)

  lazy val indexTimeFormat = ElasticsearchPulseActor.indexTimeFormat()

  private lazy val es = new ElasticsearchClient(url)

  def receive = {

    case InfoRequest ⇒ reply(info)

    case StatsRequest ⇒ reply(stats)

    case Publish(event, publishEventValue) ⇒ reply((validateEvent andThen publish(publishEventValue) andThen broadcast(publishEventValue))(Event.expandTags(event)), classOf[EventIndexError])

    case Query(envelope) ⇒ reply((validateEventQuery andThen eventQuery(envelope.page, envelope.perPage))(envelope.request), classOf[EventQueryError])

    case GetPercolator(name) ⇒ reply(Future.successful(getPercolator(name)))

    case RegisterPercolator(name, tags, kind, message) ⇒ registerPercolator(name, tags, kind, message)

    case UnregisterPercolator(name) ⇒ unregisterPercolator(name)

    case any ⇒ unsupported(UnsupportedPulseRequest(any))
  }

  private def info = es.health map { health ⇒ Map[String, Any]("type" → "elasticsearch", "elasticsearch" → health) }

  private def publish(publishEventValue: Boolean)(event: Event): Future[Any] = {
    implicit val formats = SerializationFormat(OffsetDateTimeSerializer, new EnumNameSerializer(Aggregator))
    val (indexName, typeName) = indexTypeName(event.`type`)
    log.debug(s"Pulse publish an event to index '$indexName/$typeName': ${event.tags}")

    val attachment = (publishEventValue, event.value) match {
      case (true, str: String) ⇒ Map(typeName → str)
      case (true, any)         ⇒ Map("value" → write(any)(DefaultFormats), typeName → (if (typeName == Event.defaultType) "" else any))
      case (false, _)          ⇒ Map("value" → "")
    }

    val data = Extraction.decompose(if (publishEventValue) event else event.copy(value = None)) merge Extraction.decompose(attachment)

    es.index[ElasticsearchIndexResponse](indexName, typeName, data) map {
      case r: ElasticsearchIndexResponse ⇒ event.copy(id = Option(convertId(r._id)))
      case other ⇒
        log.error(s"Unexpected index result: ${other.toString}.")
        other
    }
  }

  private def broadcast(publishEventValue: Boolean): (Future[Any] ⇒ Future[Any]) = _.map {
    case event: Event ⇒ percolate(publishEventValue)(event)
    case other        ⇒ other
  }

  protected def eventQuery(page: Int, perPage: Int)(query: EventQuery): Future[Any] = {
    log.debug(s"Pulse query: $query")
    query.aggregator match {
      case None                                    ⇒ getEvents(query, page, perPage)
      case Some(Aggregator(Aggregator.`count`, _)) ⇒ countEvents(query)
      case Some(Aggregator(aggregator, field))     ⇒ aggregateEvents(query, aggregator, field)
      case _                                       ⇒ throw new UnsupportedOperationException
    }
  }

  protected def getEvents(query: EventQuery, page: Int, perPage: Int) = {
    implicit val formats = SerializationFormat(OffsetDateTimeSerializer, new EnumNameSerializer(Aggregator))
    val (p, pp) = OffsetEnvelope.normalize(page, perPage, EventRequestEnvelope.maxPerPage)

    es.search[ElasticsearchSearchResponse](indexName, constructSearch(query, p, pp)) map {
      case ElasticsearchSearchResponse(hits) ⇒
        EventResponseEnvelope(hits.hits.flatMap(hit ⇒ Option(read[Event](write(hit._source)).copy(id = Option(convertId(hit._id))))), hits.total, p, pp)
      case other ⇒ reportException(EventQueryError(other))
    }
  }

  protected def countEvents(eventQuery: EventQuery): Future[Any] = es.count(indexName, constructQuery(eventQuery)) map {
    case ElasticsearchCountResponse(count) ⇒ LongValueAggregationResult(count)
    case other                             ⇒ reportException(EventQueryError(other))
  }

  private def constructSearch(eventQuery: EventQuery, page: Int, perPage: Int): Map[Any, Any] = {
    constructQuery(eventQuery) +
      ("from" → ((page - 1) * perPage)) +
      ("size" → perPage) +
      ("sort" → Map("timestamp" → Map("order" → "desc")))
  }

  private def constructQuery(eventQuery: EventQuery): Map[Any, Any] = {
    Map("query" →
      Map("bool" →
        Map(
          "must" → Map("match_all" → Map()),
          "filter" → Map("bool" →
            Map("must" → List(constructTagQuery(eventQuery.tags), constructTypeQuery(eventQuery.`type`), constructTimeRange(eventQuery.timestamp)).filter(_.isDefined).map(_.get)))
        )))
  }

  private def constructTagQuery(tags: Set[String]): Option[List[Map[String, Any]]] = {
    if (tags.nonEmpty) Option((for (tag ← tags) yield Map("term" → Map("tags" → tag))).toList) else None
  }

  private def constructTypeQuery(`type`: Option[String]): Option[Map[String, Any]] = {
    if (`type`.nonEmpty) Option(Map("term" → Map("type" → `type`.get))) else None
  }

  private def constructTimeRange(timeRange: Option[TimeRange]): Option[Map[Any, Any]] = timeRange match {
    case Some(tr) ⇒
      val query = Map(
        "lt" → tr.lt,
        "lte" → tr.lte,
        "gt" → tr.gt,
        "gte" → tr.gte
      ).filter(_._2.isDefined).map { case (k, v) ⇒ k → v.get }
      if (query.isEmpty) None else Some(Map("range" → Map("timestamp" → query)))

{'fraction_non_alphanumeric': 0.12257023399105113, 'fraction_numerical': 0.0008802171202229883, 'mean_word_length': 4.187975646879757, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '831071', 'n_tokens_mistral': 4127, 'n_tokens_neox': 4169, 'n_words': 1006}