5289615050

Committed 16 Jun 2023 11:52AM UTC coverage: 71.429% (-0.2%) from 71.579%

Build Type

push

Committed by

Commit Message

Update dependencies, migrate to cats-helper MeasureDuration (#3) (#4)

* Update dependencies, migrate to cats-helper MeasureDuration

* Fix scala versions in CI

* Bump plugin versions

(cherry picked from commit fb5255b0a)

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

70 of 98 relevant lines covered (71.43%)

1.39 hits per line

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

69.66

/src/main/scala/com/evolutiongaming/cluster/ddata/SafeReplicator.scala

package com.evolutiongaming.cluster.ddata

import akka.actor.{Actor, ActorLogging, ActorRef, ActorRefFactory, ActorSystem, Props}
import akka.cluster.ddata.Replicator.{ReadConsistency, WriteConsistency}
import akka.cluster.ddata.{DistributedData, Flag, GCounter, GSet, Key, LWWMap, ManyVersionVector, ORMap, ORMultiMap, ORSet, OneVersionVector, PNCounter, PNCounterMap, ReplicatedData, VersionVector, Replicator => R}
import akka.pattern.ask
import akka.util.Timeout
import cats.effect.{Resource, Sync}
import cats.implicits._
import cats.{Applicative, Monad}
import com.evolutiongaming.catshelper.{FromFuture, MeasureDuration, ToFuture}
import com.evolutiongaming.cluster.ddata.{ReplicatorError => E}
import com.evolutiongaming.smetrics
import com.evolutiongaming.smetrics.MetricsHelper._
import com.evolutiongaming.smetrics.{MeasureDuration => _, _}

import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ExecutionContext, Future}

/**
  * Typesafe api for [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html]]
  *
  * Akka Distributed Data is useful when you need to share data between nodes in an Akka Cluster.
  * The data is accessed with an actor providing a key-value store like API.
  * The keys are unique identifiers with type information of the data values.
  * The values are Conflict Free Replicated Data Types (CRDTs).
  * All data entries are spread to all nodes, or nodes with a certain role, in the cluster via direct replication and gossip based dissemination.
  * You have fine grained control of the consistency level for reads and writes.
  */
trait SafeReplicator[F[_], A <: ReplicatedData] {

  /**
    * [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#get]]
    *
    * To retrieve the current value of a data
    *
    * @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
    */
  def get(implicit consistency: ReadConsistency): F[Option[A]]

  /**
    * [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#update]]
    *
    * To modify and replicate a data value
    * The current data value for the key of the Update is passed as parameter to the modify function of the Update.
    * The function is supposed to return the new value of the data, which will then be replicated according to the given consistency level.
    *
    * @param modify      The modify function is called by the Replicator actor and must therefore be a pure function
    *                    that only uses the data parameter and stable fields from enclosing scope.
    *                    It must for example not access the sender (sender()) reference of an enclosing actor.
    * @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
    */
  def update(modify: Option[A] => A)(implicit consistency: WriteConsistency): F[Unit]

  /**
    * [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#delete]]
    *
    * A deleted key cannot be reused again,
    * but it is still recommended to delete unused data entries because that reduces the replication overhead when new nodes join the cluster.
    * Subsequent delete, update and get calls will fail with Deleted or AlreadyDeleted. Subscribers will stop
    *
    * @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
    */
  def delete(implicit consistency: WriteConsistency): F[Boolean]

  /**
    * [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#subscribe]]
    *
    * Subscribers will be notified periodically with the configured notify-subscribers-interval
    *
    * @param onStop    called when the subscriber is stopped or data is deleted
    * @param onChanged called when the data is updated.
    * @return Unsubscribe function, you should call to unregister
    */
  def subscribe(
    onStop: F[Unit],
    onChanged: A => F[Unit])(implicit
    factory: ActorRefFactory,
    executor: ExecutionContext
  ): Resource[F, Unit]

