<commit_before>package mediamath.metrics

import com.codahale.metrics._
import com.fasterxml.jackson.databind.ObjectMapper
import dispatch._
import Defaults._
import java.util.concurrent.TimeUnit
import scala.collection
import scala.collection.JavaConversions._
import java.net.{Inet4Address, NetworkInterface}
import collection._
import java.util.{SortedMap => JavaSortedMap}
import scala.language.existentials
import scala.collection.immutable.ListMap
import org.slf4j.LoggerFactory
import scala.concurrent.Await
import scala.concurrent.duration._

object QasinoReporter {

  val DEFAULT_PORT = 15597

  val defaultColumnsTypes = Seq[String](
    "string" // host
  )

  val defaultColumnsNames = Seq[String](
    "host" // host
  )

  val defaultColumnsValues = Array[Any](
    getFirstNonLoopbackAddress // host
  )

	val registryNameSeparator = "_"
	val illegalCharRegex = new scala.util.matching.Regex("""[^A-Za-z0-9_]""")

	def sanitizeString(name: String): String = {
		// Remove any instances of the illegal characters from the name
		illegalCharRegex.replaceAllIn(name.toLowerCase, registryNameSeparator)
	}

	def sanitizeRegistry(registry: MetricRegistry): MetricRegistry = {
		// Return a new MetricRegistry with names sanitized for qasino
		val sanitizedRegistry = new MetricRegistry
		val metricMap = mapAsScalaMap(registry.getMetrics)
		for ((name, metric) <- metricMap) {
			sanitizedRegistry.register(sanitizeString(name), metric)
		}
		sanitizedRegistry
	}

	def getFirstNonLoopbackAddress: String = {
		val interfaces = NetworkInterface.getNetworkInterfaces
		val nonLoopbackInterfaces = interfaces.filter(!_.isLoopback)
		val default = "unavailable"
		var address = default
		for (i <- nonLoopbackInterfaces if address == default) {
			for (addr <- i.getInetAddresses if addr.isInstanceOf[Inet4Address] && address == default) {
				address = addr.getHostAddress
			}
		}
		address
	}

  def forRegistry(registry: MetricRegistry): Builder.type = {
    Builder.registry = registry
    Builder
  }

  object Builder {
    private[metrics] var registry: MetricRegistry = new MetricRegistry
    private[metrics] var host: String = "localhost"
    private[metrics] var port: Int = DEFAULT_PORT
    private[metrics] var secure: Boolean = false
    private[metrics] var uri: String = "request"
    private[metrics] var db_op: String = "add_table_data"
    private[metrics] var name: String = "QasinoReporter"
    private[metrics] var db_persist: Boolean = false
    private[metrics] var groupings: Set[String] = SortedSet.empty
    private[metrics] var filter: MetricFilter = MetricFilter.ALL
    private[metrics] var rateUnit: TimeUnit = TimeUnit.SECONDS
    private[metrics] var durationUnit: TimeUnit = TimeUnit.MILLISECONDS

    def withPort(port: Int): this.type = {
      Builder.port = port
      this
    }

    def withHost(host: String): this.type = {
      Builder.host = host
      this
    }

    def withSecure(secure: Boolean = true): this.type = {
      Builder.secure = secure
      this
    }

    def withUri(uri: String): this.type = {
      Builder.uri = uri
      this
    }

    def withOp(db_op: String): this.type = {
      Builder.db_op = db_op
      this
    }

    def withName(name: String): this.type = {
      Builder.name = name
      this
    }

    def withPersist(db_persist: Boolean = true): this.type = {
      Builder.db_persist = db_persist
      this
    }

    def withGroupings(groupings: Set[String]): this.type = {
      Builder.groupings = groupings.map {sanitizeString}
      this
    }

    def withFilter(filter: MetricFilter): this.type = {
      Builder.filter = filter
      this
    }

    def convertRatesTo(rateUnit: TimeUnit): this.type = {
      Builder.rateUnit = rateUnit
      this
    }

    def convertDurationsTo(durationUnit: TimeUnit): this.type = {
      Builder.durationUnit = durationUnit
      this
    }

    def build(): QasinoReporter = {
      new QasinoReporter
    }
  }

  private def registryHasCollisions(registry: MetricRegistry): Boolean = {
    // Check whether we have any name collisions after some sanitizing
    val namesSet = mutable.Set[String]()
    val registryNames = registry.getNames
    for (name <- asScalaSet(registryNames)) {
      val sanitizedName = sanitizeString(name)
      namesSet.add(sanitizedName)
    }
    namesSet.size < registryNames.size()
  }

