9893622978

Committed 11 Jul 2024 02:48PM UTC coverage: 81.102% (-0.05%) from 81.151%

Build Type

push

Committed by

Commit Message

Update sbt to 1.10.1 (#643)

## About this PR
📦 Updates [org.scala-sbt:sbt](https://github.com/sbt/sbt) from `1.10.0`
to `1.10.1`

📜 [GitHub Release
Notes](https://github.com/sbt/sbt/releases/tag/v1.10.1) - [Version
Diff](https://github.com/sbt/sbt/compare/v1.10.0...v1.10.1)

## Usage
✅ **Please merge!**

I'll automatically update this PR to resolve conflicts as long as you
don't change it yourself.

If you'd like to skip this version, you can just close this PR. If you
have any feedback, just mention me in the comments below.

Configure Scala Steward for your repository with a
[`.scala-steward.conf`](https://github.com/scala-steward-org/scala-steward/blob/902bdd228/docs/repo-specific-configuration.md)
file.

_Have a fantastic day writing Scala!_

<details>
<summary>⚙ Adjust future updates</summary>

Add this to your `.scala-steward.conf` file to ignore future updates of
this dependency:
```
updates.ignore = [ { groupId = "org.scala-sbt", artifactId = "sbt" } ]
```
Or, add this to slow down future updates of this dependency:
```
dependencyOverrides = [{
  pullRequests = { frequency = "30 days" },
  dependency = { groupId = "org.scala-sbt", artifactId = "sbt" }
}]
```
</details>

<sup>
labels: library-update, early-semver-patch, semver-spec-patch,
version-scheme:early-semver, commit-count:1
</sup>

Run Details

3313 of 4085 relevant lines covered (81.1%)

1.36 hits per line

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

82.93

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

package com.evolutiongaming.kafka.journal

import cats.*
import cats.effect.syntax.all.*
import cats.effect.{Async, *}
import cats.syntax.all.*
import com.evolution.scache.{Cache, ExpiringCache}
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.evolutiongaming.skafka.consumer.ConsumerConfig
import com.evolutiongaming.skafka.{Offset, Partition, Topic}
import com.evolutiongaming.smetrics.*
import com.evolutiongaming.smetrics.MetricsHelper.*

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

evolution-gaming / kafka-journal / 9893622978

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