  /**
    * Notifies the subscribers immediately, to not wait for periodic updates
    */
  def flushChanges: F[Unit]
}

object SafeReplicator {

  def of[F[_] : Sync : FromFuture : ToFuture, A <: ReplicatedData](
    key: Key[A],
    timeout: FiniteDuration)(implicit
    system: ActorSystem
  ): F[SafeReplicator[F, A]] = {

    for {
      replicator <- Sync[F].delay { DistributedData(system).replicator }
    } yield {
      apply[F, A](key, timeout, replicator)
    }
  }

  def apply[F[_] : Sync : FromFuture : ToFuture, A <: ReplicatedData](
    key: Key[A],
    timeout: FiniteDuration,
    replicator: ActorRef
  ): SafeReplicator[F, A] = {

    def askF(cmd: R.Command[A]): F[Any] = {
      FromFuture[F].apply { replicator.ask(cmd)(Timeout(timeout), ActorRef.noSender) }
    }

    new SafeReplicator[F, A] {

      def get(implicit consistency: ReadConsistency) = {
        val get = R.Get(key, consistency)
        askF(get).flatMap {
          case a: R.GetResponse[_] => a.asInstanceOf[R.GetResponse[A]] match {
            case a: R.GetSuccess[A]     => a.dataValue.some.pure[F]
            case _: R.NotFound[A]       => none[A].pure[F]
            case _: R.GetDataDeleted[A] => E.dataDeleted.raiseError[F, Option[A]]
            case _: R.GetFailure[A]     => E.getFailure.raiseError[F, Option[A]]
          }
          case _: R.DataDeleted[_] => E.dataDeleted.raiseError[F, Option[A]]
          case a                   => E.unknown(s"unknown reply $a").raiseError[F, Option[A]]
        }
      }


      def update(modify: Option[A] => A)(implicit consistency: WriteConsistency) = {
        val update = R.Update(key, consistency, None)(modify)
        askF(update).flatMap {
          case a: R.UpdateResponse[_] => a.asInstanceOf[R.UpdateResponse[A]] match {
            case _: R.UpdateSuccess[A]     => ().pure[F]
            case _: R.UpdateTimeout[A]     => E.timeout.raiseError[F, Unit]
            case a: R.ModifyFailure[A]     => E.modifyFailure(a.errorMessage, a.cause).raiseError[F, Unit]
            case _: R.StoreFailure[A]      => E.storeFailure.raiseError[F, Unit]
            case _: R.UpdateDataDeleted[_] => E.dataDeleted.raiseError[F, Unit]
          }
          case _: R.DataDeleted[_]    => E.dataDeleted.raiseError[F, Unit]
          case a                      => E.unknown(s"unknown reply $a").raiseError[F, Unit]
        }
      }


      def delete(implicit consistency: WriteConsistency) = {
        val delete = R.Delete(key, consistency)
        askF(delete).flatMap {
          case a: R.DeleteResponse[_] => a.asInstanceOf[R.DeleteResponse[A]] match {
            case _: R.DeleteSuccess[A]            => true.pure[F]
            case _: R.DataDeleted[A]              => false.pure[F]
            case _: R.ReplicationDeleteFailure[A] => E.replicationFailure.raiseError[F, Boolean]
            case _: R.StoreFailure[A]             => E.storeFailure.raiseError[F, Boolean]
          }
          case a                      => E.unknown(s"unknown reply $a").raiseError[F, Boolean]
        }
      }


      def subscribe(
        onStop: F[Unit],
        onChanged: A => F[Unit])(implicit
        factory: ActorRefFactory,
        executor: ExecutionContext
      ) = {

        def actor() = new Actor with ActorLogging {

          var future = Future.unit

          val handleError = (error: Throwable) => {
            Sync[F].delay { log.error(error, s"$key onChanged failed $error") }
          }

          private def rcvChanged(a: A): Unit = {
            val effect = onChanged(a).handleErrorWith(handleError)
            future = future.flatMap { _ => ToFuture[F].apply(effect) }
          }

          override def preStart(): Unit = {
            replicator.tell(R.Subscribe(key, self), self)
            super.preStart()
          }

          def receive = {
            case a @ R.Changed(`key`)    => rcvChanged(a.get(key))
            case R.DataDeleted(`key`, _) => context.stop(self)
            case a                       => log.warning(s"$key unexpected $a")
          }

          override def postStop(): Unit = {
            future.flatMap { _ => ToFuture[F].apply { onStop } }
            replicator.tell(R.Unsubscribe(key, self), self)
            super.postStop()
          }
        }

        val props = Props(actor())

        val result = for {
          ref <- Sync[F].delay { factory.actorOf(props) }
        } yield {
          val release = Sync[F].delay { factory.stop(ref) }
          ((), release)
        }
        Resource(result)
      }

      def flushChanges = {
        Sync[F].delay { replicator ! R.FlushChanges }
      }
    }
  }