  private def hasIllegalColumnNames(registry: MetricRegistry): Boolean = {
    var hasIllegalColName = false
    for (name <- registry.getNames if !hasIllegalColName) {
      val thisGrouping: Option[String] =
        Builder.groupings.toSeq.sortBy(_.length).reverse.find(s => name.startsWith(s + "_"))
      val suffix: String = if (thisGrouping.isDefined) {
        name.drop(thisGrouping.get.length + 1)
      }
      else name
      hasIllegalColName = suffix.matches("^[^A-Za-z].*")
    }
    hasIllegalColName
  }
}
import QasinoReporter._

class QasinoReporter extends
		ScheduledReporter(
      Builder.registry,
      Builder.name,
      Builder.filter,
      Builder.rateUnit,
      Builder.durationUnit) {
	val registry: MetricRegistry = Builder.registry
	val host: String = Builder.host
	val port: Int = Builder.port
	val secure: Boolean = Builder.secure
	val uri: String = Builder.uri
	val db_op: String = Builder.db_op
	val db_persist: Boolean = Builder.db_persist
	val name: String = Builder.name
	val groupings: Set[String] = Builder.groupings
	val filter: MetricFilter = Builder.filter
	val rateUnit: TimeUnit = Builder.rateUnit
	val durationUnit: TimeUnit = Builder.durationUnit

  // Set up logger
  val log = LoggerFactory.getLogger(this.getClass)

  // Set up Dispatch HTTP client
	private val dispatchHost = if (secure) dispatch.host(host, port).secure else dispatch.host(host, port)
	private val dispatchRequest = (dispatchHost / uri).POST <<? Map("op" -> db_op)

	// JSON mapper singleton
	private val mapper = new ObjectMapper()

	object QasinoRequestIdentifier extends scala.Enumeration {
		// Enumeration for all the JSON keys for qasino for safety
		type QasinoRequestIdentifier = Value
		val op, identity, tablename, table, column_names, column_types, rows, persist = Value
	}
	import QasinoRequestIdentifier._

	// Default map for JSON
	private[this] val db_persist_int = if (db_persist) 1 else 0
	private val defaultDataJson = mutable.Map[String, Any](
		op.toString -> db_op,
		identity.toString -> getFirstNonLoopbackAddress,
		persist.toString -> db_persist_int,<commit_msg>SONIC-1722 added withUsername and withPassword for secure connection<commit_after>package mediamath.metrics

import com.codahale.metrics._
import com.fasterxml.jackson.databind.ObjectMapper
import dispatch._
import Defaults._
import java.util.concurrent.TimeUnit
import scala.collection
import scala.collection.JavaConversions._
import java.net.{Inet4Address, NetworkInterface}
import collection._
import java.util.{SortedMap => JavaSortedMap}
import scala.language.existentials
import scala.collection.immutable.ListMap
import org.slf4j.LoggerFactory
import scala.concurrent.Await
import scala.concurrent.duration._

object QasinoReporter {

  val DEFAULT_PORT = 15597

  val defaultColumnsTypes = Seq[String](
    "string" // host
  )

  val defaultColumnsNames = Seq[String](
    "host" // host
  )

  val defaultColumnsValues = Array[Any](
    getFirstNonLoopbackAddress // host
  )

	val registryNameSeparator = "_"
	val illegalCharRegex = new scala.util.matching.Regex("""[^A-Za-z0-9_]""")

	def sanitizeString(name: String): String = {
		// Remove any instances of the illegal characters from the name
		illegalCharRegex.replaceAllIn(name.toLowerCase, registryNameSeparator)
	}

	def sanitizeRegistry(registry: MetricRegistry): MetricRegistry = {
		// Return a new MetricRegistry with names sanitized for qasino
		val sanitizedRegistry = new MetricRegistry
		val metricMap = mapAsScalaMap(registry.getMetrics)
		for ((name, metric) <- metricMap) {
			sanitizedRegistry.register(sanitizeString(name), metric)
		}
		sanitizedRegistry
	}

	def getFirstNonLoopbackAddress: String = {
		val interfaces = NetworkInterface.getNetworkInterfaces
		val nonLoopbackInterfaces = interfaces.filter(!_.isLoopback)
		val default = "unavailable"
		var address = default
		for (i <- nonLoopbackInterfaces if address == default) {
			for (addr <- i.getInetAddresses if addr.isInstanceOf[Inet4Address] && address == default) {
				address = addr.getHostAddress
			}
		}
		address
	}

  def forRegistry(registry: MetricRegistry): Builder.type = {
    Builder.registry = registry
    Builder
  }

  object Builder {
    private[metrics] var registry: MetricRegistry = new MetricRegistry
    private[metrics] var host: String = "localhost"
    private[metrics] var port: Int = DEFAULT_PORT
    private[metrics] var secure: Boolean = false
    private[metrics] var username: String = ""
    private[metrics] var password: String = ""
    private[metrics] var uri: String = "request"
    private[metrics] var db_op: String = "add_table_data"
    private[metrics] var name: String = "QasinoReporter"
    private[metrics] var db_persist: Boolean = false
    private[metrics] var groupings: Set[String] = SortedSet.empty
    private[metrics] var filter: MetricFilter = MetricFilter.ALL
    private[metrics] var rateUnit: TimeUnit = TimeUnit.SECONDS
    private[metrics] var durationUnit: TimeUnit = TimeUnit.MILLISECONDS

    def withPort(port: Int): this.type = {
      Builder.port = port
      this
    }

    def withHost(host: String): this.type = {
      Builder.host = host
      this
    }

    def withSecure(secure: Boolean = true): this.type = {
      Builder.secure = secure
      this
    }

    def withUsername(username: String): this.type = {
      Builder.username = username
      this
    }

    def withPassword(password: String): this.type = {
      Builder.password = password
      this
    }

    def withUri(uri: String): this.type = {
      Builder.uri = uri
      this
    }

    def withOp(db_op: String): this.type = {
      Builder.db_op = db_op
      this
    }

    def withName(name: String): this.type = {
      Builder.name = name
      this
    }

    def withPersist(db_persist: Boolean = true): this.type = {
      Builder.db_persist = db_persist
      this
    }

    def withGroupings(groupings: Set[String]): this.type = {
      Builder.groupings = groupings.map {sanitizeString}
      this
    }

    def withFilter(filter: MetricFilter): this.type = {
      Builder.filter = filter
      this
    }

    def convertRatesTo(rateUnit: TimeUnit): this.type = {
      Builder.rateUnit = rateUnit
      this
    }

    def convertDurationsTo(durationUnit: TimeUnit): this.type = {
      Builder.durationUnit = durationUnit
      this
    }

    def build(): QasinoReporter = {
      new QasinoReporter
    }
  }

  private def registryHasCollisions(registry: MetricRegistry): Boolean = {
    // Check whether we have any name collisions after some sanitizing
    val namesSet = mutable.Set[String]()
    val registryNames = registry.getNames
    for (name <- asScalaSet(registryNames)) {
      val sanitizedName = sanitizeString(name)
      namesSet.add(sanitizedName)
    }
    namesSet.size < registryNames.size()
  }

  private def hasIllegalColumnNames(registry: MetricRegistry): Boolean = {
    var hasIllegalColName = false
    for (name <- registry.getNames if !hasIllegalColName) {
      val thisGrouping: Option[String] =
        Builder.groupings.toSeq.sortBy(_.length).reverse.find(s => name.startsWith(s + "_"))
      val suffix: String = if (thisGrouping.isDefined) {
        name.drop(thisGrouping.get.length + 1)
      }
      else name
      hasIllegalColName = suffix.matches("^[^A-Za-z].*")
    }
    hasIllegalColName
  }
}
import QasinoReporter._

class QasinoReporter extends
		ScheduledReporter(
      Builder.registry,
      Builder.name,
      Builder.filter,
      Builder.rateUnit,
      Builder.durationUnit) {
	val registry: MetricRegistry = Builder.registry
	val host: String = Builder.host
	val port: Int = Builder.port
	val secure: Boolean = Builder.secure
  val username: String = Builder.username
  val password: String = Builder.password
	val uri: String = Builder.uri
	val db_op: String = Builder.db_op
	val db_persist: Boolean = Builder.db_persist
	val name: String = Builder.name
	val groupings: Set[String] = Builder.groupings
	val filter: MetricFilter = Builder.filter
	val rateUnit: TimeUnit = Builder.rateUnit
	val durationUnit: TimeUnit = Builder.durationUnit

  // Set up logger
  val log = LoggerFactory.getLogger(this.getClass)

  // Set up Dispatch HTTP client
	private val dispatchHost = if (secure)
    dispatch.host(host, port).secure
  else
    dispatch.host(host, port)
	private val dispatchRequest = if (secure)
    (dispatchHost / uri).POST <<? Map("op" -> db_op) as_!(username, password)
  else
    (dispatchHost / uri).POST <<? Map("op" -> db_op)

	// JSON mapper singleton
	private val mapper = new ObjectMapper()

	object QasinoRequestIdentifier extends scala.Enumeration {
		// Enumeration for all the JSON keys for qasino for safety
		type QasinoRequestIdentifier = Value
		val op, identity, tablename, table, column_names, column_types, rows, persist = Value
	}
	import QasinoRequestIdentifier._

	// Default map for JSON
	private[this] val db_persist_int = if (db_persist) 1 else 0
	private val defaultDataJson = mutable.Map[String, Any](
		op.toString -> db_op,
		identity.toString -> getFirstNonLoopbackAddress,
		persist.toString -> db_persist_int,
{'fraction_non_alphanumeric': 0.09389810426540285, 'fraction_numerical': 0.002221563981042654, 'mean_word_length': 3.5318791946308723, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3232365', 'n_tokens_mistral': 4218, 'n_tokens_neox': 4045, 'n_words': 1037}