Added logic for primary Clowder instance which handles multiple extractors problem and multiple indexing issue

Author: ddey2

CHANGELOG.md

@@ -5,6 +5,9 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## Unreleased
+- 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)
+
 ## 1.20.1 - 2022-04-04
 
 ### Fixed

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

@@ -190,6 +190,7 @@ mongodbURI = "mongodb://127.0.0.1:27017/clowder"
 # All requests to extractors and any other external process is send
 # using rabbitmq, this will setup the connection information.
 clowder.rabbitmq.uri="amqp://guest:guest@localhost:5672/%2f"
+clowder.primary=true
 #clowder.rabbitmq.managmentPort=15672
 #clowder.rabbitmq.exchange=clowder
 # Following variable will use the url when sending messages over rabbitmq. This can be used in a docker

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}