<commit_before>package se.chimps.bitziness.core.services.metrics.adapters

import akka.actor.{ActorLogging, ActorRef}
import akka.http.scaladsl.model.headers.Host
import akka.http.scaladsl.model.{HttpResponse, Uri, HttpMethods, HttpRequest}
import se.chimps.bitziness.core.endpoints.http.{ConnectionBuilder, HttpClientEndpoint}
import se.chimps.bitziness.core.generic.ActorFactory
import se.chimps.bitziness.core.services.metrics.MetricService

class InfluxDbMetricsService(val settings:InfluxSettings, override val host:String) extends MetricService {

  var influxdb:ActorRef = _

  override def metric(host:String, service: String, name: String, metric: Long, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"$name,host=$host,service=$service${convert(state)}${meta(metadata)} value=${metric}i"
  }

  override def metric(host:String, service: String, name: String, metric: BigDecimal, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"$name,host=$host,service=$service${convert(state)}${meta(metadata)} value=${metric.toString()}"
  }

  override def metric(host:String, service: String, name: String, metric: Boolean, state: Option[String], metadata: Map[String, String]): Unit = {
    val value = metric match {
      case true => "t"
      case _ => "f"
    }

    influxdb ! s"$name,host=$host,service=$service${convert(state)}${meta(metadata)} value=${value}"
  }

  override def metric(host:String, service: String, name: String, metric: String, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"""$name,host=$host,service=$service${convert(state)}${meta(metadata)} value=\"${metric}\""""
  }

  override def initialize(): Unit = {
    super.initialize()
    val factory = ActorFactory(s"influxdb-${settings.host}:${settings.port}", () => new InfluxDbEndpoint(self, settings))
    influxdb = initEndpoint(factory)
  }

  def meta(metadata:Map[String, String]):String = {
    val meta = metadata.map(m => s"${m._1}=${m._2}").mkString(",")

    if (meta.length > 0) {
      s",$meta"
    } else {
      ""
    }
  }

  def convert(state:Option[String]):String = {
    state.map(s => s",state=$s").getOrElse("")
  }
}

case class InfluxSettings(host:String, port:Int, db:String, username:Option[String], password:Option[String], secure:Boolean = false, buffer:Int = 100)

class InfluxDbEndpoint(val service:ActorRef, val settings:InfluxSettings) extends HttpClientEndpoint {
  var buffer:Seq[String] = Seq()

  // TODO implement a send every X second feature.

  override def setupConnection(builder: ConnectionBuilder): ActorRef = {
    builder.host(settings.host, settings.port).build(settings.secure)
  }

  def uri:String = s"/write?db=${settings.db}${settings.username.map(u => s"&u=$u").getOrElse("")}${settings.password.map(p => s"&p=$p").getOrElse("")}"

  override def receive: Receive = {
    case s:String => bufferAndSend(s)
    case res:HttpResponse =>
  }

  def bufferAndSend(metric:String) = {
    if (buffer.size + 1 >= settings.buffer) {
      val data = (buffer ++ Seq(metric)).mkString("\n")
      send(HttpRequest(HttpMethods.POST, uri = Uri(uri), entity = data))
      buffer = Seq()
    } else {
      buffer = buffer ++ Seq(metric)
    }
  }
}
<commit_msg>Add escsaping to Influxdb metric service..<commit_after>package se.chimps.bitziness.core.services.metrics.adapters

import akka.actor.{ActorLogging, ActorRef}
import akka.http.scaladsl.model.headers.Host
import akka.http.scaladsl.model.{HttpResponse, Uri, HttpMethods, HttpRequest}
import se.chimps.bitziness.core.endpoints.http.{ConnectionBuilder, HttpClientEndpoint}
import se.chimps.bitziness.core.generic.ActorFactory
import se.chimps.bitziness.core.services.metrics.MetricService

class InfluxDbMetricsService(val settings:InfluxSettings, override val host:String) extends MetricService {

  var influxdb:ActorRef = _

  override def metric(host:String, service: String, name: String, metric: Long, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"${escape(name)},host=${escape(host)},service=${escape(service)}${convert(state)}${meta(metadata)} value=${metric}i"
  }

  override def metric(host:String, service: String, name: String, metric: BigDecimal, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"${escape(name)},host=${escape(host)},service=${escape(service)}${convert(state)}${meta(metadata)} value=${metric.toString()}"
  }

  override def metric(host:String, service: String, name: String, metric: Boolean, state: Option[String], metadata: Map[String, String]): Unit = {
    val value = metric match {
      case true => "t"
      case _ => "f"
    }

    influxdb ! s"${escape(name)},host=${escape(host)},service=${escape(service)}${convert(state)}${meta(metadata)} value=$value"
  }

  override def metric(host:String, service: String, name: String, metric: String, state: Option[String], metadata: Map[String, String]): Unit = {
    influxdb ! s"""${escape(name)},host=${escape(host)},service=${escape(service)}${convert(state)}${meta(metadata)} value=\"${metric}\""""
  }

  override def initialize(): Unit = {
    super.initialize()
    val factory = ActorFactory(s"influxdb-${settings.host}:${settings.port}", () => new InfluxDbEndpoint(self, settings))
    influxdb = initEndpoint(factory)
  }

  def meta(metadata:Map[String, String]):String = {
    val meta = metadata.map(m => s"${m._1}=${escape(m._2)}").mkString(",")

    if (meta.length > 0) {
      s",$meta"
    } else {
      ""
    }
  }

  def convert(state:Option[String]):String = {
    state.map(s => s",state=${escape(s)}").getOrElse("")
  }

  def escape(str:String):String = str.replace(" ", "\\ ")
}

case class InfluxSettings(host:String, port:Int, db:String, username:Option[String], password:Option[String], secure:Boolean = false, buffer:Int = 100)

class InfluxDbEndpoint(val service:ActorRef, val settings:InfluxSettings) extends HttpClientEndpoint {
  var buffer:Seq[String] = Seq()

  // TODO implement a send every X second feature.

  override def setupConnection(builder: ConnectionBuilder): ActorRef = {
    builder.host(settings.host, settings.port).build(settings.secure)
  }

  def uri:String = s"/write?db=${settings.db}${settings.username.map(u => s"&u=$u").getOrElse("")}${settings.password.map(p => s"&p=$p").getOrElse("")}"

  override def receive: Receive = {
    case s:String => bufferAndSend(s)
    case res:HttpResponse =>
  }

  def bufferAndSend(metric:String) = {
    if (buffer.size + 1 >= settings.buffer) {
      val data = (buffer ++ Seq(metric)).mkString("\n")
      send(HttpRequest(HttpMethods.POST, uri = Uri(uri), entity = data))
      buffer = Seq()
    } else {
      buffer = buffer ++ Seq(metric)
    }
  }
}

{'fraction_non_alphanumeric': 0.1672561155319776, 'fraction_numerical': 0.00206307102858827, 'mean_word_length': 5.000884173297966, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, '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': '1224770', 'n_tokens_mistral': 2203, 'n_tokens_neox': 2107, 'n_words': 428}