Skip to content

Commit ea3ec60

Browse files
ghostdogprghostdogpr
authored
Add metrics (#116)
1 parent 83f09fa commit ea3ec60

File tree

5 files changed

+68
-10
lines changed

5 files changed

+68
-10
lines changed

‎entities/src/main/scala/com/devsisters/shardcake/Metrics.scala‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package com.devsisters.shardcake
2+
3+
import zio.metrics.Metric
4+
import zio.metrics.Metric.Gauge
5+
6+
object Metrics {
7+
val shards: Gauge[Double] = Metric.gauge("shardcake.shards")
8+
val entities: Gauge[Double] = Metric.gauge("shardcake.entities")
9+
val singletons: Gauge[Double] = Metric.gauge("shardcake.singletons")
10+
}

‎entities/src/main/scala/com/devsisters/shardcake/Sharding.scala‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,9 @@ class Sharding private (
6767
singletons.updateZIO { singletons =>
6868
ZIO.foreach(singletons) {
6969
case (name, run, None) =>
70-
ZIO.logDebug(s"Starting singleton $name") *> run.forkDaemon.map(fiber => (name, run, Some(fiber)))
70+
ZIO.logDebug(s"Starting singleton $name") *>
71+
Metrics.singletons.tagged("name", name).increment *>
72+
run.forkDaemon.map(fiber => (name, run, Some(fiber)))
7173
case other => ZIO.succeed(other)
7274
}
7375
}
@@ -80,7 +82,9 @@ class Sharding private (
8082
singletons.updateZIO { singletons =>
8183
ZIO.foreach(singletons) {
8284
case (name, run, Some(fiber)) =>
83-
ZIO.logDebug(s"Stopping singleton $name") *> fiber.interrupt.as((name, run, None))
85+
ZIO.logDebug(s"Stopping singleton $name") *>
86+
Metrics.singletons.tagged("name", name).decrement *>
87+
fiber.interrupt.as((name, run, None))
8488
case other => ZIO.succeed(other)
8589
}
8690
}
@@ -96,7 +100,8 @@ class Sharding private (
96100
ZIO
97101
.unlessZIO(isShuttingDown) {
98102
shardAssignments.update(shards.foldLeft(_) { case (map, shard) => map.updated(shard, address) }) *>
99-
startSingletonsIfNeeded <*
103+
Metrics.shards.incrementBy(shards.size) *>
104+
startSingletonsIfNeeded *>
100105
ZIO.logDebug(s"Assigned shards: $shards")
101106
}
102107
.unit
@@ -111,7 +116,8 @@ class Sharding private (
111116
_.entityManager.terminateEntitiesOnShards(shards) // this will return once all shards are terminated
112117
)
113118
) *>
114-
stopSingletonsIfNeeded <*
119+
Metrics.shards.decrementBy(shards.size) *>
120+
stopSingletonsIfNeeded *>
115121
ZIO.logDebug(s"Unassigned shards: $shards")
116122

117123
private[shardcake] def isEntityOnLocalShards(recipientType: RecipientType[_], entityId: String): UIO[Boolean] =
@@ -130,6 +136,9 @@ class Sharding private (
130136
): UIO[Unit] = {
131137
val assignments = assignmentsOpt.flatMap { case (k, v) => v.map(k -> _) }
132138
ZIO.logDebug("Received new shard assignments") *>
139+
Metrics.shards
140+
.set(assignmentsOpt.count { case (_, podOpt) => podOpt.contains(address) })
141+
.when(fromShardManager) *>
133142
(if (fromShardManager) shardAssignments.update(map => if (map.isEmpty) assignments else map)
134143
else
135144
shardAssignments.update(map =>

‎entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ package com.devsisters.shardcake.internal
22

33
import com.devsisters.shardcake.errors.EntityNotManagedByThisPod
44
import com.devsisters.shardcake._
5+
import zio.metrics.{ Metric, MetricState }
6+
import zio.metrics.MetricKeyType.Gauge
57
import zio.{ Config => _, _ }
68

79
import java.util.concurrent.TimeUnit
@@ -57,6 +59,8 @@ private[shardcake] object EntityManager {
5759
config: Config,
5860
entityMaxIdleTime: Option[Duration]
5961
) extends EntityManager[Req] {
62+
private val gauge = Metrics.entities.tagged("type", recipientType.name)
63+
6064
private def startExpirationFiber(entityId: String): UIO[Fiber[Nothing, Unit]] = {
6165
val maxIdleTime = entityMaxIdleTime getOrElse config.entityMaxIdleTime
6266

@@ -150,10 +154,12 @@ private[shardcake] object EntityManager {
150154
queue <- Queue.unbounded[Req]
151155
// start the expiration fiber
152156
expirationFiber <- startExpirationFiber(entityId)
157+
_ <- gauge.increment
153158
_ <- behavior(entityId, queue)
154159
.ensuring(
155160
// shutdown the queue when the fiber ends
156161
entities.update(_ - entityId) *>
162+
gauge.decrement *>
157163
entitiesLastReceivedAt.update(_ - entityId) *>
158164
queue.shutdown *>
159165
expirationFiber.interrupt

‎manager/src/main/scala/com/devsisters/shardcake/ManagerMetrics.scala‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
package com.devsisters.shardcake
2+
3+
import zio.metrics.Metric
4+
import zio.metrics.Metric.{ Counter, Gauge }
5+
6+
object ManagerMetrics {
7+
val pods: Gauge[Double] = Metric.gauge("shardcake.pods")
8+
val assignedShards: Gauge[Double] = Metric.gauge("shardcake.shards_assigned")
9+
val unassignedShards: Gauge[Double] = Metric.gauge("shardcake.shards_unassigned")
10+
11+
val rebalances: Counter[Long] = Metric.counter("shardcake.rebalances")
12+
val podHealthChecked: Counter[Long] = Metric.counter("shardcake.pod_health_checked")
13+
}

‎manager/src/main/scala/com/devsisters/shardcake/ShardManager.scala‎

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ class ShardManager(
3737
.succeed(OffsetDateTime.now())
3838
.map(cdt => state.copy(pods = state.pods.updated(pod.address, PodWithMetadata(pod, cdt))))
3939
)
40+
_ <- ManagerMetrics.pods.increment
4041
_ <- eventsHub.publish(ShardingEvent.PodRegistered(pod.address))
4142
_ <- ZIO.when(state.unassignedShards.nonEmpty)(rebalance(false))
4243
_ <- persistPods.forkDaemon
@@ -45,7 +46,8 @@ class ShardManager(
4546
def notifyUnhealthyPod(podAddress: PodAddress): UIO[Unit] =
4647
ZIO
4748
.whenZIO(stateRef.get.map(_.pods.contains(podAddress))) {
48-
eventsHub.publish(ShardingEvent.PodHealthChecked(podAddress)) *>
49+
ManagerMetrics.podHealthChecked.tagged("pod_address", podAddress.toString).increment *>
50+
eventsHub.publish(ShardingEvent.PodHealthChecked(podAddress)) *>
4951
ZIO.unlessZIO(healthApi.isAlive(podAddress))(
5052
ZIO.logWarning(s"$podAddress is not alive, unregistering") *> unregister(podAddress)
5153
)
@@ -73,6 +75,7 @@ class ShardManager(
7375
)
7476
)
7577
}
78+
_ <- ManagerMetrics.pods.decrement
7679
_ <- eventsHub.publish(ShardingEvent.PodUnregistered(podAddress))
7780
_ <- eventsHub
7881
.publish(ShardingEvent.ShardsUnassigned(podAddress, unassignments))
@@ -92,7 +95,8 @@ class ShardManager(
9295
decideAssignmentsForUnassignedShards(state)
9396
else decideAssignmentsForUnbalancedShards(state, config.rebalanceRate)
9497
areChanges = assignments.nonEmpty || unassignments.nonEmpty
95-
_ <- ZIO.logDebug(s"Rebalancing (rebalanceImmediately=$rebalanceImmediately)").when(areChanges)
98+
_ <- (ZIO.logDebug(s"Rebalancing (rebalanceImmediately=$rebalanceImmediately)") *>
99+
ManagerMetrics.rebalances.increment).when(areChanges)
96100
// ping pods first to make sure they are ready and remove those who aren't
97101
failedPingedPods <- ZIO
98102
.foreachPar(assignments.keySet ++ unassignments.keySet)(pod =>
@@ -114,9 +118,11 @@ class ShardManager(
114118
(podApi.unassignShards(pod, shards) *> updateShardsState(shards, None)).foldZIO(
115119
_ => ZIO.succeed((Set(pod), shards)),
116120
_ =>
117-
eventsHub
118-
.publish(ShardingEvent.ShardsUnassigned(pod, shards))
119-
.as((Set.empty, Set.empty))
121+
ManagerMetrics.assignedShards.tagged("pod_address", pod.toString).decrementBy(shards.size) *>
122+
ManagerMetrics.unassignedShards.incrementBy(shards.size) *>
123+
eventsHub
124+
.publish(ShardingEvent.ShardsUnassigned(pod, shards))
125+
.as((Set.empty, Set.empty))
120126
)
121127
}
122128
.map(_.unzip)
@@ -131,7 +137,12 @@ class ShardManager(
131137
.foreachPar(filteredAssignments.toList) { case (pod, shards) =>
132138
(podApi.assignShards(pod, shards) *> updateShardsState(shards, Some(pod))).foldZIO(
133139
_ => ZIO.succeed(Set(pod)),
134-
_ => eventsHub.publish(ShardingEvent.ShardsAssigned(pod, shards)).as(Set.empty)
140+
_ =>
141+
ManagerMetrics.assignedShards
142+
.tagged("pod_address", pod.toString)
143+
.incrementBy(shards.size) *>
144+
ManagerMetrics.unassignedShards.decrementBy(shards.size) *>
145+
eventsHub.publish(ShardingEvent.ShardsAssigned(pod, shards)).as(Set.empty)
135146
)
136147
}
137148
.map(_.flatten.toSet)
@@ -201,6 +212,15 @@ object ShardManager {
201212
filteredPods.map { case (k, v) => k -> PodWithMetadata(v, cdt) },
202213
(1 to config.numberOfShards).map(_ -> None).toMap ++ filteredAssignments
203214
)
215+
_ <- ManagerMetrics.pods.incrementBy(initialState.pods.size)
216+
_ <- ZIO.foreachDiscard(initialState.shards) { case (_, podAddressOpt) =>
217+
podAddressOpt match {
218+
case Some(podAddress) =>
219+
ManagerMetrics.assignedShards.tagged("pod_address", podAddress.toString).increment
220+
case None =>
221+
ManagerMetrics.unassignedShards.increment
222+
}
223+
}
204224
state <- Ref.Synchronized.make(initialState)
205225
rebalanceSemaphore <- Semaphore.make(1)
206226
eventsHub <- Hub.unbounded[ShardingEvent]

0 commit comments

Comments
 (0)