Merge pull request #34 from daniellienert/feature/failure-trehsold-for-switching-index

FEATURE: Introduce threshold for failed jobs for index switching

Author: daniellienert

Classes/UpdateAliasJob.php

@@ -38,6 +38,12 @@ class UpdateAliasJob implements JobInterface
      */
     protected $indexPostfix;
 
+    /**
+     * @Flow\InjectConfiguration(path="acceptedFailedJobs")
+     * @var int
+     */
+    protected $acceptedFailedJobs = -1;
+
     /**
      * @param string $indexPostfix
      */
@@ -48,9 +54,6 @@ public function __construct($indexPostfix)
     }
 
     /**
-     * Execute the job
-     * A job should finish itself after successful execution using the queue methods.
-     *
      * @param QueueInterface $queue
      * @param Message $message The original message
      * @return boolean TRUE if the job was executed successfully and the message should be finished
@@ -60,9 +63,13 @@ public function __construct($indexPostfix)
      */
     public function execute(QueueInterface $queue, Message $message): bool
     {
-        $this->nodeIndexer->setIndexNamePostfix($this->indexPostfix);
-        $this->nodeIndexer->updateIndexAlias();
-        $this->log(sprintf('action=indexing step=index-switched alias=%s', $this->indexPostfix), LOG_NOTICE);
+        if ($this->shouldIndexBeSwitched($queue)) {
+            $this->nodeIndexer->setIndexNamePostfix($this->indexPostfix);
+            $this->nodeIndexer->updateIndexAlias();
+            $this->log(sprintf('action=indexing step=index-switched alias=%s message="Index was switched successfully"', $this->indexPostfix), LOG_NOTICE);
+        } else {
+            $this->log(sprintf('action=indexing step=index-switched alias=%s message="Index was not switched due to %s failed batches in the current queue"', $this->indexPostfix, $queue->countFailed()), LOG_ERR);
+        }
 
         return true;
     }
@@ -86,4 +93,21 @@ public function getLabel(): string
     {
         return sprintf('ElasticSearch Indexing Job (%s)', $this->getIdentifier());
     }
+
+    /**
+     * @param QueueInterface $queue
+     * @return bool
+     */
+    protected function shouldIndexBeSwitched(QueueInterface $queue): bool
+    {
+        if ($this->acceptedFailedJobs === -1) {
+            return true;
+        }
+
+        if ($queue->countFailed() <= $this->acceptedFailedJobs) {
+            return true;
+        }
+
+        return false;
+    }
 }

Configuration/Settings.yaml

@@ -4,6 +4,11 @@ Flowpack:
       enableLiveAsyncIndexing: true
       # Change size of single batch jobs via this setting
       batchSize: 500
+
+      # Number of failed jobs accepted to switch the alias to the new index
+      # -1 means, accept any number of failures
+      acceptedFailedJobs: 0
+
   JobQueue:
     Common:
       presets:

README.md

@@ -80,8 +80,7 @@ You can use this CLI command to process indexing job:
 
 # Supervisord configuration
 
-You can use tools like ```supervisord``` to manage long runing process. Bellow you can
-found a basic configuration:
+You can use tools like ```supervisord``` to manage long running processes. Bellow you can find a basic configuration:
 
     [supervisord]