  trait Metrics[F[_]] {

    def latency(name: String, latency: FiniteDuration): F[Unit]

    def size(size: Long): F[Unit]
  }

  object Metrics {

    def empty[F[_] : Applicative]: Metrics[F] = const(().pure[F])


    def const[F[_]](unit: F[Unit]): Metrics[F] = new Metrics[F] {

      def latency(name: String, latency: FiniteDuration) = unit

      def size(size: Long) = unit
    }


    type Prefix = String

    object Prefix {
      val Empty: Prefix = "ddata"
    }

    def of[F[_] : Monad](
      registry: CollectorRegistry[F],
      prefix: Prefix = Prefix.Empty
    ): Resource[F, String => SafeReplicator.Metrics[F]] = {

      val latencySummary = registry.summary(
        name = s"${ prefix }_latency",
        help = "Latency in seconds",
        quantiles = Quantiles(
          Quantile(0.9, 0.05),
          Quantile(0.99, 0.005)),
        labels = LabelNames("key", "name"))

      val sizeGauge = registry.gauge(
        name = s"${ prefix }_size",
        help = "Size of distributed data",
        labels = LabelNames("key"))

      for {
        latencySummary <- latencySummary
        sizeGauge      <- sizeGauge
      } yield {
        key: String =>
          new SafeReplicator.Metrics[F] {

            def latency(name: String, latency: FiniteDuration) = {
              latencySummary.labels(key, name).observe(latency.toNanos.nanosToSeconds)
            }

            def size(size: Long) = {
              sizeGauge.labels(key).set(size.toDouble)
            }
          }
      }
    }


    trait DataMetrics[F[_], A <: ReplicatedData] {

      def apply(metrics: Metrics[F], a: A): F[Unit]
    }

    object DataMetrics {

      implicit def gCounterDataSize[F[_] : Applicative]: DataMetrics[F, GCounter] = empty
      implicit def gSetDataSize[F[_] : Applicative, A]: DataMetrics[F, GSet[A]] = size { a: GSet[A] => a.size }

      implicit def lwwMapDataSize[F[_] : Applicative, K, V]: DataMetrics[F, LWWMap[K, V]] = size { a: LWWMap[K, V] => a.size }

      implicit def orMapDataSize[F[_] : Applicative, K, V <: ReplicatedData]: DataMetrics[F, ORMap[K, V]] = size { a: ORMap[_, _] => a.size }
      implicit def orMultiMapDataSize[F[_] : Applicative, K, V]: DataMetrics[F, ORMultiMap[K, V]] = size { a: ORMultiMap[K, V] => a.size }
      implicit def orSetDataSize[F[_] : Applicative, A]: DataMetrics[F, ORSet[A]] = size { a: ORSet[A] => a.size }

      implicit def pnCounterDataSize[F[_] : Applicative]: DataMetrics[F, PNCounter] = empty
      implicit def pnCounterMapDataSize[F[_] : Applicative, A]: DataMetrics[F, PNCounterMap[A]] = size { a: PNCounterMap[A] => a.size }

