8094751572

Committed 29 Feb 2024 10:44AM UTC coverage: 81.589% (-0.3%) from 81.861%

Build Type

push

Committed by

Commit Message

Revert "Restore binary backwards compatibility of EventualCassandraConfig."

This reverts commit 16c18228c.

Run Details

3 of 3 new or added lines in 2 files covered. (100.0%)

11 existing lines in 3 files now uncovered.

3297 of 4041 relevant lines covered (81.59%)

1.36 hits per line

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

86.81

/journal/src/main/scala/com/evolutiongaming/kafka/journal/HeadCache.scala

package com.evolutiongaming.kafka.journal

import cats._
import cats.effect._
import cats.effect.syntax.all._
import cats.effect.kernel.Async
import cats.syntax.all._
import com.evolutiongaming.catshelper._
import com.evolutiongaming.kafka.journal.PartitionCache.Result
import com.evolutiongaming.kafka.journal.conversions.ConsRecordToActionHeader
import com.evolutiongaming.kafka.journal.eventual.{EventualJournal, TopicPointers}
import com.evolution.scache.{Cache, ExpiringCache}
import com.evolutiongaming.skafka.consumer.ConsumerConfig
import com.evolutiongaming.skafka.{Offset, Partition, Topic}
import com.evolutiongaming.smetrics.MetricsHelper._
import com.evolutiongaming.smetrics._

import scala.concurrent.duration._

/** Metainfo of events written to Kafka, but not yet replicated to Cassandra.
  *
  * The implementation subscribes to all events in Kafka and periodically polls
  * Cassandra to remove information about the events, which already replicated.
  *
  * The returned entries do not contain the events themselves, but only an
  * offset of the first non-replicated event, the sequence number of last event,
  * range of events to be deleted etc.
  *
  * The consuming/polling of the records will stop after a configured timeout
  * (i.e. [[HeadCacheConfig#expiry]]) for the topics where no activity /
  * recoveries happen. It will restart when the new calls to a cache come for
  * these topics again.
  *
  * TODO headcache:
  * 1. Keep 1000 last seen entries, even if replicated.
  * 2. Fail headcache when background tasks failed
  *
  * @see [[HeadInfo]] for more details on the purpose of the stored data.
  */
trait HeadCache[F[_]] {

  /** Get the information about a state of a journal stored in the cache.
    *
    * @param key
    *   Journal key including a Kafka topic where journal is stored and
    *   a journal identifier.
    * @param partition
    *   Partition where journal is stored to. The usual way to get the partition
    *   is to write a "marker" record to Kafka topic and use the partition of
    *   the marker as a current one.
    * @param offset
    *   Current [[Offset]], i.e. maximum offset where Kafka records related to a
    *   journal are located. The usual way to get such an offset is to write a
    *   "marker" record to Kafka patition and use the offset of the marker as a
    *   current one.
    *
    * @return
    *   [[HeadInfo]] with the current metainformation about non-replicated
    *   events, or `None` if it was not present in [[HeadCache]] and could not
    *   be loaded either.
    */
  def get(key: Key, partition: Partition, offset: Offset): F[Option[HeadInfo]]
}


object HeadCache {

  /** Disable cache and always return `None` */
  def empty[F[_]: Applicative]: HeadCache[F] = const(none[HeadInfo].pure[F])


  /** Disable cache and always return a predefined value */
  def const[F[_]](value: F[Option[HeadInfo]]): HeadCache[F] = {
    class Const
    new Const with HeadCache[F] {
      def get(key: Key, partition: Partition, offset: Offset) = value
    }
  }


  /** Creates new cache using a Kafka configuration and Cassandra reader.
    *
    * The created instances will report metrics to `metrics` and also will do
    * the debug logging. There is no need to call [[HeadCache#withLogs]] on
    * them.
    *
    * @param consumerConfig
    *   Kafka consumer configuration used to find new non-replicated journal
    *   events. Some of the parameters will be ignored. See
    *   [[TopicCache.Consumer#of]] for more details.
    * @param eventualJournal
    *   Cassandra (or other long term storage) data source used to remove
    *   replicated events from the cache. Usually created by calling
    *   [[EventualCassandra#of]].
    * @param metrics
    *   Interface to report the metrics to. The intended way to configure it is
    *   overriding [[KafkaJournal#metrics]] in a custom implementation of
    *   [[KafkaJournal]].
    * @return
    *   Resource which will configure a [[HeadCache]] with the passed
    *   parameters. Instance of `Resource[HeadCache]` are, obviously, reusable
    *   and there is no need to call [[HeadCache#of]] each time if parameters
    *   did not change.
    */
  def of[F[_]: Async: Parallel: Runtime: LogOf: KafkaConsumerOf: MeasureDuration: FromTry: FromJsResult: JsonCodec.Decode](
    consumerConfig: ConsumerConfig,
    eventualJournal: EventualJournal[F],
    metrics: Option[HeadCacheMetrics[F]]
  ): Resource[F, HeadCache[F]] = {
    for {
      log    <- LogOf[F].apply(HeadCache.getClass).toResource
      result <- HeadCache.of(
        Eventual(eventualJournal),
        log,
        TopicCache.Consumer.of[F](consumerConfig),
        metrics)
      result <- result.withFence
    } yield {
      result.withLog(log)
    }
  }

