diff --git a/README.md b/README.md index 5bb3887..598790d 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,108 @@ Right now there are two options available: Return type is `Resource[F, _]` rather than `F[_]` because most often underlying metrics implementation upon a call registers your metrics with shared registry. Hence `release` hook of `Resource` being used in order to de-register particular metrics from shared registry. +## Kafka module +`smetrics-prometheus-kafka` module allows defining Prometheus collectors to obtain Kafka client's internal metrics +(ones collected by a producer/consumer themselves). +### `KafkaMetricsCollector` +See the example: +```scala +val collectorRegistry: CollectorRegistry = ??? +val consumer: Consumer[IO, String, String] = ??? +val collector = new KafkaMetricsCollector[IO](consumer.clientMetrics) +collectorRegistry.register(collector) +``` +In the example above `KafkaMetricsCollector` will call `consumer.clientMetrics` each time the registry attempts to +collect metric samples + +#### Multiple clients in the same VM +Creating multiple instances of `KafkaMetricsCollector` for different producers or consumers and attempting +to register them in `CollectorRegistry` will cause an error as they will contain metrics with the same name +which is prohibited by `CollectorRegistry`. There are multiple ways to mitigate the issue: +1. Using prefixes +`KafkaMetricsCollector` allows passing an optional prefix which can be prepended to all metrics' names. +This will result in multiple sets of metrics, e.g. `{prefix_1}_producer_metrics_request_size_avg` and +`{prefix_2}_producer_metrics_request_size_avg` +2. Combining the output of `clientMetrics` methods +See the example below: +```scala +val collectorRegistry: CollectorRegistry = ??? +val consumer1: Consumer[IO, String, String] = ??? +val consumer2: Consumer[IO, String, String] = ??? +val getAllMetrics: IO[Seq[ClientMetric[IO]]] = for { + metrics1 <- consumer1.clientMetrics + metrics2 <- consumer2.clientMetrics +} yield metrics1 ++ metrics2 +val collector = new KafkaMetricsCollector[IO](getAllMetrics) +collectorRegistry.register(collector) +``` +Note that this approach will result in duplicate metrics in case clients have the same configuration, e.g. when two +consumers have the same `client.id`: +``` +consumer_metrics_connection_creation_rate{client_id="client1",} 0.0 +consumer_metrics_connection_creation_rate{client_id="client1",} 0.0 +``` +3. Using `KafkaMetricsRegistry` described below. + +### `KafkaMetricsRegistry` +`KafkaMetricsRegistry` is an abstraction which aims to simplify gathering metrics from multiple clients in the same VM. +It allows 'registering' functions obtaining metrics from different clients, aggregating them into a single list +of metrics when collected. This allows defining clients in different code units with the only requirement of registering +them in `KafkaMetricsRegistry`. The registered functions will be saved in a `Ref` and invoked every time metrics +are collected. +Please note that `KafkaMetricsRegistry` doesn't extend Prometheus' `Collector`, thus it's still +necessary to create a single instance of `KafkaMetricsCollector` and register it with `CollectorRegistry`. +There are two ways of using `KafkaMetricsRegistry`: +1. Manual registration of each client +```scala +val collectorRegistry: CollectorRegistry = ??? +val consumerOf: ConsumerOf[F] = ??? +for { + kafkaRegistry <- KafkaMetricsRegistry.ref[IO].toResource + // Manually register each client after creating + consumer1 <- consumerOf.apply[K, V](config) + _ <- kafkaRegistry.register(consumer1.clientMetrics) + consumer2 <- consumerOf.apply[K, V](config) + _ <- kafkaRegistry.register(consumer2.clientMetrics) + // Create and register a single collector + kafkaCollector = new KafkaMetricsCollector[F](kafkaRegistry.collectAll) + _ <- F.delay(prometheusRegistry.register(kafkaCollector)).toResource +} yield () +``` +2. Wrapping `ConsumerOf` or `ProducerOf` with a syntax extension +```scala +import com.evolutiongaming.smetrics.kafka.syntax._ + +val collectorRegistry: CollectorRegistry = ??? +val consumerConfig1 = + ConsumerConfig.Default.copy( + groupId = Some("group1"), + common = CommonConfig.Default.copy(clientId = Some("client1")) + ) + +val consumerConfig2 = + ConsumerConfig.Default.copy( + groupId = Some("group2"), + common = CommonConfig.Default.copy(clientId = Some("client2")) + ) + +for { + kafkaRegistry <- KafkaMetricsRegistry.ref[F].toResource + // All consumers created with this factory will automatically register their metrics functions to `kafkaRegistry` + consumerOf = ConsumerOf.apply1[F]().withNativeMetrics(kafkaRegistry) + consumer1 <- consumerOf.apply[String, String](consumerConfig1) + consumer2 <- consumerOf.apply[String, String](consumerConfig2) + // Create and register a single collector + kafkaCollector = new KafkaMetricsCollector[F](kafkaRegistry.collectAll) + _ <- F.delay(prometheusRegistry.register(kafkaCollector)).toResource +} yield () +``` +#### Metrics duplication +`KafkaMetricsRegistry` deduplicates metrics by default. It can be turned off by using a different factory method +accepting `allowDuplicates` parameter. +When using it in the default mode it's important to use different `client.id` values for different clients inside a +single VM, otherwise only one of them will be picked (order is not guaranteed). + ## Setup ```scala diff --git a/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/KafkaMetricsRegistry.scala b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/KafkaMetricsRegistry.scala new file mode 100644 index 0000000..193effa --- /dev/null +++ b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/KafkaMetricsRegistry.scala @@ -0,0 +1,98 @@ +package com.evolutiongaming.smetrics.kafka + +import cats.effect.kernel.Sync +import cats.effect.{Ref, Resource} +import cats.syntax.all._ +import com.evolutiongaming.skafka.ClientMetric + +import java.util.UUID + +/** Allows reporting metrics of multiple Kafka clients inside a single VM. + * Note that it's still necessary to create an accompanying instance of `KafkaMetricsCollector` and register it + * with Prometheus' collector. + * + * Example: + * {{{ + * val prometheusRegistry: CollectorRegistry = ... + * val kafkaRegistry: KafkaMetricsRegistry[F] = ... + * val kafkaCollector = new KafkaMetricsCollector[F](kafkaRegistry.collectAll) + * val consumerOf: ConsumerOf[F] = ... + * + * for { + * _ <- F.delay(prometheusRegistry.register(kafkaCollector)).toResource + * consumer <- consumerOf.apply(config) + * _ <- kafkaRegistry.register(consumer.clientMetrics) + * } yield ... + * }}} + * + * To avoid manually registering each client there are syntax extension, wrapping `ProducerOf` and `ConsumerOf`, + * see `com.evolutiongaming.smetrics.kafka.syntax`. + * + * Example: + * {{{ + * import com.evolutiongaming.smetrics.kafka.syntax._ + * + * val prometheusRegistry: CollectorRegistry = ... + * val kafkaRegistry: KafkaMetricsRegistry[F] = ... + * val consumerOf = ConsumerOf.apply1[F]().withNativeMetrics(kafkaRegistry) + * val kafkaCollector = new KafkaMetricsCollector[F](kafkaRegistry.collectAll) + * + * for { + * _ <- F.delay(prometheusRegistry.register(kafkaCollector)).toResource + * // usage of `consumerOf` as usual ... + * } yield ... + * }}} + * + * */ +trait KafkaMetricsRegistry[F[_]] { + /** + * Register a function to obtain a list of client metrics. + * Normally, you would pass [[com.evolutiongaming.skafka.consumer.Consumer.clientMetrics]] or + * [[com.evolutiongaming.skafka.producer.Producer.clientMetrics]] + * + * @return synthetic ID of registered function + */ + def register(metrics: F[Seq[ClientMetric[F]]]): Resource[F, UUID] + + /** Collect metrics from all registered functions */ + def collectAll: F[Seq[ClientMetric[F]]] +} + +object KafkaMetricsRegistry { + private final class FromRef[F[_] : Sync](ref: Ref[F, Map[UUID, F[Seq[ClientMetric[F]]]]], allowDuplicates: Boolean) + extends KafkaMetricsRegistry[F] { + override def register(metrics: F[Seq[ClientMetric[F]]]): Resource[F, UUID] = { + val acquire: F[UUID] = for { + id <- Sync[F].delay(UUID.randomUUID()) + _ <- ref.update(m => m + (id -> metrics)) + } yield id + + def release(id: UUID): F[Unit] = + ref.update(m => m - id) + + Resource.make(acquire)(id => release(id)) + } + + override def collectAll: F[Seq[ClientMetric[F]]] = + ref.get.flatMap { map: Map[UUID, F[Seq[ClientMetric[F]]]] => + map.values.toList.sequence.map { metrics => + val results: List[ClientMetric[F]] = metrics.flatten + + if (allowDuplicates) { + results + } else { + results + .groupBy(metric => (metric.name, metric.group, metric.tags)) + .map { case (_, values) => values.head } + .toSeq + } + } + } + } + + def ref[F[_] : Sync](allowDuplicates: Boolean): F[KafkaMetricsRegistry[F]] = { + Ref.of[F, Map[UUID, F[Seq[ClientMetric[F]]]]](Map.empty).map(ref => new FromRef[F](ref, allowDuplicates)) + } + + def ref[F[_] : Sync]: F[KafkaMetricsRegistry[F]] = ref[F](allowDuplicates = false) +} diff --git a/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredConsumerOf.scala b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredConsumerOf.scala new file mode 100644 index 0000000..45621f4 --- /dev/null +++ b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredConsumerOf.scala @@ -0,0 +1,24 @@ +package com.evolutiongaming.smetrics.kafka + +import cats.effect.Resource +import com.evolutiongaming.skafka.FromBytes +import com.evolutiongaming.skafka.consumer.{Consumer, ConsumerConfig, ConsumerOf} + +class MeteredConsumerOf[F[_]](consumerOf: ConsumerOf[F], + kafkaMetricsRegistry: KafkaMetricsRegistry[F]) extends ConsumerOf[F] { + + override def apply[K, V](config: ConsumerConfig)(implicit + fromBytesK: FromBytes[F, K], + fromBytesV: FromBytes[F, V]): Resource[F, Consumer[F, K, V]] = { + for { + consumer <- consumerOf.apply[K, V](config) + _ <- kafkaMetricsRegistry.register(consumer.clientMetrics) + } yield consumer + } +} + +object MeteredConsumerOf { + def wrap[F[_]](consumerOf: ConsumerOf[F], + kafkaMetricsRegistry: KafkaMetricsRegistry[F]): MeteredConsumerOf[F] = + new MeteredConsumerOf[F](consumerOf, kafkaMetricsRegistry) +} diff --git a/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredProducerOf.scala b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredProducerOf.scala new file mode 100644 index 0000000..141f77e --- /dev/null +++ b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/MeteredProducerOf.scala @@ -0,0 +1,20 @@ +package com.evolutiongaming.smetrics.kafka + +import cats.effect.Resource +import com.evolutiongaming.skafka.producer.{Producer, ProducerConfig, ProducerOf} + +class MeteredProducerOf[F[_]](producerOf: ProducerOf[F], + kafkaMetricsRegistry: KafkaMetricsRegistry[F]) extends ProducerOf[F] { + override def apply(config: ProducerConfig): Resource[F, Producer[F]] = { + for { + producer <- producerOf.apply(config) + _ <- kafkaMetricsRegistry.register(producer.clientMetrics) + } yield producer + } +} + +object MeteredProducerOf { + def wrap[F[_]](producerOf: ProducerOf[F], + kafkaMetricsRegistry: KafkaMetricsRegistry[F]): MeteredProducerOf[F] = + new MeteredProducerOf[F](producerOf, kafkaMetricsRegistry) +} diff --git a/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/syntax.scala b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/syntax.scala new file mode 100644 index 0000000..2b000e8 --- /dev/null +++ b/modules/kafka/src/main/scala-2/com/evolutiongaming/smetrics/kafka/syntax.scala @@ -0,0 +1,14 @@ +package com.evolutiongaming.smetrics.kafka + +import com.evolutiongaming.skafka.consumer.ConsumerOf +import com.evolutiongaming.skafka.producer.ProducerOf + +object syntax { + implicit final class MeteredProducerOfOps[F[_]](val producerOf: ProducerOf[F]) extends AnyVal { + def withNativeMetrics(registry: KafkaMetricsRegistry[F]): ProducerOf[F] = MeteredProducerOf.wrap(producerOf, registry) + } + + implicit final class MeteredConsumerOfOps[F[_]](val consumerOf: ConsumerOf[F]) extends AnyVal { + def withNativeMetrics(registry: KafkaMetricsRegistry[F]): ConsumerOf[F] = MeteredConsumerOf.wrap(consumerOf, registry) + } +}