11031647492

Committed 25 Sep 2024 11:00AM UTC coverage: 72.675% (-2.9%) from 75.566%

Build # 11031647492

Build Type

push

github

Committed by

web-flow

Commit Message

Add histogram-based metrics (#445)

* Add histogram-based metrics (#440)

* Extract ProducerMetrics implementations into named classes

* Add Histograms implementation for ProducerMetrics

* Extract ConsumerMetrics implementations into named classes

* Add Histogram implementation of ConsumerMetrics

* Name Throwables in MonadCancel

* Docs for latencyBuckets

* Apply suggestions from code review

Co-authored-by: Mareks Rampāns <8796159+mr-git@users.noreply.github.com>

* Update skafka/src/main/scala/com/evolutiongaming/skafka/consumer/ConsumerMetrics.scala

Co-authored-by: Mareks Rampāns <8796159+mr-git@users.noreply.github.com>

* changed name

* fix scala 3

* Apply suggestions from code review

Co-authored-by: Mareks Rampāns <8796159+mr-git@users.noreply.github.com>

* bump minor version

---------

Co-authored-by: Mareks Rampāns <8796159+mr-git@users.noreply.github.com>
Co-authored-by: Denys Fakhritdinov <denys.fakhritdinov@gmail.com>

(cherry picked from commit 0b370129b)

* bump stuff

Run Details

51 of 161 new or added lines in 2 files covered. (31.68%)

14 existing lines in 6 files now uncovered.

1383 of 1903 relevant lines covered (72.67%)

1.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/modules/metrics/src/main/scala/com/evolutiongaming/skafka/metrics/KafkaMetricsCollector.scala

package com.evolutiongaming.skafka.metrics
import cats.Monad
import cats.syntax.all._
import com.evolutiongaming.catshelper.CatsHelper._
import com.evolutiongaming.catshelper.ToTry
import com.evolutiongaming.skafka.ClientMetric
import io.prometheus.client.Collector
import io.prometheus.client.Collector.MetricFamilySamples
import io.prometheus.client.Collector.MetricFamilySamples.Sample

import java.util
import scala.jdk.CollectionConverters._
import scala.util.matching.Regex

/**
 * Prometheus collector for Kafka client metrics.
 *
 * Metrics from kafka-clients are pull-based (meaning we have to manually call methods that return metrics), this
 * doesn't allow us to use the default push-based approach of Prometheus with passing an instance of `CollectorRegistry`
 * to the code being observed.
 *
 * This class is a workaround for this problem. It's a Prometheus collector that evaluates a given
 * `F[Seq[ClientMetric[F]]]` on each `collect()` call and returns the result as a list of `MetricFamilySamples`.
 *
 * Please note it ignores metrics and labels containing some symbols (like '-' and '.'). This is because Prometheus
 * doesn't allow such symbols in metric names and label names. Also, it ignores metrics without a name or a description.
 * See [[https://prometheus.io/docs/practices/naming/]] for more details.
 *
 * Example:
 * {{{
 * val collectorRegistry: CollectorRegistry = ??? // your Prometheus Java collector registry
 * val consumer: Consumer[F, K, V]          = ??? // your consumer
 * val collector: KafkaMetricsCollector[F]  = new KafkaMetricsCollector[F](consumer.clientMetrics)
 * collectorRegistry.register(collector)
 * }}}
 **/
class KafkaMetricsCollector[F[_]: Monad: ToTry](kafkaClientMetrics: F[Seq[ClientMetric[F]]], prefix: Option[String] = None) extends Collector {

  protected def getCollectorType(metric: ClientMetric[F]): Collector.Type = {
    // https://prometheus.io/docs/practices/naming/#metric-names
    if (metric.name.endsWith("total")) Collector.Type.COUNTER
    else if (metric.name.contains("count")) Collector.Type.SUMMARY
    else Collector.Type.GAUGE
  }

  private val MetricNameRegex: Regex = "[a-zA-Z_:][a-zA-Z0-9_:]*".r
  private val LabelNameRegex: Regex = "[a-zA-Z_][a-zA-Z0-9_]*".r

  protected def getPrometheusName(metric: ClientMetric[F]): Option[String] = {
    (metric.name, metric.description) match {
      case (name, description) if name.nonEmpty && description.nonEmpty =>
        val prometheusName =
          (prefix.toList :+ metric.group :+ metric.name).mkString("_").replaceAll("-", "_")

        if (MetricNameRegex.findFirstIn(prometheusName).contains(prometheusName)) prometheusName.some else None

      case _ => None
    }
  }

  override def collect(): util.List[MetricFamilySamples] = {
    for {
      metrics <- kafkaClientMetrics
      metricsGroups = metrics.groupBy(m => (m.name, m.group)).values.toList
      result <- metricsGroups.traverse { metricsGroup =>
        val metric = metricsGroup.head
        val prometheusName = getPrometheusName(metric)
        val collectorType = getCollectorType(metric)

        prometheusName match {
          case Some(name) =>
            metricsGroup.toVector.traverse { metric =>
              val tags = metric.tags.flatMap {
                case (key, value) =>
                  val prometheusKey = key.replaceAll("-", "_")
                  if (LabelNameRegex.findFirstIn(prometheusKey).contains(prometheusKey)) (prometheusKey -> value).some else None
              }
              val tagsKeys = tags.keys.toList.asJava
              val tagsValues = tags.values.toList.asJava
              metric.value.map {
                case v: Number =>
                  new Sample(name, tagsKeys, tagsValues, v.doubleValue()).some

                case _ => Option.empty[Sample]
              }
            }.map(_.flatten).map {
              case samples if samples.nonEmpty =>
                new MetricFamilySamples(name, collectorType, metric.description, samples.asJava).some

              case _ => Option.empty[MetricFamilySamples]
            }

          case None => Option.empty[MetricFamilySamples].pure
        }
      }.map(_.flatten.asJava)
    } yield result
  }.toTry.get
}

1	package com.evolutiongaming.skafka.metrics
2	import cats.Monad
3	import cats.syntax.all._
4	import com.evolutiongaming.catshelper.CatsHelper._
5	import com.evolutiongaming.catshelper.ToTry
6	import com.evolutiongaming.skafka.ClientMetric
7	import io.prometheus.client.Collector
8	import io.prometheus.client.Collector.MetricFamilySamples
9	import io.prometheus.client.Collector.MetricFamilySamples.Sample
10
11	import java.util
12	import scala.jdk.CollectionConverters._
13	import scala.util.matching.Regex
14
15	/**
16	* Prometheus collector for Kafka client metrics.
17	*
18	* Metrics from kafka-clients are pull-based (meaning we have to manually call methods that return metrics), this
19	* doesn't allow us to use the default push-based approach of Prometheus with passing an instance of `CollectorRegistry`
20	* to the code being observed.
21	*
22	* This class is a workaround for this problem. It's a Prometheus collector that evaluates a given
23	* `F[Seq[ClientMetric[F]]]` on each `collect()` call and returns the result as a list of `MetricFamilySamples`.
24	*
25	* Please note it ignores metrics and labels containing some symbols (like '-' and '.'). This is because Prometheus
26	* doesn't allow such symbols in metric names and label names. Also, it ignores metrics without a name or a description.
27	* See [[https://prometheus.io/docs/practices/naming/]] for more details.
28	*
29	* Example:
30	* {{{
31	* val collectorRegistry: CollectorRegistry = ??? // your Prometheus Java collector registry
32	* val consumer: Consumer[F, K, V] = ??? // your consumer
33	* val collector: KafkaMetricsCollector[F] = new KafkaMetricsCollector[F](consumer.clientMetrics)
34	* collectorRegistry.register(collector)
35	* }}}
36	**/
37	class KafkaMetricsCollector[F[_]: Monad: ToTry](kafkaClientMetrics: F[Seq[ClientMetric[F]]], prefix: Option[String] = None) extends Collector {
38
39	protected def getCollectorType(metric: ClientMetric[F]): Collector.Type = {
40	// https://prometheus.io/docs/practices/naming/#metric-names
41	if (metric.name.endsWith("total")) Collector.Type.COUNTER	×
42	else if (metric.name.contains("count")) Collector.Type.SUMMARY	×
43	else Collector.Type.GAUGE	×
44	}
45
46	private val MetricNameRegex: Regex = "[a-zA-Z_:][a-zA-Z0-9_:]*".r	×
47	private val LabelNameRegex: Regex = "[a-zA-Z_][a-zA-Z0-9_]*".r	×
48
49	protected def getPrometheusName(metric: ClientMetric[F]): Option[String] = {
50	(metric.name, metric.description) match {
51	case (name, description) if name.nonEmpty && description.nonEmpty =>	×
52	val prometheusName =
53	(prefix.toList :+ metric.group :+ metric.name).mkString("_").replaceAll("-", "_")	×
54
55	if (MetricNameRegex.findFirstIn(prometheusName).contains(prometheusName)) prometheusName.some else None	×
56
57	case _ => None	×
58	}
59	}
60
61	override def collect(): util.List[MetricFamilySamples] = {
62	for {
63	metrics <- kafkaClientMetrics	×
64	metricsGroups = metrics.groupBy(m => (m.name, m.group)).values.toList	×
65	result <- metricsGroups.traverse { metricsGroup =>	×
66	val metric = metricsGroup.head	×
67	val prometheusName = getPrometheusName(metric)	×
68	val collectorType = getCollectorType(metric)	×
69
70	prometheusName match {
71	case Some(name) =>
72	metricsGroup.toVector.traverse { metric =>	×
73	val tags = metric.tags.flatMap {	×
UNCOV 74	case (key, value) =>	×
75	val prometheusKey = key.replaceAll("-", "_")	×
76	if (LabelNameRegex.findFirstIn(prometheusKey).contains(prometheusKey)) (prometheusKey -> value).some else None	×
77	}
78	val tagsKeys = tags.keys.toList.asJava	×
79	val tagsValues = tags.values.toList.asJava	×
80	metric.value.map {	×
81	case v: Number =>
82	new Sample(name, tagsKeys, tagsValues, v.doubleValue()).some	×
83
84	case _ => Option.empty[Sample]	×
85	}
86	}.map(_.flatten).map {	×
87	case samples if samples.nonEmpty =>	×
88	new MetricFamilySamples(name, collectorType, metric.description, samples.asJava).some	×
89
90	case _ => Option.empty[MetricFamilySamples]	×
91	}
92
93	case None => Option.empty[MetricFamilySamples].pure	×
94	}
95	}.map(_.flatten.asJava)	×
96	} yield result
97	}.toTry.get	×
98	}

evolution-gaming / skafka / 11031647492

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous