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.Address
23 import akka.actor.typed.Behavior
24 import akka.actor.typed.scaladsl.Behaviors
25 import akka.actor.typed.scaladsl.adapter.TypedActorSystemOps
26 import akka.cluster.Cluster
27 import org.make.constructr.coordination.Coordination
28 
29 import scala.concurrent.duration.DurationInt
30 import scala.util.{Failure, Success}
31 
32 object MakeDowningActor {
33   val name: String = "MakeDowningActor"
34 
35   sealed trait Protocol
36   case object AutoDown extends Protocol
37   final case class MembersReceived(members: Set[Address]) extends Protocol
38   final case class CoordinationError(e: Throwable) extends Protocol
39 
40   def apply(): Behavior[Protocol] = {
41     Behaviors.setup { context =>
42       val coordination = Coordination(context.system.name, context.system.toClassic)
43       Behaviors.withTimers { timers =>
44         timers.startTimerAtFixedRate(AutoDown, 10.seconds)
45 
46         Behaviors.receiveMessage {
47           case AutoDown =>
48             context.pipeToSelf(coordination.getNodes()) {
49               case Success(nodes) => MembersReceived(nodes)
50               case Failure(e)     => CoordinationError(e)
51             }
52             Behaviors.same
53           case MembersReceived(nodes) =>
54             val cluster = Cluster(context.system)
55             val members = cluster.state.members
56             members.foreach { member =>
57               if (!nodes.contains(member.uniqueAddress.address)) {
58                 context.log.warn(
59                   s"Downing node ${member.uniqueAddress.address.toString} since it is no longer present in coordination"
60                 )
61                 cluster.down(member.uniqueAddress.address)
62               }
63             }
64             Behaviors.same
65           case CoordinationError(e) =>
66             context.log.error("Error while retrieving nodes", e)
67             Behaviors.same
68         }
69       }
70     }
71   }
72 
73 }