9680730492

Committed 26 Jun 2024 01:36PM UTC coverage: 81.049% (-0.02%) from 81.073%

Build Type

push

Committed by

Commit Message

Update testcontainers-scala-cassandra to 0.41.4 (#626)

## About this PR
📦 Updates
[com.dimafeng:testcontainers-scala-cassandra](https://github.com/testcontainers/testcontainers-scala)
from `0.41.3` to `0.41.4`

## 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/9d9460afc/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 = "com.dimafeng", artifactId = "testcontainers-scala-cassandra" } ]
```
Or, add this to slow down future updates of this dependency:
```
dependencyOverrides = [{
  pullRequests = { frequency = "30 days" },
  dependency = { groupId = "com.dimafeng", artifactId = "testcontainers-scala-cassandra" }
}]
```
</details>

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

Run Details

3353 of 4137 relevant lines covered (81.05%)

1.36 hits per line

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

84.44

/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
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	×
496	.labels(topic)
497	.observe(age.toNanos.nanosToSeconds)	×
498	a <- diffSummary	×
499	.labels(topic, "kafka")	×
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 / 9680730492

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