Add KafkaMetricsRegistry

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

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)
+}

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)
+}

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)
+}

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)
+  }
+}