  /** Creates new cache using Kafka and Cassandra data sources.
    *
    * The method also allows to change the default configuration in form of
    * [[HeadCacheConfig]], i.e. to make the polling faster for testing purposes.
    *
    * @param eventual
    *   Cassandra data source.
    * @param log
    *   Logger to use for [[TopicCache#withLog]]. Note, that only [[TopicCache]]
    *   debug logging will be affected by this. One needs to call
    *   [[HeadCache#withLog]] if debug logging for [[HeadCache]] is required.
    * @param consumer
    *   Kakfa data source factory. The reason why it is factory (i.e.
    *   `Resource`) is that [[HeadCache]] will try to recreate consumer in case
    *   of the failure.
    * @param metrics
    *   Interface to report the metrics to. The intended way to configure it is
    *   overriding [[KafkaJournal#metrics]] in a custom implementation of
    *   [[KafkaJournal]].
    * @param config
    *   Cache configuration. It is recommended to keep it default, and only
    *   change it for unit testing purposes.
    * @return
    *   Resource which will configure a [[HeadCache]] with the passed
    *   parameters. Instance of `Resource[HeadCache]` are, obviously, reusable
    *   and there is no need to call [[HeadCache#of]] each time if parameters
    *   did not change.
    */
  def of[F[_]: Async: Parallel: Runtime: FromJsResult: MeasureDuration: JsonCodec.Decode](
    eventual: Eventual[F],
    log: Log[F],
    consumer: Resource[F, TopicCache.Consumer[F]],
    metrics: Option[HeadCacheMetrics[F]],
    config: HeadCacheConfig = HeadCacheConfig.default
  ): Resource[F, HeadCache[F]] = {

    val consRecordToActionHeader = ConsRecordToActionHeader[F]
    for {
      cache <- Cache.expiring(
        ExpiringCache.Config[F, Topic, TopicCache[F]](expireAfterRead = config.expiry),
        partitions = 1.some)
      cache <- metrics.fold(cache.pure[Resource[F, *]]) { metrics => cache.withMetrics(metrics.cache) }
    } yield {
      class Main
      new Main with HeadCache[F] {

        def get(key: Key, partition: Partition, offset: Offset) = {
          val topic = key.topic
          val log1 = log.prefixed(topic)
          cache
            .getOrUpdateResource(topic) {
              TopicCache
                .of(
                  eventual,
                  topic,
                  log1,
                  consumer,
                  config,
                  consRecordToActionHeader,
                  metrics.map { _.headCache })
                .map { cache =>
                  metrics
                    .fold(cache) { metrics => cache.withMetrics(topic, metrics.headCache) }
                    .withLog(log1)
                }
            }
            .flatMap { cache =>
              cache
                .get(key.id, partition, offset)
                .flatMap { _.toNow }
                .map {
                  case a: Result.Now.Value   => a.value.some
                  case Result.Now.Ahead      => HeadInfo.empty.some
                  case Result.Now.Limited    => none
                  case _: Result.Now.Timeout => none
                }
            }
        }
      }
    }
  }


  /** Lighweight wrapper over [[EventualJournal]].
    *
    * Allows easier stubbing in unit tests.
    */
  trait Eventual[F[_]] {

    /** Gets the last replicated offset for a partition topic.
      *
      * @see [[EventualJournal#offset]] for more details.
      */
    def pointer(topic: Topic, partition: Partition): F[Option[Offset]]

  }

  object Eventual {

    def apply[F[_]](implicit F: Eventual[F]): Eventual[F] = F

    /** Wrap [[EventualJournal]] into [[Eventual]] */
    def apply[F[_]](eventualJournal: EventualJournal[F]): Eventual[F] = {
      class Main
      new Main with HeadCache.Eventual[F] {
        def pointer(topic: Topic, partition: Partition): F[Option[Offset]] = eventualJournal.offset(topic, partition)
      }
    }

    /** Always return `None` as an offset, i.e. pretend nothing ever replicates.
      *
      * Only useful for testing purposes.
      */
    def empty[F[_]: Applicative]: Eventual[F] = const(TopicPointers.empty.pure[F])