      implicit def flagDataSize[F[_] : Applicative]: DataMetrics[F, Flag] = empty

      implicit def versionVectorDataSize[F[_] : Applicative]: DataMetrics[F, VersionVector] = empty
      implicit def oneVersionVectorDataSize[F[_] : Applicative]: DataMetrics[F, OneVersionVector] = empty
      implicit def manyVersionVectorDataSize[F[_] : Applicative]: DataMetrics[F, ManyVersionVector] = size { a: ManyVersionVector => a.versions.size }


      def empty[F[_] : Applicative, A <: ReplicatedData]: DataMetrics[F, A] = new DataMetrics[F, A] {
        def apply(metrics: Metrics[F], a: A) = ().pure[F]
      }

      private def size[F[_], A <: ReplicatedData](size: A => Int): DataMetrics[F, A] = new DataMetrics[F, A] {
        def apply(metrics: Metrics[F], a: A) = metrics.size(size(a).toLong)
      }
    }
  }


  implicit class SafeReplicatorOps[F[_], A <: ReplicatedData](val self: SafeReplicator[F, A]) extends AnyVal {

    @deprecated("Use `withMetrics1` instead", "3.1.0")
    def withMetrics(
      metrics: Metrics[F],
      refFactory: ActorRefFactory)(implicit
      dataMetrics: Metrics.DataMetrics[F, A],
      F: Sync[F],
      measureDuration: smetrics.MeasureDuration[F]
    ): Resource[F, SafeReplicator[F, A]] = {
      withMetrics1(metrics, refFactory)(dataMetrics, F, measureDuration.toCatsHelper)
    }

    def withMetrics1(
      metrics: Metrics[F],
      refFactory: ActorRefFactory)(implicit
      dataMetrics: Metrics.DataMetrics[F, A],
      F: Sync[F],
      measureDuration: MeasureDuration[F]
    ): Resource[F, SafeReplicator[F, A]] = {

      def latency[B](name: String)(fa: F[B]): F[B] = {
        for {
          d <- measureDuration.start
          b <- fa.attempt
          d <- d
          _ <- metrics.latency(name, d)
          b <- b.liftTo[F]
        } yield b
      }

      val onChanged = (a: A) => dataMetrics(metrics, a)

      for {
        _ <- self.subscribe(().pure[F], onChanged)(refFactory, refFactory.dispatcher)
      } yield {

        new SafeReplicator[F, A] {

          def get(implicit consistency: ReadConsistency) = {
            latency("get") { self.get }
          }

          def update(modify: Option[A] => A)(implicit consistency: WriteConsistency) = {
            latency("update") { self.update(modify) }
          }

          def delete(implicit consistency: WriteConsistency) = {
            latency("delete") { self.delete }
          }

          def subscribe(
            onStop: F[Unit],
            onChanged: A => F[Unit])(implicit
            factory: ActorRefFactory,
            executor: ExecutionContext
          ) = {
            val result = latency("subscribe") { self.subscribe(onStop, onChanged).allocated }
            Resource(result)
          }

          def flushChanges = {
            latency("flushChanges") { self.flushChanges }
          }
        }
      }
    }
  }
}

