forked from TouK/nussknacker
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Default InfluxDb reporter for standalone-app metrics
- Loading branch information
Showing
4 changed files
with
113 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
75 changes: 75 additions & 0 deletions
75
...ussknacker/engine/standalone/utils/metrics/dropwizard/influxdb/InfluxDbHttpReporter.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package pl.touk.nussknacker.engine.standalone.utils.metrics.dropwizard.influxdb | ||
|
|
||
| import com.typesafe.scalalogging.LazyLogging | ||
| import io.dropwizard.metrics5.influxdb._ | ||
| import io.dropwizard.metrics5.{MetricName, MetricRegistry} | ||
| import sttp.client._ | ||
|
|
||
| import java.lang | ||
| import java.nio.charset.StandardCharsets | ||
| import scala.collection.mutable.ArrayBuffer | ||
| import scala.concurrent.duration.Duration | ||
| import scala.util.{Failure, Success, Try} | ||
|
|
||
| object InfluxDbHttpReporter { | ||
|
|
||
| def build(metricRegistry: MetricRegistry, conf: InfluxSenderConfig): InfluxDbReporter = | ||
| InfluxDbReporter.forRegistry(metricRegistry) | ||
| .prefixedWith( | ||
| conf.prefix.map(MetricName.build(_)).getOrElse(MetricName.empty()) | ||
| .tagged("host", conf.host) | ||
| .tagged("env", conf.environment) | ||
| .tagged("type", conf.`type`) | ||
| ).build(new InfluxDbHttpSender(conf)) | ||
| } | ||
|
|
||
| class InfluxDbHttpSender(conf: InfluxSenderConfig) extends InfluxDbSender with LazyLogging { | ||
| private implicit val backend: SttpBackend[Try, Nothing, NothingT] = TryHttpURLConnectionBackend() | ||
|
|
||
| private val buffer = ArrayBuffer[String]() | ||
|
|
||
| logger.info(s"InfluxSender created with url: ${conf.req.uri}") | ||
|
|
||
| override def connect(): Unit = {} | ||
|
|
||
| override def send(measurement: lang.StringBuilder): Unit = buffer.append(measurement.toString) | ||
|
|
||
| override def flush(): Unit = { | ||
| val data = buffer.mkString | ||
| logger.debug(s"Sending ${buffer.size} metrics for conf $conf") | ||
| buffer.clear() | ||
| val answer = conf.req.body(data).send() | ||
|
|
||
| answer match { | ||
| case Success(res) if res.code.isSuccess => // nothing | ||
| case Success(res) => logger.warn(s"Failed to send data to influx: ${res.code.code}, ${res.body}") | ||
| case Failure(ex) => logger.warn(s"Failed to send data to influx: ${ex.getMessage}", ex) | ||
| } | ||
| } | ||
|
|
||
| override def disconnect(): Unit = {} | ||
|
|
||
| override def isConnected: Boolean = true | ||
|
|
||
| override def close(): Unit = {} | ||
| } | ||
|
|
||
|
|
||
| case class InfluxSenderConfig(url: String, | ||
| database: String, | ||
| `type`: String, | ||
| host: String, | ||
| environment: String, | ||
| prefix: Option[String], | ||
| retentionPolicy: Option[String], | ||
| username: Option[String], | ||
| password: Option[String], | ||
| reporterPolling: Duration) { | ||
|
|
||
|
|
||
| private val params = ("db" -> database) :: username.map("u" -> _).toList ::: password.map("p" -> _).toList ::: retentionPolicy.map("rp" -> _).toList | ||
|
|
||
| def req: RequestT[Identity, Either[String, String], Nothing] = basicRequest.post(uri"$url?$params") | ||
| .contentType("application/json", StandardCharsets.UTF_8.name()) | ||
|
|
||
| } |
28 changes: 28 additions & 0 deletions
28
...cker/engine/standalone/utils/metrics/dropwizard/influxdb/StandaloneInfluxDbReporter.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package pl.touk.nussknacker.engine.standalone.utils.metrics.dropwizard.influxdb | ||
|
|
||
| import com.typesafe.config.Config | ||
| import com.typesafe.scalalogging.LazyLogging | ||
| import io.dropwizard.metrics5.MetricRegistry | ||
| import io.dropwizard.metrics5.influxdb.InfluxDbReporter | ||
|
|
||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| object StandaloneInfluxDbReporter extends LazyLogging { | ||
|
|
||
| import net.ceedubs.ficus.Ficus._ | ||
| import net.ceedubs.ficus.readers.ArbitraryTypeReader._ | ||
|
|
||
| def createAndRunReporterIfConfigured(metricRegistry: MetricRegistry, config: Config): Option[InfluxDbReporter] = { | ||
| config.getAs[InfluxSenderConfig]("metrics.influx").map { influxSenderConfig => | ||
| logger.info("Found Influxdb metrics reporter config, starting reporter") | ||
| val reporter = InfluxDbHttpReporter.build(metricRegistry, influxSenderConfig) | ||
| reporter.start(influxSenderConfig.reporterPolling.toSeconds, TimeUnit.SECONDS) | ||
| reporter | ||
| } orElse { | ||
| logger.info("Influxdb metrics reporter config not found") | ||
| None | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|