    /** Ignore topic and specify offset to return by partition.
      *
      * Only useful for testing purposes.
      */
    def const[F[_]: Applicative](value: F[TopicPointers]): Eventual[F] = {
      class Const
      new Const with Eventual[F] {
        def pointer(topic: Topic, partition: Partition): F[Option[Offset]] = value.map(_.values.get(partition))
      }
    }
  }

  private abstract sealed class WithFence

  private abstract sealed class WithLog

  implicit class HeadCacheOps[F[_]](val self: HeadCache[F]) extends AnyVal {

    def mapK[G[_]](f: F ~> G): HeadCache[G] = new HeadCache[G] {

      def get(key: Key, partition: Partition, offset: Offset) = {
        f(self.get(key, partition, offset))
      }
    }

    /** Log debug messages on every call to the class methods.
      *
      * The messages will go to DEBUG level, so it is also necessary to enable
      * it in logger configuration.
      */
    def withLog(log: Log[F])(implicit F: FlatMap[F], measureDuration: MeasureDuration[F]): HeadCache[F] = {
      new WithLog with HeadCache[F] {

        def get(key: Key, partition: Partition, offset: Offset) = {
          for {
            d <- MeasureDuration[F].start
            a <- self.get(key, partition, offset)
            d <- d
            _ <- log.debug(s"get in ${ d.toMillis }ms, key: $key, offset: $partition:$offset, result: $a")
          } yield a
        }
      }
    }

    /** Prevents cache methods to be used after `Resource` was released.
      *
      * [[ReleasedError]] will be raised if [[HeadCache#get]] is called after
      * resource is released.
      *
      * It may prevent certain kind of bugs, when several caches are,
      * accidentially, alive and working.
      */
    def withFence(implicit F: Sync[F]): Resource[F, HeadCache[F]] = {
      Resource
        .make { Ref[F].of(().pure[F]) } { _.set(ReleasedError.raiseError[F, Unit]) }
        .map { ref =>
          new WithFence with HeadCache[F] {
            def get(key: Key, partition: Partition, offset: Offset) = {
              ref
                .get
                .flatten
                .productR { self.get(key, partition, offset) }
            }
          }
        }
    }
  }


  /** Provides methods to update the metrics for [[HeadCache]] internals */
  trait Metrics[F[_]] {

    /** Report duration and result of cache hits, i.e. [[TopicCache#get]].
      *
      * @param topic
      *   Topic journal is being stored in.
      * @param latency
      *   Duration of [[TopicCache#get]] call.
      * @param result
      *   Result of the call, i.e. "ahead", "limited", "timeout" or "failure".
      * @param now
      *   If result was [[PartitionCache.Result.Now]], i.e. entry was already in
      *   cache.
      */
    def get(topic: Topic, latency: FiniteDuration, result: String, now: Boolean): F[Unit]

    /** Report health of all [[PartitionCache]] instances related to a topic.
      *
      * @param topic
      *   Topic which these [[PartitionCache]] instances are related to.
      * @param entries
      *   Number of distinct journals stored in a topic cache. If it is too
      *   close to [[HeadCacheConfig.Partition#maxSize]] multiplied by number of
      *   partitions, the cache might not work efficiently.
      * @param listeners
      *   Number of listeners waiting after [[PartitionCache#get]] call. Too
      *   many of them might mean that cache is not being loaded fast enough.
      */
    def meters(topic: Topic, entries: Int, listeners: Int): F[Unit]

    /** Report the latency and number of records coming from Kafka.
      *
      * I.e. how long it took for a next element in a stream returned by
      * [[HeadCacheConsumption#apply]] to get from a journal writer to this
      * cache.
      *
      * @param topic
      *   Topic being read by [[HeadCacheConsumption]].
      * @param age
      *   Time it took for an element to reach [[HeadCache]].
      * @param diff
      *   The number of elements added to cache by this batch, i.e. returned by
      *   [[PartitionCache#add]].
      */
    def consumer(topic: Topic, age: FiniteDuration, diff: Long): F[Unit]

    /** Report the number of records coming from Cassandra.
      *
      * @param topic
      *   Topic being read by [[Eventual]].
      * @param diff
      *   The number of elements remove from cache by this batch, i.e. returned
      *   by [[PartitionCache#remove]].
      */
    def storage(topic: Topic, diff: Long): F[Unit]
  }

  object Metrics {

    /** Does not do anything, ignores metric reports */
    def empty[F[_]: Applicative]: Metrics[F] = const(().pure[F])


    /** Calls a passed effect when metrics are reported.
      *
      * May only be useful for tests, as the reported parameters are ignored.
      */
    def const[F[_]](unit: F[Unit]): Metrics[F] = {
      class Const
      new Const with Metrics[F] {

        def get(topic: Topic, latency: FiniteDuration, result: String, hit: Boolean) = unit

        def meters(topic: Topic, entries: Int, listeners: Int) = unit

        def consumer(topic: Topic, age: FiniteDuration, diff: Long) = unit

        def storage(topic: Topic, diff: Long) = unit
      }
    }