1	package com.evolutiongaming.cluster.ddata
2
3	import akka.actor.{Actor, ActorLogging, ActorRef, ActorRefFactory, ActorSystem, Props}
4	import akka.cluster.ddata.Replicator.{ReadConsistency, WriteConsistency}
5	import akka.cluster.ddata.{DistributedData, Flag, GCounter, GSet, Key, LWWMap, ManyVersionVector, ORMap, ORMultiMap, ORSet, OneVersionVector, PNCounter, PNCounterMap, ReplicatedData, VersionVector, Replicator => R}
6	import akka.pattern.ask
7	import akka.util.Timeout
8	import cats.effect.{Resource, Sync}
9	import cats.implicits._
10	import cats.{Applicative, Monad}
11	import com.evolutiongaming.catshelper.{FromFuture, MeasureDuration, ToFuture}
12	import com.evolutiongaming.cluster.ddata.{ReplicatorError => E}
13	import com.evolutiongaming.smetrics
14	import com.evolutiongaming.smetrics.MetricsHelper._
15	import com.evolutiongaming.smetrics.{MeasureDuration => _, _}
16
17	import scala.concurrent.duration.FiniteDuration
18	import scala.concurrent.{ExecutionContext, Future}
19
20	/**
21	* Typesafe api for [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html]]
22	*
23	* Akka Distributed Data is useful when you need to share data between nodes in an Akka Cluster.
24	* The data is accessed with an actor providing a key-value store like API.
25	* The keys are unique identifiers with type information of the data values.
26	* The values are Conflict Free Replicated Data Types (CRDTs).
27	* All data entries are spread to all nodes, or nodes with a certain role, in the cluster via direct replication and gossip based dissemination.
28	* You have fine grained control of the consistency level for reads and writes.
29	*/
30	trait SafeReplicator[F[_], A <: ReplicatedData] {
31
32	/**
33	* [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#get]]
34	*
35	* To retrieve the current value of a data
36	*
37	* @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
38	*/
39	def get(implicit consistency: ReadConsistency): F[Option[A]]
40
41	/**
42	* [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#update]]
43	*
44	* To modify and replicate a data value
45	* The current data value for the key of the Update is passed as parameter to the modify function of the Update.
46	* The function is supposed to return the new value of the data, which will then be replicated according to the given consistency level.
47	*
48	* @param modify The modify function is called by the Replicator actor and must therefore be a pure function
49	* that only uses the data parameter and stable fields from enclosing scope.
50	* It must for example not access the sender (sender()) reference of an enclosing actor.
51	* @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
52	*/
53	def update(modify: Option[A] => A)(implicit consistency: WriteConsistency): F[Unit]
54
55	/**
56	* [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#delete]]
57	*
58	* A deleted key cannot be reused again,
59	* but it is still recommended to delete unused data entries because that reduces the replication overhead when new nodes join the cluster.
60	* Subsequent delete, update and get calls will fail with Deleted or AlreadyDeleted. Subscribers will stop
61	*
62	* @param consistency [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#consistency]]
63	*/
64	def delete(implicit consistency: WriteConsistency): F[Boolean]
65
66	/**
67	* [[https://doc.akka.io/docs/akka/2.5.22/distributed-data.html#subscribe]]
68	*
69	* Subscribers will be notified periodically with the configured notify-subscribers-interval
70	*
71	* @param onStop called when the subscriber is stopped or data is deleted
72	* @param onChanged called when the data is updated.
73	* @return Unsubscribe function, you should call to unregister
74	*/
75	def subscribe(
76	onStop: F[Unit],
77	onChanged: A => F[Unit])(implicit
78	factory: ActorRefFactory,
79	executor: ExecutionContext
80	): Resource[F, Unit]
81
82	/**
83	* Notifies the subscribers immediately, to not wait for periodic updates
84	*/
85	def flushChanges: F[Unit]
86	}
87
88	object SafeReplicator {
89
90	def of[F[_] : Sync : FromFuture : ToFuture, A <: ReplicatedData](
91	key: Key[A],
92	timeout: FiniteDuration)(implicit
93	system: ActorSystem
94	): F[SafeReplicator[F, A]] = {
95
96	for {
97	replicator <- Sync[F].delay { DistributedData(system).replicator }	×
98	} yield {
99	apply[F, A](key, timeout, replicator)	×
100	}
101	}
102
103	def apply[F[_] : Sync : FromFuture : ToFuture, A <: ReplicatedData](
104	key: Key[A],
105	timeout: FiniteDuration,
106	replicator: ActorRef
107	): SafeReplicator[F, A] = {
108
109	def askF(cmd: R.Command[A]): F[Any] = {
110	FromFuture[F].apply { replicator.ask(cmd)(Timeout(timeout), ActorRef.noSender) }	2✔
111	}
112
113	new SafeReplicator[F, A] {	2✔
114
115	def get(implicit consistency: ReadConsistency) = {
116	val get = R.Get(key, consistency)	2✔
117	askF(get).flatMap {	2✔
118	case a: R.GetResponse[_] => a.asInstanceOf[R.GetResponse[A]] match {	2✔
119	case a: R.GetSuccess[A] => a.dataValue.some.pure[F]	2✔
120	case _: R.NotFound[A] => none[A].pure[F]	2✔
121	case _: R.GetDataDeleted[A] => E.dataDeleted.raiseError[F, Option[A]]	×
122	case _: R.GetFailure[A] => E.getFailure.raiseError[F, Option[A]]	2✔
123	}
124	case _: R.DataDeleted[_] => E.dataDeleted.raiseError[F, Option[A]]	2✔
125	case a => E.unknown(s"unknown reply $a").raiseError[F, Option[A]]	×
126	}
127	}
128
129
130	def update(modify: Option[A] => A)(implicit consistency: WriteConsistency) = {
131	val update = R.Update(key, consistency, None)(modify)	2✔
132	askF(update).flatMap {	2✔
133	case a: R.UpdateResponse[_] => a.asInstanceOf[R.UpdateResponse[A]] match {	2✔
134	case _: R.UpdateSuccess[A] => ().pure[F]	2✔
135	case _: R.UpdateTimeout[A] => E.timeout.raiseError[F, Unit]	2✔
136	case a: R.ModifyFailure[A] => E.modifyFailure(a.errorMessage, a.cause).raiseError[F, Unit]	2✔
137	case _: R.StoreFailure[A] => E.storeFailure.raiseError[F, Unit]	2✔
138	case _: R.UpdateDataDeleted[_] => E.dataDeleted.raiseError[F, Unit]	×
139	}
140	case _: R.DataDeleted[_] => E.dataDeleted.raiseError[F, Unit]	2✔
141	case a => E.unknown(s"unknown reply $a").raiseError[F, Unit]	×
142	}
143	}
144
145
146	def delete(implicit consistency: WriteConsistency) = {
147	val delete = R.Delete(key, consistency)	2✔
148	askF(delete).flatMap {	2✔
149	case a: R.DeleteResponse[_] => a.asInstanceOf[R.DeleteResponse[A]] match {	2✔
150	case _: R.DeleteSuccess[A] => true.pure[F]	2✔
151	case _: R.DataDeleted[A] => false.pure[F]	2✔
152	case _: R.ReplicationDeleteFailure[A] => E.replicationFailure.raiseError[F, Boolean]	2✔
153	case _: R.StoreFailure[A] => E.storeFailure.raiseError[F, Boolean]	2✔
154	}
155	case a => E.unknown(s"unknown reply $a").raiseError[F, Boolean]	×
156	}
157	}
158
159
160	def subscribe(
161	onStop: F[Unit],
162	onChanged: A => F[Unit])(implicit
163	factory: ActorRefFactory,
164	executor: ExecutionContext
165	) = {
166
167	def actor() = new Actor with ActorLogging {	2✔
168
169	var future = Future.unit	2✔
170
171	val handleError = (error: Throwable) => {
172	Sync[F].delay { log.error(error, s"$key onChanged failed $error") }	×
173	}
174
175	private def rcvChanged(a: A): Unit = {
176	val effect = onChanged(a).handleErrorWith(handleError)	2✔
177	future = future.flatMap { _ => ToFuture[F].apply(effect) }	2✔
178	}
179
180	override def preStart(): Unit = {
181	replicator.tell(R.Subscribe(key, self), self)	2✔
182	super.preStart()	2✔
183	}
184
185	def receive = {	2✔
186	case a @ R.Changed(`key`) => rcvChanged(a.get(key))	2✔
187	case R.DataDeleted(`key`, _) => context.stop(self)	2✔
188	case a => log.warning(s"$key unexpected $a")	×
189	}
190
191	override def postStop(): Unit = {
192	future.flatMap { _ => ToFuture[F].apply { onStop } }	2✔
193	replicator.tell(R.Unsubscribe(key, self), self)	2✔
194	super.postStop()	2✔
195	}
196	}
197
198	val props = Props(actor())	2✔
199
200	val result = for {
201	ref <- Sync[F].delay { factory.actorOf(props) }	2✔
202	} yield {
203	val release = Sync[F].delay { factory.stop(ref) }	2✔
204	((), release)	2✔
205	}
206	Resource(result)	2✔
207	}
208
209	def flushChanges = {
210	Sync[F].delay { replicator ! R.FlushChanges }	2✔
211	}
212	}
213	}
214
215
216	trait Metrics[F[_]] {
217
218	def latency(name: String, latency: FiniteDuration): F[Unit]
219
220	def size(size: Long): F[Unit]
221	}
222
223	object Metrics {
224
225	def empty[F[_] : Applicative]: Metrics[F] = const(().pure[F])	×
226
227
228	def const[F[_]](unit: F[Unit]): Metrics[F] = new Metrics[F] {	×
229
230	def latency(name: String, latency: FiniteDuration) = unit
231
232	def size(size: Long) = unit
233	}
234
235
236	type Prefix = String
237
238	object Prefix {
239	val Empty: Prefix = "ddata"	2✔
240	}
241
242	def of[F[_] : Monad](
243	registry: CollectorRegistry[F],
244	prefix: Prefix = Prefix.Empty
245	): Resource[F, String => SafeReplicator.Metrics[F]] = {
246
247	val latencySummary = registry.summary(	2✔
248	name = s"${ prefix }_latency",	1✔
249	help = "Latency in seconds",	2✔
250	quantiles = Quantiles(	2✔
251	Quantile(0.9, 0.05),	2✔
252	Quantile(0.99, 0.005)),	2✔
253	labels = LabelNames("key", "name"))	2✔
254
255	val sizeGauge = registry.gauge(	2✔
256	name = s"${ prefix }_size",	1✔
257	help = "Size of distributed data",	2✔
258	labels = LabelNames("key"))	2✔
259
260	for {
261	latencySummary <- latencySummary	2✔
262	sizeGauge <- sizeGauge	2✔
263	} yield {
264	key: String =>
265	new SafeReplicator.Metrics[F] {
266
267	def latency(name: String, latency: FiniteDuration) = {
268	latencySummary.labels(key, name).observe(latency.toNanos.nanosToSeconds)
269	}
270
271	def size(size: Long) = {
272	sizeGauge.labels(key).set(size.toDouble)
273	}
274	}
275	}
276	}
277
278
279	trait DataMetrics[F[_], A <: ReplicatedData] {
280
281	def apply(metrics: Metrics[F], a: A): F[Unit]
282	}
283
284	object DataMetrics {
285
286	implicit def gCounterDataSize[F[_] : Applicative]: DataMetrics[F, GCounter] = empty	2✔
287	implicit def gSetDataSize[F[_] : Applicative, A]: DataMetrics[F, GSet[A]] = size { a: GSet[A] => a.size }	×
288
289	implicit def lwwMapDataSize[F[_] : Applicative, K, V]: DataMetrics[F, LWWMap[K, V]] = size { a: LWWMap[K, V] => a.size }	×
290
291	implicit def orMapDataSize[F[_] : Applicative, K, V <: ReplicatedData]: DataMetrics[F, ORMap[K, V]] = size { a: ORMap[_, _] => a.size }	×
292	implicit def orMultiMapDataSize[F[_] : Applicative, K, V]: DataMetrics[F, ORMultiMap[K, V]] = size { a: ORMultiMap[K, V] => a.size }	×
293	implicit def orSetDataSize[F[_] : Applicative, A]: DataMetrics[F, ORSet[A]] = size { a: ORSet[A] => a.size }	×
294
295	implicit def pnCounterDataSize[F[_] : Applicative]: DataMetrics[F, PNCounter] = empty	×
296	implicit def pnCounterMapDataSize[F[_] : Applicative, A]: DataMetrics[F, PNCounterMap[A]] = size { a: PNCounterMap[A] => a.size }	×
297
298	implicit def flagDataSize[F[_] : Applicative]: DataMetrics[F, Flag] = empty	×
299
300	implicit def versionVectorDataSize[F[_] : Applicative]: DataMetrics[F, VersionVector] = empty	×
301	implicit def oneVersionVectorDataSize[F[_] : Applicative]: DataMetrics[F, OneVersionVector] = empty	×
302	implicit def manyVersionVectorDataSize[F[_] : Applicative]: DataMetrics[F, ManyVersionVector] = size { a: ManyVersionVector => a.versions.size }	×
303
304
305	def empty[F[_] : Applicative, A <: ReplicatedData]: DataMetrics[F, A] = new DataMetrics[F, A] {	2✔
306	def apply(metrics: Metrics[F], a: A) = ().pure[F]	×
307	}
308
309	private def size[F[_], A <: ReplicatedData](size: A => Int): DataMetrics[F, A] = new DataMetrics[F, A] {	×
310	def apply(metrics: Metrics[F], a: A) = metrics.size(size(a).toLong)	×
311	}
312	}
313	}
314
315
316	implicit class SafeReplicatorOps[F[_], A <: ReplicatedData](val self: SafeReplicator[F, A]) extends AnyVal {
317
318	@deprecated("Use `withMetrics1` instead", "3.1.0")
319	def withMetrics(
320	metrics: Metrics[F],
321	refFactory: ActorRefFactory)(implicit
322	dataMetrics: Metrics.DataMetrics[F, A],
323	F: Sync[F],
324	measureDuration: smetrics.MeasureDuration[F]
325	): Resource[F, SafeReplicator[F, A]] = {
326	withMetrics1(metrics, refFactory)(dataMetrics, F, measureDuration.toCatsHelper)	×
327	}
328
329	def withMetrics1(
330	metrics: Metrics[F],
331	refFactory: ActorRefFactory)(implicit
332	dataMetrics: Metrics.DataMetrics[F, A],
333	F: Sync[F],
334	measureDuration: MeasureDuration[F]
335	): Resource[F, SafeReplicator[F, A]] = {
336
337	def latency[B](name: String)(fa: F[B]): F[B] = {
338	for {
339	d <- measureDuration.start	2✔
340	b <- fa.attempt	2✔
341	d <- d	2✔
342	_ <- metrics.latency(name, d)	2✔
343	b <- b.liftTo[F]	2✔
344	} yield b
345	}
346
347	val onChanged = (a: A) => dataMetrics(metrics, a)	×
348
349	for {
350	_ <- self.subscribe(().pure[F], onChanged)(refFactory, refFactory.dispatcher)	2✔
351	} yield {
352
353	new SafeReplicator[F, A] {
354
355	def get(implicit consistency: ReadConsistency) = {
356	latency("get") { self.get }
357	}
358
359	def update(modify: Option[A] => A)(implicit consistency: WriteConsistency) = {
360	latency("update") { self.update(modify) }
361	}
362
363	def delete(implicit consistency: WriteConsistency) = {
364	latency("delete") { self.delete }
365	}
366
367	def subscribe(
368	onStop: F[Unit],
369	onChanged: A => F[Unit])(implicit
370	factory: ActorRefFactory,
371	executor: ExecutionContext
372	) = {
373	val result = latency("subscribe") { self.subscribe(onStop, onChanged).allocated }
374	Resource(result)
375	}
376
377	def flushChanges = {
378	latency("flushChanges") { self.flushChanges }
379	}
380	}
381	}
382	}
383	}
384	}

evolution-gaming / ddata-tools / 5289615050

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