core/src/main/scala/org/apache/gearpump/cluster/main/Master.scala - incubator-retired-gearpump - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.gearpump.cluster.main

 import java.util.concurrent.TimeUnit

 import akka.actor._
 import akka.cluster.ClusterEvent._
 import akka.cluster.{MemberStatus, Member, Cluster}
 import akka.cluster.ddata.DistributedData
 import akka.cluster.singleton.{ClusterSingletonProxySettings, ClusterSingletonProxy, ClusterSingletonManagerSettings, ClusterSingletonManager}
 import com.typesafe.config.ConfigValueFactory
 import org.apache.gearpump.cluster.ClusterConfig
 import org.apache.gearpump.cluster.master.Master.MasterListUpdated
 import org.apache.gearpump.cluster.master.{Master => MasterActor, MasterNode}
 import org.apache.gearpump.util.Constants._
 import org.apache.gearpump.util.LogUtil.ProcessType
 import org.apache.gearpump.util.{AkkaApp, Constants, LogUtil}
 import org.slf4j.Logger

 import scala.collection.JavaConverters._
 import scala.collection.immutable
 import scala.concurrent.Await
 import scala.concurrent.duration._

 object Master extends AkkaApp with ArgumentsParser {

   private var LOG: Logger = LogUtil.getLogger(getClass)

   override def akkaConfig: Config = ClusterConfig.master()

   override val options: Array[(String, CLIOption[Any])] =
     Array("ip" -> CLIOption[String]("<master ip address>", required = true),
       "port" -> CLIOption("<master port>", required = true))

   override val description = "Start Master daemon"

   def main(akkaConf: Config, args: Array[String]): Unit = {

     this.LOG = {
       LogUtil.loadConfiguration(akkaConf, ProcessType.MASTER)
       LogUtil.getLogger(getClass)
     }

     val config = parse(args)
     master(config.getString("ip"), config.getInt("port"), akkaConf)
   }

   private def verifyMaster(master: String, port: Int, masters: Iterable[String]) = {
     masters.exists { hostPort =>
       hostPort == s"$master:$port"
     }
   }

   private def master(ip: String, port: Int, akkaConf: Config): Unit = {
     val masters = akkaConf.getStringList(Constants.GEARPUMP_CLUSTER_MASTERS).asScala

     if (!verifyMaster(ip, port, masters)) {
       LOG.error(s"The provided ip $ip and port $port doesn't conform with config at " +
         s"gearpump.cluster.masters: ${masters.mkString(", ")}")
       System.exit(-1)
     }

     val masterList = masters.map(master => s"akka.tcp://${MASTER}@$master").toList.asJava
     val quorum = masterList.size() / 2 + 1
     val masterConfig = akkaConf.
       withValue("akka.remote.netty.tcp.port", ConfigValueFactory.fromAnyRef(port)).
       withValue(NETTY_TCP_HOSTNAME, ConfigValueFactory.fromAnyRef(ip)).
       withValue("akka.cluster.seed-nodes", ConfigValueFactory.fromAnyRef(masterList)).
       withValue(s"akka.cluster.role.${MASTER}.min-nr-of-members",
         ConfigValueFactory.fromAnyRef(quorum))

     LOG.info(s"Starting Master Actor system $ip:$port, master list: ${masters.mkString(";")}")
     val system = ActorSystem(MASTER, masterConfig)

     val replicator = DistributedData(system).replicator
     LOG.info(s"Replicator path: ${replicator.path}")

     // Starts singleton manager
     val singletonManager = system.actorOf(ClusterSingletonManager.props(
       singletonProps = Props(classOf[MasterWatcher], MASTER),
       terminationMessage = PoisonPill,
       settings = ClusterSingletonManagerSettings(system).withSingletonName(MASTER_WATCHER)
         .withRole(MASTER)),
       name = SINGLETON_MANAGER)

     // Start master proxy
     val masterProxy = system.actorOf(ClusterSingletonProxy.props(
       singletonManagerPath = s"/user/${SINGLETON_MANAGER}",
       // The effective singleton is s"${MASTER_WATCHER}/$MASTER" instead of s"${MASTER_WATCHER}".
       // Master is created when there is a majority of machines started.
       settings = ClusterSingletonProxySettings(system)
         .withSingletonName(s"${MASTER_WATCHER}/$MASTER").withRole(MASTER)),
       name = MASTER
     )

     LOG.info(s"master proxy is started at ${masterProxy.path}")

     val mainThread = Thread.currentThread()
     Runtime.getRuntime().addShutdownHook(new Thread() {
       override def run(): Unit = {
         if (!system.whenTerminated.isCompleted) {
           LOG.info("Triggering shutdown hook....")

           system.stop(masterProxy)
           val cluster = Cluster(system)
           cluster.leave(cluster.selfAddress)
           cluster.down(cluster.selfAddress)
           try {
             Await.result(system.whenTerminated, Duration(3, TimeUnit.SECONDS))
           } catch {
             case ex: Exception => // Ignore
           }
           system.terminate()
           mainThread.join()
         }
       }
     })

     Await.result(system.whenTerminated, Duration.Inf)
   }
 }

 class MasterWatcher(role: String) extends Actor with ActorLogging {
   import context.dispatcher

   val cluster = Cluster(context.system)

   val config = context.system.settings.config
   val masters = config.getList("akka.cluster.seed-nodes")
   val quorum = masters.size() / 2 + 1

   val system = context.system

   // Sorts by age, oldest first
   val ageOrdering = Ordering.fromLessThan[Member] { (a, b) => a.isOlderThan(b) }
   var membersByAge: immutable.SortedSet[Member] = immutable.SortedSet.empty(ageOrdering)

   def receive: Receive = null

   // Subscribes to MemberEvent, re-subscribe when restart
   override def preStart(): Unit = {
     cluster.subscribe(self, classOf[MemberEvent])
     context.become(waitForInit)
   }
   override def postStop(): Unit = {
     cluster.unsubscribe(self)
   }

   def matchingRole(member: Member): Boolean = member.hasRole(role)

   def waitForInit: Receive = {
     case state: CurrentClusterState => {
       membersByAge = immutable.SortedSet.empty(ageOrdering) ++ state.members.filter(m =>
         m.status == MemberStatus.Up && matchingRole(m))

       if (membersByAge.size < quorum) {
         membersByAge.iterator.mkString(",")
         log.info(s"We cannot get a quorum, $quorum, " +
           s"shutting down...${membersByAge.iterator.mkString(",")}")
         context.become(waitForShutdown)
         self ! MasterWatcher.Shutdown
       } else {
         val master = context.actorOf(Props(classOf[MasterActor]), MASTER)
         notifyMasterMembersChange(master)
         context.become(waitForClusterEvent(master))
       }
     }
   }

   def waitForClusterEvent(master: ActorRef): Receive = {
     case MemberUp(m) if matchingRole(m) => {
       membersByAge += m
       notifyMasterMembersChange(master)
     }
     case mEvent: MemberEvent if (mEvent.isInstanceOf[MemberExited] ||
       mEvent.isInstanceOf[MemberRemoved]) && matchingRole(mEvent.member) => {
       log.info(s"member removed ${mEvent.member}")
       val m = mEvent.member
       membersByAge -= m
       if (membersByAge.size < quorum) {
         log.info(s"We cannot get a quorum, $quorum, " +
           s"shutting down...${membersByAge.iterator.mkString(",")}")
         context.become(waitForShutdown)
         self ! MasterWatcher.Shutdown
       } else {
         notifyMasterMembersChange(master)
       }
     }
   }

   private def notifyMasterMembersChange(master: ActorRef): Unit = {
     val masters = membersByAge.toList.map{ member =>
       MasterNode(member.address.host.getOrElse("Unknown-Host"),
         member.address.port.getOrElse(0))
     }
     master ! MasterListUpdated(masters)
   }

   def waitForShutdown: Receive = {
     case MasterWatcher.Shutdown => {
       cluster.unsubscribe(self)
       cluster.leave(cluster.selfAddress)
       context.stop(self)
       system.scheduler.scheduleOnce(Duration.Zero) {
         try {
           Await.result(system.whenTerminated, Duration(3, TimeUnit.SECONDS))
         } catch {
           case ex: Exception => // Ignore
         }
         system.terminate()
       }
     }
   }
 }

 object MasterWatcher {
   object Shutdown
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.gearpump.cluster.main

	import java.util.concurrent.TimeUnit

	import akka.actor._
	import akka.cluster.ClusterEvent._
	import akka.cluster.{MemberStatus, Member, Cluster}
	import akka.cluster.ddata.DistributedData
	import akka.cluster.singleton.{ClusterSingletonProxySettings, ClusterSingletonProxy, ClusterSingletonManagerSettings, ClusterSingletonManager}
	import com.typesafe.config.ConfigValueFactory
	import org.apache.gearpump.cluster.ClusterConfig
	import org.apache.gearpump.cluster.master.Master.MasterListUpdated
	import org.apache.gearpump.cluster.master.{Master => MasterActor, MasterNode}
	import org.apache.gearpump.util.Constants._
	import org.apache.gearpump.util.LogUtil.ProcessType
	import org.apache.gearpump.util.{AkkaApp, Constants, LogUtil}
	import org.slf4j.Logger

	import scala.collection.JavaConverters._
	import scala.collection.immutable
	import scala.concurrent.Await
	import scala.concurrent.duration._

	object Master extends AkkaApp with ArgumentsParser {

	private var LOG: Logger = LogUtil.getLogger(getClass)

	override def akkaConfig: Config = ClusterConfig.master()

	override val options: Array[(String, CLIOption[Any])] =
	Array("ip" -> CLIOption[String]("<master ip address>", required = true),
	"port" -> CLIOption("<master port>", required = true))

	override val description = "Start Master daemon"

	def main(akkaConf: Config, args: Array[String]): Unit = {

	this.LOG = {
	LogUtil.loadConfiguration(akkaConf, ProcessType.MASTER)
	LogUtil.getLogger(getClass)
	}

	val config = parse(args)
	master(config.getString("ip"), config.getInt("port"), akkaConf)
	}

	private def verifyMaster(master: String, port: Int, masters: Iterable[String]) = {
	masters.exists { hostPort =>
	hostPort == s"$master:$port"
	}
	}

	private def master(ip: String, port: Int, akkaConf: Config): Unit = {
	val masters = akkaConf.getStringList(Constants.GEARPUMP_CLUSTER_MASTERS).asScala

	if (!verifyMaster(ip, port, masters)) {
	LOG.error(s"The provided ip $ip and port $port doesn't conform with config at " +
	s"gearpump.cluster.masters: ${masters.mkString(", ")}")
	System.exit(-1)
	}

	val masterList = masters.map(master => s"akka.tcp://${MASTER}@$master").toList.asJava
	val quorum = masterList.size() / 2 + 1
	val masterConfig = akkaConf.
	withValue("akka.remote.netty.tcp.port", ConfigValueFactory.fromAnyRef(port)).
	withValue(NETTY_TCP_HOSTNAME, ConfigValueFactory.fromAnyRef(ip)).
	withValue("akka.cluster.seed-nodes", ConfigValueFactory.fromAnyRef(masterList)).
	withValue(s"akka.cluster.role.${MASTER}.min-nr-of-members",
	ConfigValueFactory.fromAnyRef(quorum))

	LOG.info(s"Starting Master Actor system $ip:$port, master list: ${masters.mkString(";")}")
	val system = ActorSystem(MASTER, masterConfig)

	val replicator = DistributedData(system).replicator
	LOG.info(s"Replicator path: ${replicator.path}")

	// Starts singleton manager
	val singletonManager = system.actorOf(ClusterSingletonManager.props(
	singletonProps = Props(classOf[MasterWatcher], MASTER),
	terminationMessage = PoisonPill,
	settings = ClusterSingletonManagerSettings(system).withSingletonName(MASTER_WATCHER)
	.withRole(MASTER)),
	name = SINGLETON_MANAGER)

	// Start master proxy
	val masterProxy = system.actorOf(ClusterSingletonProxy.props(
	singletonManagerPath = s"/user/${SINGLETON_MANAGER}",
	// The effective singleton is s"${MASTER_WATCHER}/$MASTER" instead of s"${MASTER_WATCHER}".
	// Master is created when there is a majority of machines started.
	settings = ClusterSingletonProxySettings(system)
	.withSingletonName(s"${MASTER_WATCHER}/$MASTER").withRole(MASTER)),
	name = MASTER
	)

	LOG.info(s"master proxy is started at ${masterProxy.path}")

	val mainThread = Thread.currentThread()
	Runtime.getRuntime().addShutdownHook(new Thread() {
	override def run(): Unit = {
	if (!system.whenTerminated.isCompleted) {
	LOG.info("Triggering shutdown hook....")

	system.stop(masterProxy)
	val cluster = Cluster(system)
	cluster.leave(cluster.selfAddress)
	cluster.down(cluster.selfAddress)
	try {
	Await.result(system.whenTerminated, Duration(3, TimeUnit.SECONDS))
	} catch {
	case ex: Exception => // Ignore
	}
	system.terminate()
	mainThread.join()
	}
	}
	})

	Await.result(system.whenTerminated, Duration.Inf)
	}
	}

	class MasterWatcher(role: String) extends Actor with ActorLogging {
	import context.dispatcher

	val cluster = Cluster(context.system)

	val config = context.system.settings.config
	val masters = config.getList("akka.cluster.seed-nodes")
	val quorum = masters.size() / 2 + 1

	val system = context.system

	// Sorts by age, oldest first
	val ageOrdering = Ordering.fromLessThan[Member] { (a, b) => a.isOlderThan(b) }
	var membersByAge: immutable.SortedSet[Member] = immutable.SortedSet.empty(ageOrdering)

	def receive: Receive = null

	// Subscribes to MemberEvent, re-subscribe when restart
	override def preStart(): Unit = {
	cluster.subscribe(self, classOf[MemberEvent])
	context.become(waitForInit)
	}
	override def postStop(): Unit = {
	cluster.unsubscribe(self)
	}

	def matchingRole(member: Member): Boolean = member.hasRole(role)

	def waitForInit: Receive = {
	case state: CurrentClusterState => {
	membersByAge = immutable.SortedSet.empty(ageOrdering) ++ state.members.filter(m =>
	m.status == MemberStatus.Up && matchingRole(m))

	if (membersByAge.size < quorum) {
	membersByAge.iterator.mkString(",")
	log.info(s"We cannot get a quorum, $quorum, " +
	s"shutting down...${membersByAge.iterator.mkString(",")}")
	context.become(waitForShutdown)
	self ! MasterWatcher.Shutdown
	} else {
	val master = context.actorOf(Props(classOf[MasterActor]), MASTER)
	notifyMasterMembersChange(master)
	context.become(waitForClusterEvent(master))
	}
	}
	}

	def waitForClusterEvent(master: ActorRef): Receive = {
	case MemberUp(m) if matchingRole(m) => {
	membersByAge += m
	notifyMasterMembersChange(master)
	}
	case mEvent: MemberEvent if (mEvent.isInstanceOf[MemberExited] \|\|
	mEvent.isInstanceOf[MemberRemoved]) && matchingRole(mEvent.member) => {
	log.info(s"member removed ${mEvent.member}")
	val m = mEvent.member
	membersByAge -= m
	if (membersByAge.size < quorum) {
	log.info(s"We cannot get a quorum, $quorum, " +
	s"shutting down...${membersByAge.iterator.mkString(",")}")
	context.become(waitForShutdown)
	self ! MasterWatcher.Shutdown
	} else {
	notifyMasterMembersChange(master)
	}
	}
	}

	private def notifyMasterMembersChange(master: ActorRef): Unit = {
	val masters = membersByAge.toList.map{ member =>
	MasterNode(member.address.host.getOrElse("Unknown-Host"),
	member.address.port.getOrElse(0))
	}
	master ! MasterListUpdated(masters)
	}

	def waitForShutdown: Receive = {
	case MasterWatcher.Shutdown => {
	cluster.unsubscribe(self)
	cluster.leave(cluster.selfAddress)
	context.stop(self)
	system.scheduler.scheduleOnce(Duration.Zero) {
	try {
	Await.result(system.whenTerminated, Duration(3, TimeUnit.SECONDS))
	} catch {
	case ex: Exception => // Ignore
	}
	system.terminate()
	}
	}
	}
	}

	object MasterWatcher {
	object Shutdown
	}