    type Prefix = String

    object Prefix {
      val default: Prefix = "headcache"
    }


    /** Registers a default set of metrics to a passed collector registry.
      *
      * Note, that creating this metrics several times with the same collector
      * registry may cause errors unless previous [[Metrics]] instance was not
      * released yet.
      *
      * The following metrics will be registered by default, but it is possible
      * to override the default `headcache` prefix to something else.
      *
      * {{{
      * headcache_get_latency HeadCache get latency in seconds
      * headcache_get_result  HeadCache `get` call result counter
      * headcache_entries     HeadCache entries
      * headcache_listeners   HeadCache listeners
      * headcache_records_age HeadCache time difference between record timestamp and now in seconds
      * headcache_diff        HeadCache offset difference between state and source
      * }}}
      *
      * @param registry
      *   smetrics collector registry.
      * @param prefix
      *   Prefix to use for the registered metrics.
      */
    def of[F[_]: Monad](
      registry: CollectorRegistry[F],
      prefix: Prefix = Prefix.default
    ): Resource[F, Metrics[F]] = {

      val getLatencySummary = registry.summary(
        name = s"${ prefix }_get_latency",
        help = "HeadCache get latency in seconds",
        quantiles = Quantiles.Default,
        labels = LabelNames("topic", "result", "now"))

      val getResultCounter = registry.counter(
        name = s"${ prefix }_get_result",
        help = "HeadCache `get` call result counter",
        labels = LabelNames("topic", "result", "now"))

      val entriesGauge = registry.gauge(
        name = s"${ prefix }_entries",
        help = "HeadCache entries",
        labels = LabelNames("topic"))

      val listenersGauge = registry.gauge(
        name = s"${ prefix }_listeners",
        help = "HeadCache listeners",
        labels = LabelNames("topic"))

      val ageSummary = registry.summary(
        name = s"${ prefix }_records_age",
        help = "HeadCache time difference between record timestamp and now in seconds",
        quantiles = Quantiles.Default,
        labels = LabelNames("topic"))

      val diffSummary = registry.summary(
        name = s"${ prefix }_diff",
        help = "HeadCache offset difference between state and source",
        quantiles = Quantiles.Default,
        labels = LabelNames("topic", "source"))

      for {
        getLatencySummary <- getLatencySummary
        getResultCounter  <- getResultCounter
        entriesGauge      <- entriesGauge
        listenersGauge    <- listenersGauge
        ageSummary        <- ageSummary
        diffSummary       <- diffSummary
      } yield {

        class Main
        new Main with Metrics[F] {

          def get(topic: Topic, latency: FiniteDuration, result: String, now: Boolean) = {
            val nowLabel = if (now) "now" else "later"
            for {
              _ <- getLatencySummary
                .labels(topic, result, nowLabel)
                .observe(latency.toNanos.nanosToSeconds)
              a <- getResultCounter
                .labels(topic, result, nowLabel)
                .inc()
            } yield a
          }

          def meters(topic: Topic, entries: Int, listeners: Int) = {
            for {
              _ <- listenersGauge
                .labels(topic)
                .set(listeners.toDouble)
              a <- entriesGauge
                .labels(topic)
                .set(entries.toDouble)
            } yield a
          }

          def consumer(topic: Topic, age: FiniteDuration, diff: Long) = {
            for {
              _ <- ageSummary
                .labels(topic)
                .observe(age.toNanos.nanosToSeconds)
              a <- diffSummary
                .labels(topic, "kafka")
                .observe(diff.toDouble)
            } yield a
          }

          def storage(topic: Topic, diff: Long) = {
            diffSummary
              .labels(topic, "storage")
              .observe(diff.toDouble)
          }
        }
      }
    }
  }
}

