//设置package包名称以及导入依赖的类
package config

import java.net.InetSocketAddress
import java.util.concurrent.TimeUnit._
import com.codahale.metrics.graphite.{Graphite, GraphiteReporter}
import com.codahale.metrics.{MetricFilter, SharedMetricRegistries}
import play.api._

object Config extends GlobalSettings {

  override def onStart(app: Application) {
      val metricsEnabled = app.configuration.getString("metrics.enabled").getOrElse("false").toBoolean
      if (metricsEnabled) {
        setupGraphite(app)
      }
      Logger.info("Application has started")

   }


 private def setupGraphite(app: Application)  = {

//  val graphite = new PickledGraphite(new InetSocketAddress("graphite.example.com", 2004));


  val metricsPrefix = app.configuration.getString("metrics.prefix").getOrElse("NO-PREFIX")
  val metricsHost = app.configuration.getString("metrics.server.address").getOrElse("localhost")
  val metricsPort = app.configuration.getInt("metrics.server.port").getOrElse(2003)

  val graphite = new Graphite(new InetSocketAddress(metricsHost, metricsPort));

  val registry = SharedMetricRegistries.getOrCreate("default")
  val reporter : GraphiteReporter =  GraphiteReporter.forRegistry(registry)
                                          .prefixedWith(s"$metricsPrefix.${java.net.InetAddress.getLocalHost.getHostName}")
                                          .convertRatesTo(SECONDS)
                                          .convertDurationsTo(MILLISECONDS)
                                          .filter(MetricFilter.ALL)
                                          .build(graphite);
  reporter.start(1, SECONDS);

 }
} 

//设置package包名称以及导入依赖的类
package mesosphere.marathon.api

import java.io.StringWriter
import java.util.concurrent.TimeUnit
import javax.servlet.http.HttpServletRequest
import javax.ws.rs._
import javax.ws.rs.core.{ Context, MediaType, Response }

import com.codahale.metrics.{ MetricFilter, MetricRegistry }
import com.codahale.metrics.annotation.Timed
import com.codahale.metrics.json.MetricsModule
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.inject.Inject
import mesosphere.marathon.MarathonConf
import mesosphere.marathon.io.IO
import mesosphere.marathon.plugin.auth.AuthorizedResource.SystemConfig
import mesosphere.marathon.plugin.auth.{ Authenticator, Authorizer, ViewResource }


@Path("")
@Consumes(Array(MediaType.APPLICATION_JSON))
@Produces(Array(MarathonMediaType.PREFERRED_APPLICATION_JSON))
class SystemResource @Inject() (metrics: MetricRegistry, val config: MarathonConf)(implicit
  val authenticator: Authenticator,
    val authorizer: Authorizer) extends RestResource with AuthResource {

  private[this] lazy val mapper = new ObjectMapper().registerModule(
    new MetricsModule(TimeUnit.SECONDS, TimeUnit.SECONDS, false, MetricFilter.ALL)
  )

  @GET
  @Path("ping")
  @Timed
  def ping(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
    withAuthorization(ViewResource, SystemConfig){
      ok("pong")
    }
  }

  @GET
  @Path("metrics")
  @Timed
  def metrics(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
    withAuthorization(ViewResource, SystemConfig){
      IO.using(new StringWriter()) { writer =>
        mapper.writer().writeValue(writer, metrics)
        ok(writer.toString)
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.jgrier.flinkstuff.metrics

import java.util.concurrent.TimeUnit

import com.codahale.metrics.{MetricFilter, ScheduledReporter}
import metrics_influxdb.{HttpInfluxdbProtocol, InfluxdbReporter}
import metrics_influxdb.api.measurements.CategoriesMetricMeasurementTransformer
import org.apache.flink.dropwizard.ScheduledDropwizardReporter
import org.apache.flink.metrics.MetricConfig


class InfluxDbReporter extends ScheduledDropwizardReporter {
  override def getReporter(metricConfig: MetricConfig): ScheduledReporter = {

    val server = metricConfig.getString("server", "localhost")
    val port = metricConfig.getInteger("port", 8086)
    val user = metricConfig.getString("user", "admin")
    val password = metricConfig.getString("password", "admin")
    val db = metricConfig.getString("db", "flink")

    InfluxdbReporter.forRegistry(registry)
      .protocol(new HttpInfluxdbProtocol(server, port, user, password, db))
      .convertRatesTo(TimeUnit.SECONDS)
      .convertDurationsTo(TimeUnit.MILLISECONDS)
      .filter(MetricFilter.ALL)
      .skipIdleMetrics(false)
      .transformer(new CategoriesMetricMeasurementTransformer("host", "process_type", "tm_id", "job_name", "task_name", "subtask_index"))
      .build()
  }
} 

//设置package包名称以及导入依赖的类
package utility.metric

import java.net.InetSocketAddress
import java.util.concurrent.TimeUnit._

import com.codahale.metrics.graphite.{Graphite, GraphiteReporter}
import com.codahale.metrics.{MetricFilter, SharedMetricRegistries}
import config.AppConfig


object Graphite {

  lazy val config = AppConfig

  val graphite = new Graphite(new InetSocketAddress(
    config.metricHost, config.metricPort))

  lazy val prefix: String = config.metricPrefix

  val reporter = GraphiteReporter.forRegistry(
    SharedMetricRegistries.getOrCreate(prefix))
    .prefixedWith(s"$prefix.${java.net.InetAddress.getLocalHost.getHostName}")
    .convertRatesTo(SECONDS)
    .convertDurationsTo(MILLISECONDS)
    .filter(MetricFilter.ALL)
    .build(graphite)

  def startGraphite(): Unit = {
    if (config.metricEnabled) {
      reporter.start(config.metricRefreshInterval, SECONDS)
    }
  }
}