Merge pull request #348 from clowder-framework/ddey2/327

Added logic for primary Clowder instance which handles multiple extra…

Author: robkooper

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - swagger lint action
 - When downloading a file with a `'` in the name it would save the file as blob
 - Fix for a rare race condition with masonry where tiles could end up overlapping in space page.
+- Fixes bug where same extractor shows up multiple times and all Clowder instances index db on reindex [#327](https://github.com/clowder-framework/clowder/issues/327)
 
 ### Changed
 - Changed `Enabled By SuperAdmin` to read `Enabled by Server Admin` [#344](https://github.com/clowder-framework/clowder/issues/344)

app/services/mongodb/MongoDBQueueService.scala

@@ -93,7 +93,8 @@ trait MongoDBQueueService {
 
   // start pool to being processing queue actions
   def listen() = {
-    if (queueTimer == null) {
+    //only if this is the primary clowder instance
+    if (queueTimer == null && configuration.getBoolean("clowder.primary").getOrElse(true)) {
       // TODO: Need to make these in a separate pool
       queueTimer = Akka.system().scheduler.schedule(0 seconds, 5 millis) {
         getNextQueuedAction match {

app/services/rabbitmq/RabbitMQMessageService.scala

@@ -128,24 +128,27 @@ class RabbitMQMessageService extends MessageService {
         new MsgConsumer(channel.get, event_filter.get)
       )
 
-      // Start actor to listen to extractor heartbeats
-      Logger.info("Starting extractor heartbeat listener")
-      // create fanout exchange if it doesn't already exist
-      channel.get.exchangeDeclare("extractors", "fanout", true)
-      // anonymous queue
-      val heartbeatsQueue = channel.get.queueDeclare().getQueue
-      // bind queue to exchange
-      channel.get.queueBind(heartbeatsQueue, "extractors", "*")
-      extractorsHeartbeats = Some(Akka.system.actorOf(
-        Props(new ExtractorsHeartbeats(channel.get, heartbeatsQueue)), name = "ExtractorsHeartbeats"
-      ))
-      Logger.debug("Initializing a MsgConsumer for the ExtractorsHeartbeats")
-      channel.get.basicConsume(
-        heartbeatsQueue,
-        false, // do not auto ack
-        "ExtractorsHeartbeats", // tagging the consumer is important if you want to stop it later
-        new MsgConsumer(channel.get, extractorsHeartbeats.get)
-      )
+      //register new extractor only if this is the primary clowder instance
+      if (configuration.getBoolean("clowder.primary").getOrElse(true)) {
+        // Start actor to listen to extractor heartbeats
+        Logger.info("Starting extractor heartbeat listener")
+        // create fanout exchange if it doesn't already exist
+        channel.get.exchangeDeclare("extractors", "fanout", true)
+        // anonymous queue
+        val heartbeatsQueue = channel.get.queueDeclare().getQueue
+        // bind queue to exchange
+        channel.get.queueBind(heartbeatsQueue, "extractors", "*")
+        extractorsHeartbeats = Some(Akka.system.actorOf(
+          Props(new ExtractorsHeartbeats(channel.get, heartbeatsQueue)), name = "ExtractorsHeartbeats"
+        ))
+        Logger.debug("Initializing a MsgConsumer for the ExtractorsHeartbeats")
+        channel.get.basicConsume(
+          heartbeatsQueue,
+          false, // do not auto ack
+          "ExtractorsHeartbeats", // tagging the consumer is important if you want to stop it later
+          new MsgConsumer(channel.get, extractorsHeartbeats.get)
+        )
+      }
 
       // Setup Actor to submit new extractions to broker
       extractQueue = Some(Akka.system.actorOf(Props(new PublishDirectActor(channel = channel.get,

conf/application.conf

@@ -183,6 +183,13 @@ api.version="beta"
 # mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]
 mongodbURI = "mongodb://127.0.0.1:27017/clowder"
 
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# Clowder Primary Instance
+# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# This is required if there are multiple clowder instances. This config variable indicates
+# the primary clowder instance who takes care of special actions like registering new extractors,
+# indexing db on a reindex etc. The default value is true which means only one instance of clowder running.
+clowder.primary=true
 
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 # RabbitMQ

docker/custom.conf

@@ -37,6 +37,10 @@ smtp.host=${?SMTP_SERVER}
 service.byteStorage=services.filesystem.DiskByteStorageService
 service.byteStorage=${?CLOWDER_STORAGE}
 
+#primary Clowder instance
+clowder.primary=true
+clowder.primary=${?CLOWDER_PRIMARY}
+
 # location in case of services.filesystem.DiskByteStorageService
 clowder.diskStorage.path="/home/clowder/data"
 clowder.diskStorage.path=${?CLOWDER_DISKPATH}