1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.technical
21 
22 import akka.actor.typed.scaladsl.AskPattern.Askable
23 import akka.actor.typed.scaladsl.Behaviors
24 import akka.actor.typed.{Behavior, Scheduler}
25 import akka.cluster.sharding.ShardRegion.ClusterShardingStats
26 import akka.cluster.sharding.typed.scaladsl.EntityTypeKey
27 import akka.cluster.sharding.typed.{scaladsl, GetClusterShardingStats}
28 import akka.cluster.typed.Cluster
29 import akka.pattern.AskTimeoutException
30 import akka.util.Timeout
31 import com.typesafe.config.Config
32 import kamon.Kamon
33 import kamon.metric.Gauge
34 
35 import scala.concurrent.duration.{Duration, FiniteDuration}
36 import scala.jdk.CollectionConverters._
37 import scala.util.{Failure, Success}
38 
39 object ClusterShardingMonitor {
40   trait ClusterShardingMonitorConfiguration {
41     def refreshInterval: FiniteDuration
42     def statsTimeout: FiniteDuration
43     def shardedRegions: Seq[String]
44   }
45 
46   object ClusterShardingMonitorConfiguration {
47     def apply(config: Config): ClusterShardingMonitorConfiguration = {
48       new ClusterShardingMonitorConfiguration {
49         override val refreshInterval: FiniteDuration = {
50           Duration(
51             config.getDuration("kamon.cluster-sharding.refresh-interval").toMillis,
52             scala.concurrent.duration.MILLISECONDS
53           )
54         }
55 
56         override val statsTimeout: FiniteDuration = {
57           Duration(
58             config.getDuration("kamon.cluster-sharding.stats-timeout").toMillis,
59             scala.concurrent.duration.MILLISECONDS
60           )
61         }
62 
63         override val shardedRegions: Seq[String] = {
64           config.getStringList("kamon.cluster-sharding.regions").asScala.toSeq
65         }
66       }
67     }
68   }
69 
70   def apply(): Behavior[Protocol] = {
71     ActorSystemHelper.superviseWithBackoff(monitorClusterSharding())
72   }
73 
74   private def monitorClusterSharding(): Behavior[Protocol] = {
75     Behaviors.setup { context =>
76       val configuration = ClusterShardingMonitorConfiguration(context.system.settings.config)
77       val gauges: Map[String, ShardingGauges] = configuration.shardedRegions.map { region =>
78         region -> new ShardingGauges(region)
79       }.toMap
80 
81       implicit val timeout: Timeout = Timeout(configuration.statsTimeout * 2)
82       implicit val scheduler: Scheduler = context.system.scheduler
83 
84       Behaviors.withTimers { timers =>
85         timers.startTimerAtFixedRate(Monitor, configuration.refreshInterval)
86 
87         Behaviors.receiveMessage {
88           case Monitor =>
89             val shardingInfo = scaladsl.ClusterSharding(context.system)
90             gauges.foreachEntry {
91               case (region, regionGauges) =>
92                 val statsFuture = shardingInfo.shardState
93                   .ask(GetClusterShardingStats(EntityTypeKey(region), configuration.statsTimeout, _))
94 
95                 context.pipeToSelf(statsFuture) {
96                   case Success(stats) => StatsSuccess(regionGauges, stats)
97                   case Failure(e)     => StatsFailure(region, e)
98                 }
99             }
100             Behaviors.same
101           case StatsSuccess(regionGauges, stats) =>
102             val allProposals = stats.regions.values.map(_.stats.values.sum).sum
103             regionGauges.totalActorsCount.update(allProposals)
104 
105             val maybeRegion = stats.regions.get(Cluster(context.system).selfMember.address)
106             maybeRegion.foreach { shardStats =>
107               regionGauges.nodeActorsCount.update(shardStats.stats.values.sum)
108               regionGauges.nodeShardsCount.update(shardStats.stats.size)
109             }
110             Behaviors.same
111           case StatsFailure(region, e: AskTimeoutException) if e.getMessage.contains("terminated") =>
112             context.log.warn(s"Unable to retrieve stats for $region due to terminated actor: ${e.getMessage}")
113             Behaviors.same
114           case StatsFailure(region, e) =>
115             context.log.error(s"Unable to retrieve stats for $region:", e)
116             Behaviors.same
117 
118         }
119       }
120     }
121   }
122 
123   sealed trait Protocol
124   case object Monitor extends Protocol
125   final case class StatsSuccess(gauges: ShardingGauges, stats: ClusterShardingStats) extends Protocol
126   final case class StatsFailure(region: String, throwable: Throwable) extends Protocol
127 
128   val name: String = "actor-sharding-monitor"
129 
130   class ShardingGauges(name: String) {
131     val totalActorsCount: Gauge = Kamon.gauge("sharding-total-actors").withTag("region", name)
132     val nodeActorsCount: Gauge = Kamon.gauge("sharding-node-actors").withTag("region", name)
133     val nodeShardsCount: Gauge = Kamon.gauge("sharding-node-shards").withTag("region", name)
134   }
135 }