1	package com.evolutiongaming.kafka.journal
2
3	import cats._
4	import cats.effect._
5	import cats.effect.syntax.all._
6	import cats.effect.kernel.Async
7	import cats.syntax.all._
8	import com.evolutiongaming.catshelper._
9	import com.evolutiongaming.kafka.journal.PartitionCache.Result
10	import com.evolutiongaming.kafka.journal.conversions.ConsRecordToActionHeader
11	import com.evolutiongaming.kafka.journal.eventual.{EventualJournal, TopicPointers}
12	import com.evolution.scache.{Cache, ExpiringCache}
13	import com.evolutiongaming.skafka.consumer.ConsumerConfig
14	import com.evolutiongaming.skafka.{Offset, Partition, Topic}
15	import com.evolutiongaming.smetrics.MetricsHelper._
16	import com.evolutiongaming.smetrics._
17
18	import scala.concurrent.duration._
19
20	/** Metainfo of events written to Kafka, but not yet replicated to Cassandra.
21	*
22	* The implementation subscribes to all events in Kafka and periodically polls
23	* Cassandra to remove information about the events, which already replicated.
24	*
25	* The returned entries do not contain the events themselves, but only an
26	* offset of the first non-replicated event, the sequence number of last event,
27	* range of events to be deleted etc.
28	*
29	* The consuming/polling of the records will stop after a configured timeout
30	* (i.e. [[HeadCacheConfig#expiry]]) for the topics where no activity /
31	* recoveries happen. It will restart when the new calls to a cache come for
32	* these topics again.
33	*
34	* TODO headcache:
35	* 1. Keep 1000 last seen entries, even if replicated.
36	* 2. Fail headcache when background tasks failed
37	*
38	* @see [[HeadInfo]] for more details on the purpose of the stored data.
39	*/
40	trait HeadCache[F[_]] {
41
42	/** Get the information about a state of a journal stored in the cache.
43	*
44	* @param key
45	* Journal key including a Kafka topic where journal is stored and
46	* a journal identifier.
47	* @param partition
48	* Partition where journal is stored to. The usual way to get the partition
49	* is to write a "marker" record to Kafka topic and use the partition of
50	* the marker as a current one.
51	* @param offset
52	* Current [[Offset]], i.e. maximum offset where Kafka records related to a
53	* journal are located. The usual way to get such an offset is to write a
54	* "marker" record to Kafka patition and use the offset of the marker as a
55	* current one.
56	*
57	* @return
58	* [[HeadInfo]] with the current metainformation about non-replicated
59	* events, or `None` if it was not present in [[HeadCache]] and could not
60	* be loaded either.
61	*/
62	def get(key: Key, partition: Partition, offset: Offset): F[Option[HeadInfo]]
63	}
64
65
66	object HeadCache {
67
68	/** Disable cache and always return `None` */
69	def empty[F[_]: Applicative]: HeadCache[F] = const(none[HeadInfo].pure[F])	1✔
70
71
72	/** Disable cache and always return a predefined value */
73	def const[F[_]](value: F[Option[HeadInfo]]): HeadCache[F] = {
74	class Const
75	new Const with HeadCache[F] {	2✔
76	def get(key: Key, partition: Partition, offset: Offset) = value
77	}
78	}
79
80
81	/** Creates new cache using a Kafka configuration and Cassandra reader.
82	*
83	* The created instances will report metrics to `metrics` and also will do
84	* the debug logging. There is no need to call [[HeadCache#withLogs]] on
85	* them.
86	*
87	* @param consumerConfig
88	* Kafka consumer configuration used to find new non-replicated journal
89	* events. Some of the parameters will be ignored. See
90	* [[TopicCache.Consumer#of]] for more details.
91	* @param eventualJournal
92	* Cassandra (or other long term storage) data source used to remove
93	* replicated events from the cache. Usually created by calling
94	* [[EventualCassandra#of]].
95	* @param metrics
96	* Interface to report the metrics to. The intended way to configure it is
97	* overriding [[KafkaJournal#metrics]] in a custom implementation of
98	* [[KafkaJournal]].
99	* @return
100	* Resource which will configure a [[HeadCache]] with the passed
101	* parameters. Instance of `Resource[HeadCache]` are, obviously, reusable
102	* and there is no need to call [[HeadCache#of]] each time if parameters
103	* did not change.
104	*/
105	def of[F[_]: Async: Parallel: Runtime: LogOf: KafkaConsumerOf: MeasureDuration: FromTry: FromJsResult: JsonCodec.Decode](
106	consumerConfig: ConsumerConfig,
107	eventualJournal: EventualJournal[F],
108	metrics: Option[HeadCacheMetrics[F]]
109	): Resource[F, HeadCache[F]] = {
110	for {
111	log <- LogOf[F].apply(HeadCache.getClass).toResource	1✔
112	result <- HeadCache.of(	1✔
113	Eventual(eventualJournal),	1✔
114	log,
115	TopicCache.Consumer.of[F](consumerConfig),	1✔
116	metrics)
117	result <- result.withFence	1✔
118	} yield {
119	result.withLog(log)	1✔
120	}
121	}
122
123	/** Creates new cache using Kafka and Cassandra data sources.
124	*
125	* The method also allows to change the default configuration in form of
126	* [[HeadCacheConfig]], i.e. to make the polling faster for testing purposes.
127	*
128	* @param eventual
129	* Cassandra data source.
130	* @param log
131	* Logger to use for [[TopicCache#withLog]]. Note, that only [[TopicCache]]
132	* debug logging will be affected by this. One needs to call
133	* [[HeadCache#withLog]] if debug logging for [[HeadCache]] is required.
134	* @param consumer
135	* Kakfa data source factory. The reason why it is factory (i.e.
136	* `Resource`) is that [[HeadCache]] will try to recreate consumer in case
137	* of the failure.
138	* @param metrics
139	* Interface to report the metrics to. The intended way to configure it is
140	* overriding [[KafkaJournal#metrics]] in a custom implementation of
141	* [[KafkaJournal]].
142	* @param config
143	* Cache configuration. It is recommended to keep it default, and only
144	* change it for unit testing purposes.
145	* @return
146	* Resource which will configure a [[HeadCache]] with the passed
147	* parameters. Instance of `Resource[HeadCache]` are, obviously, reusable
148	* and there is no need to call [[HeadCache#of]] each time if parameters
149	* did not change.
150	*/
151	def of[F[_]: Async: Parallel: Runtime: FromJsResult: MeasureDuration: JsonCodec.Decode](
152	eventual: Eventual[F],
153	log: Log[F],
154	consumer: Resource[F, TopicCache.Consumer[F]],
155	metrics: Option[HeadCacheMetrics[F]],
156	config: HeadCacheConfig = HeadCacheConfig.default
157	): Resource[F, HeadCache[F]] = {
158
159	val consRecordToActionHeader = ConsRecordToActionHeader[F]	1✔
160	for {
161	cache <- Cache.expiring(	1✔
162	ExpiringCache.Config[F, Topic, TopicCache[F]](expireAfterRead = config.expiry),	1✔
163	partitions = 1.some)	1✔
164	cache <- metrics.fold(cache.pure[Resource[F, *]]) { metrics => cache.withMetrics(metrics.cache) }	1✔
165	} yield {
166	class Main
167	new Main with HeadCache[F] {	1✔
168
169	def get(key: Key, partition: Partition, offset: Offset) = {
170	val topic = key.topic	2✔
171	val log1 = log.prefixed(topic)	2✔
172	cache
173	.getOrUpdateResource(topic) {	2✔
174	TopicCache
175	.of(
176	eventual,
177	topic,
178	log1,
179	consumer,
180	config,
181	consRecordToActionHeader,
182	metrics.map { _.headCache })
183	.map { cache =>
184	metrics
185	.fold(cache) { metrics => cache.withMetrics(topic, metrics.headCache) }
186	.withLog(log1)
187	}
188	}
189	.flatMap { cache =>	2✔
190	cache
191	.get(key.id, partition, offset)	1✔
192	.flatMap { _.toNow }	1✔
193	.map {	1✔
194	case a: Result.Now.Value => a.value.some	1✔
195	case Result.Now.Ahead => HeadInfo.empty.some	1✔
196	case Result.Now.Limited => none	×
197	case _: Result.Now.Timeout => none	1✔
198	}
199	}
200	}
201	}
202	}
203	}
204
205
206	/** Lighweight wrapper over [[EventualJournal]].
207	*
208	* Allows easier stubbing in unit tests.
209	*/
210	trait Eventual[F[_]] {
211
212	/** Gets the last replicated offset for a partition topic.
213	*
214	* @see [[EventualJournal#offset]] for more details.
215	*/
216	def pointer(topic: Topic, partition: Partition): F[Option[Offset]]
217
218	}
219
220	object Eventual {
221
222	def apply[F[_]](implicit F: Eventual[F]): Eventual[F] = F
223
224	/** Wrap [[EventualJournal]] into [[Eventual]] */
225	def apply[F[_]](eventualJournal: EventualJournal[F]): Eventual[F] = {
226	class Main
227	new Main with HeadCache.Eventual[F] {	1✔
228	def pointer(topic: Topic, partition: Partition): F[Option[Offset]] = eventualJournal.offset(topic, partition)	1✔
229	}
230	}
231
232	/** Always return `None` as an offset, i.e. pretend nothing ever replicates.
233	*
234	* Only useful for testing purposes.
235	*/
236	def empty[F[_]: Applicative]: Eventual[F] = const(TopicPointers.empty.pure[F])	1✔
237
238	/** Ignore topic and specify offset to return by partition.
239	*
240	* Only useful for testing purposes.
241	*/
242	def const[F[_]: Applicative](value: F[TopicPointers]): Eventual[F] = {
243	class Const
244	new Const with Eventual[F] {	1✔
245	def pointer(topic: Topic, partition: Partition): F[Option[Offset]] = value.map(_.values.get(partition))	1✔
246	}
247	}
248	}
249
250	private abstract sealed class WithFence
251
252	private abstract sealed class WithLog
253
254	implicit class HeadCacheOps[F[_]](val self: HeadCache[F]) extends AnyVal {
255
256	def mapK[G[_]](f: F ~> G): HeadCache[G] = new HeadCache[G] {	×
257
258	def get(key: Key, partition: Partition, offset: Offset) = {
259	f(self.get(key, partition, offset))	×
260	}
261	}
262
263	/** Log debug messages on every call to the class methods.
264	*
265	* The messages will go to DEBUG level, so it is also necessary to enable
266	* it in logger configuration.
267	*/
268	def withLog(log: Log[F])(implicit F: FlatMap[F], measureDuration: MeasureDuration[F]): HeadCache[F] = {
269	new WithLog with HeadCache[F] {	1✔
270
271	def get(key: Key, partition: Partition, offset: Offset) = {
272	for {
273	d <- MeasureDuration[F].start	1✔
274	a <- self.get(key, partition, offset)	1✔
275	d <- d	1✔
276	_ <- log.debug(s"get in ${ d.toMillis }ms, key: $key, offset: $partition:$offset, result: $a")	1✔
277	} yield a	1✔
278	}
279	}
280	}
281
282	/** Prevents cache methods to be used after `Resource` was released.
283	*
284	* [[ReleasedError]] will be raised if [[HeadCache#get]] is called after
285	* resource is released.
286	*
287	* It may prevent certain kind of bugs, when several caches are,
288	* accidentially, alive and working.
289	*/
290	def withFence(implicit F: Sync[F]): Resource[F, HeadCache[F]] = {
291	Resource
292	.make { Ref[F].of(().pure[F]) } { _.set(ReleasedError.raiseError[F, Unit]) }	1✔
293	.map { ref =>	1✔
294	new WithFence with HeadCache[F] {	1✔
295	def get(key: Key, partition: Partition, offset: Offset) = {
296	ref
297	.get
298	.flatten	1✔
299	.productR { self.get(key, partition, offset) }	1✔
300	}
301	}
302	}
303	}
304	}
305
306
307	/** Provides methods to update the metrics for [[HeadCache]] internals */
308	trait Metrics[F[_]] {
309
310	/** Report duration and result of cache hits, i.e. [[TopicCache#get]].
311	*
312	* @param topic
313	* Topic journal is being stored in.
314	* @param latency
315	* Duration of [[TopicCache#get]] call.
316	* @param result
317	* Result of the call, i.e. "ahead", "limited", "timeout" or "failure".
318	* @param now
319	* If result was [[PartitionCache.Result.Now]], i.e. entry was already in
320	* cache.
321	*/
322	def get(topic: Topic, latency: FiniteDuration, result: String, now: Boolean): F[Unit]
323
324	/** Report health of all [[PartitionCache]] instances related to a topic.
325	*
326	* @param topic
327	* Topic which these [[PartitionCache]] instances are related to.
328	* @param entries
329	* Number of distinct journals stored in a topic cache. If it is too
330	* close to [[HeadCacheConfig.Partition#maxSize]] multiplied by number of
331	* partitions, the cache might not work efficiently.
332	* @param listeners
333	* Number of listeners waiting after [[PartitionCache#get]] call. Too
334	* many of them might mean that cache is not being loaded fast enough.
335	*/
336	def meters(topic: Topic, entries: Int, listeners: Int): F[Unit]
337
338	/** Report the latency and number of records coming from Kafka.
339	*
340	* I.e. how long it took for a next element in a stream returned by
341	* [[HeadCacheConsumption#apply]] to get from a journal writer to this
342	* cache.
343	*
344	* @param topic
345	* Topic being read by [[HeadCacheConsumption]].
346	* @param age
347	* Time it took for an element to reach [[HeadCache]].
348	* @param diff
349	* The number of elements added to cache by this batch, i.e. returned by
350	* [[PartitionCache#add]].
351	*/
352	def consumer(topic: Topic, age: FiniteDuration, diff: Long): F[Unit]
353
354	/** Report the number of records coming from Cassandra.
355	*
356	* @param topic
357	* Topic being read by [[Eventual]].
358	* @param diff
359	* The number of elements remove from cache by this batch, i.e. returned
360	* by [[PartitionCache#remove]].
361	*/
362	def storage(topic: Topic, diff: Long): F[Unit]
363	}
364
365	object Metrics {
366
367	/** Does not do anything, ignores metric reports */
368	def empty[F[_]: Applicative]: Metrics[F] = const(().pure[F])	1✔
369
370
371	/** Calls a passed effect when metrics are reported.
372	*
373	* May only be useful for tests, as the reported parameters are ignored.
374	*/
375	def const[F[_]](unit: F[Unit]): Metrics[F] = {
376	class Const
377	new Const with Metrics[F] {	1✔
378
379	def get(topic: Topic, latency: FiniteDuration, result: String, hit: Boolean) = unit
380
381	def meters(topic: Topic, entries: Int, listeners: Int) = unit
382
383	def consumer(topic: Topic, age: FiniteDuration, diff: Long) = unit
384
385	def storage(topic: Topic, diff: Long) = unit
386	}
387	}
388
389
390	type Prefix = String
391
392	object Prefix {
393	val default: Prefix = "headcache"	1✔
394	}
395
396
397	/** Registers a default set of metrics to a passed collector registry.
398	*
399	* Note, that creating this metrics several times with the same collector
400	* registry may cause errors unless previous [[Metrics]] instance was not
401	* released yet.
402	*
403	* The following metrics will be registered by default, but it is possible
404	* to override the default `headcache` prefix to something else.
405	*
406	* {{{
407	* headcache_get_latency HeadCache get latency in seconds
408	* headcache_get_result HeadCache `get` call result counter
409	* headcache_entries HeadCache entries
410	* headcache_listeners HeadCache listeners
411	* headcache_records_age HeadCache time difference between record timestamp and now in seconds
412	* headcache_diff HeadCache offset difference between state and source
413	* }}}
414	*
415	* @param registry
416	* smetrics collector registry.
417	* @param prefix
418	* Prefix to use for the registered metrics.
419	*/
420	def of[F[_]: Monad](
421	registry: CollectorRegistry[F],
422	prefix: Prefix = Prefix.default
423	): Resource[F, Metrics[F]] = {
424
425	val getLatencySummary = registry.summary(	1✔
426	name = s"${ prefix }_get_latency",
427	help = "HeadCache get latency in seconds",	1✔
428	quantiles = Quantiles.Default,	1✔
429	labels = LabelNames("topic", "result", "now"))	1✔
430
431	val getResultCounter = registry.counter(	1✔
432	name = s"${ prefix }_get_result",
433	help = "HeadCache `get` call result counter",	1✔
434	labels = LabelNames("topic", "result", "now"))	1✔
435
436	val entriesGauge = registry.gauge(	1✔
437	name = s"${ prefix }_entries",
438	help = "HeadCache entries",	1✔
439	labels = LabelNames("topic"))	1✔
440
441	val listenersGauge = registry.gauge(	1✔
442	name = s"${ prefix }_listeners",
443	help = "HeadCache listeners",	1✔
444	labels = LabelNames("topic"))	1✔
445
446	val ageSummary = registry.summary(	1✔
447	name = s"${ prefix }_records_age",
448	help = "HeadCache time difference between record timestamp and now in seconds",	1✔
449	quantiles = Quantiles.Default,	1✔
450	labels = LabelNames("topic"))	1✔
451
452	val diffSummary = registry.summary(	1✔
453	name = s"${ prefix }_diff",
454	help = "HeadCache offset difference between state and source",	1✔
455	quantiles = Quantiles.Default,	1✔
456	labels = LabelNames("topic", "source"))	1✔
457
458	for {
459	getLatencySummary <- getLatencySummary	1✔
460	getResultCounter <- getResultCounter	1✔
461	entriesGauge <- entriesGauge	1✔
462	listenersGauge <- listenersGauge	1✔
463	ageSummary <- ageSummary	1✔
464	diffSummary <- diffSummary	1✔
465	} yield {
466
467	class Main
468	new Main with Metrics[F] {	1✔
469
470	def get(topic: Topic, latency: FiniteDuration, result: String, now: Boolean) = {
471	val nowLabel = if (now) "now" else "later"	1✔
472	for {
473	_ <- getLatencySummary	1✔
474	.labels(topic, result, nowLabel)
475	.observe(latency.toNanos.nanosToSeconds)	1✔
476	a <- getResultCounter	1✔
477	.labels(topic, result, nowLabel)	1✔
478	.inc()	1✔
479	} yield a
480	}
481
482	def meters(topic: Topic, entries: Int, listeners: Int) = {
483	for {
484	_ <- listenersGauge	×
485	.labels(topic)
486	.set(listeners.toDouble)	×
487	a <- entriesGauge	×
488	.labels(topic)
489	.set(entries.toDouble)	×
490	} yield a
491	}
492
493	def consumer(topic: Topic, age: FiniteDuration, diff: Long) = {
494	for {
495	_ <- ageSummary	1✔
496	.labels(topic)
497	.observe(age.toNanos.nanosToSeconds)	1✔
UNCOV 498	a <- diffSummary	×
UNCOV 499	.labels(topic, "kafka")	×
UNCOV 500	.observe(diff.toDouble)	×
501	} yield a
502	}
503
504	def storage(topic: Topic, diff: Long) = {
505	diffSummary
506	.labels(topic, "storage")	×
507	.observe(diff.toDouble)	×
508	}
509	}
510	}
511	}
512	}
513	}

evolution-gaming / kafka-journal / 8094751572

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