Merge pull request #53 from digiaonline/improve-reindex-settings

Improve re-indexing speed
digiaonline · Nov 7, 2017 · c796db5 · c796db5
2 parents bdcd741 + e0fe352
commit c796db5
Show file tree

Hide file tree

Showing 17 changed files with 376 additions and 163 deletions.
diff --git a/src/Console/ApplyMigrationCommand.php b/src/Console/ApplyMigrationCommand.php
@@ -8,6 +8,8 @@
 use Nord\Lumen\Elasticsearch\Pipelines\Stages\CheckIndexExistsStage;
 use Nord\Lumen\Elasticsearch\Pipelines\Stages\CreateIndexStage;
 use Nord\Lumen\Elasticsearch\Pipelines\Stages\DetermineTargetVersionStage;
+use Nord\Lumen\Elasticsearch\Pipelines\Stages\ReIndexStage;
+use Nord\Lumen\Elasticsearch\Pipelines\Stages\StoreIndexSettingsStage;
 use Nord\Lumen\Elasticsearch\Pipelines\Stages\UpdateIndexAliasStage;
 
 /**
@@ -48,6 +50,8 @@ public function handle()
             new DetermineTargetVersionStage(),
             new CheckIndexExistsStage($this->elasticsearchService),
             new CreateIndexStage($this->elasticsearchService),
+            new StoreIndexSettingsStage($this->elasticsearchService),
+            new ReIndexStage($this->elasticsearchService),
             new UpdateIndexAliasStage($this->elasticsearchService),
         ]);
 

diff --git a/src/Pipelines/Payloads/ApplyMigrationPayload.php b/src/Pipelines/Payloads/ApplyMigrationPayload.php
@@ -19,6 +19,11 @@ class ApplyMigrationPayload extends MigrationPayload
      */
     private $batchSize;
 
+    /**
+     * @var int
+     */
+    private $numberOfReplicas;
+
     /**
      * ApplyMigrationPayload constructor.
      *
@@ -71,4 +76,20 @@ public function getBatchSize()
     {
         return $this->batchSize;
     }
+
+    /**
+     * @return int
+     */
+    public function getNumberOfReplicas()
+    {
+        return $this->numberOfReplicas;
+    }
+
+    /**
+     * @param int $numberOfReplicas
+     */
+    public function setNumberOfReplicas($numberOfReplicas)
+    {
+        $this->numberOfReplicas = $numberOfReplicas;
+    }
 }
diff --git a/src/Pipelines/Stages/CreateIndexStage.php b/src/Pipelines/Stages/CreateIndexStage.php
@@ -37,21 +37,6 @@ public function __invoke($payload)
         // Create the new index
         $this->elasticsearchService->indices()->create($payload->getTargetConfiguration());
 
-        // Reindex data from the old index to the new, but only if the old index exists (not true on brand new setups)
-        if ($this->elasticsearchService->indices()->exists(['index' => $payload->getIndexName()])) {
-            $this->elasticsearchService->reindex([
-                'body' => [
-                    'source' => [
-                        'index' => $payload->getIndexName(),
-                        'size'  => $payload->getBatchSize(),
-                    ],
-                    'dest'   => [
-                        'index' => $payload->getTargetVersionName(),
-                    ],
-                ],
-            ]);
-        }
-
         return $payload;
     }
 }
diff --git a/src/Pipelines/Stages/ReIndexStage.php b/src/Pipelines/Stages/ReIndexStage.php
@@ -0,0 +1,61 @@
+<?php
+
+namespace Nord\Lumen\Elasticsearch\Pipelines\Stages;
+
+use League\Pipeline\StageInterface;
+use Nord\Lumen\Elasticsearch\Contracts\ElasticsearchServiceContract;
+
+/**
+ * Class ReIndexStage
+ * @package Nord\Lumen\Elasticsearch\Pipelines\Stages
+ */
+class ReIndexStage implements StageInterface
+{
+
+    /**
+     * @var ElasticsearchServiceContract
+     */
+    private $elasticsearchService;
+
+    /**
+     * CheckIndexExistsStage constructor.
+     *
+     * @param ElasticsearchServiceContract $elasticsearchService
+     */
+    public function __construct(ElasticsearchServiceContract $elasticsearchService)
+    {
+        $this->elasticsearchService = $elasticsearchService;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function __invoke($payload)
+    {
+        // Reindex data from the old index to the new, but only if the old index exists (not true on brand new setups)
+        if ($this->elasticsearchService->indices()->exists(['index' => $payload->getIndexName()])) {
+            // Temporarily change some index settings to speed up the process
+            $this->elasticsearchService->indices()->putSettings([
+                'index' => $payload->getTargetVersionName(),
+                'body'  => [
+                    'refresh_interval'   => -1,
+                    'number_of_replicas' => 0,
+                ],
+            ]);
+
+            $this->elasticsearchService->reindex([
+                'body' => [
+                    'source' => [
+                        'index' => $payload->getIndexName(),
+                        'size'  => $payload->getBatchSize(),
+                    ],
+                    'dest'   => [
+                        'index' => $payload->getTargetVersionName(),
+                    ],
+                ],
+            ]);
+        }
+
+        return $payload;
+    }
+}
diff --git a/src/Pipelines/Stages/StoreIndexSettingsStage.php b/src/Pipelines/Stages/StoreIndexSettingsStage.php
@@ -0,0 +1,48 @@
+<?php
+
+namespace Nord\Lumen\Elasticsearch\Pipelines\Stages;
+
+use League\Pipeline\StageInterface;
+use Nord\Lumen\Elasticsearch\Contracts\ElasticsearchServiceContract;
+use Nord\Lumen\Elasticsearch\Pipelines\Payloads\ApplyMigrationPayload;
+
+/**
+ * Class StoreIndexSettingsStage
+ * @package Nord\Lumen\Elasticsearch\Pipelines\Stages
+ */
+class StoreIndexSettingsStage implements StageInterface
+{
+
+    /**
+     * @var ElasticsearchServiceContract
+     */
+    private $elasticsearchService;
+
+    /**
+     * CheckIndexExistsStage constructor.
+     *
+     * @param ElasticsearchServiceContract $elasticsearchService
+     */
+    public function __construct(ElasticsearchServiceContract $elasticsearchService)
+    {
+        $this->elasticsearchService = $elasticsearchService;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function __invoke($payload)
+    {
+        /** @var ApplyMigrationPayload $payload */
+
+        // Store the current number_of_replicas setting value in the payload
+        $settings = $this->elasticsearchService->indices()->getSettings([
+            'index' => $payload->getTargetVersionName(),
+        ]);
+
+        $indexSettings = $settings[$payload->getTargetVersionName()]['settings']['index'];
+        $payload->setNumberOfReplicas($indexSettings['number_of_replicas']);
+
+        return $payload;
+    }
+}
diff --git a/src/Pipelines/Stages/UpdateIndexAliasStage.php b/src/Pipelines/Stages/UpdateIndexAliasStage.php
@@ -54,6 +54,15 @@ public function __invoke($payload)
             }
         }
 
+        // Revert temporary index settings
+        $indices->putSettings([
+            'index' => $payload->getTargetVersionName(),
+            'body'  => [
+                'refresh_interval'   => '1s',
+                'number_of_replicas' => $payload->getNumberOfReplicas(),
+            ],
+        ]);
+
         // Update the alias definition
         $indices->updateAliases([
             'body' => [

diff --git a/tests/Pipelines/Payloads/ApplyMigrationPayloadTest.php b/tests/Pipelines/Payloads/ApplyMigrationPayloadTest.php
@@ -26,5 +26,9 @@ public function testGettersSetters()
         $this->assertTrue(is_array($payload->getTargetConfiguration()));
         $this->assertEquals('content_7', $payload->getTargetVersionName());
         $this->assertEquals(100, $payload->getBatchSize());
+
+        // Test number of replicas
+        $payload->setNumberOfReplicas(5);
+        $this->assertEquals(5, $payload->getNumberOfReplicas());
     }
 }
diff --git a/tests/Pipelines/Stages/AbstractStageTestCase.php b/tests/Pipelines/Stages/AbstractStageTestCase.php
@@ -0,0 +1,45 @@
+<?php
+
+namespace Nord\Lumen\Elasticsearch\Tests\Pipelines\Stages;
+
+use Elasticsearch\Namespaces\IndicesNamespace;
+use Nord\Lumen\Elasticsearch\Contracts\ElasticsearchServiceContract;
+use Nord\Lumen\Elasticsearch\Tests\TestCase;
+
+/**
+ * Class AbstractStageTestCase
+ * @package Nord\Lumen\Elasticsearch\Tests\Pipelines\Stages;
+ */
+abstract class AbstractStageTestCase extends TestCase
+{
+
+    /**
+     * @param array $methods
+     *
+     * @return IndicesNamespace|\PHPUnit_Framework_MockObject_MockObject
+     */
+    protected function getMockedIndices($methods = [])
+    {
+        return $this->getMockBuilder(IndicesNamespace::class)
+                    ->disableOriginalConstructor()
+                    ->setMethods($methods)
+                    ->getMock();
+    }
+
+    /**
+     * @param IndicesNamespace|\PHPUnit_Framework_MockObject_MockObject $mockedIndices
+     *
+     * @return ElasticsearchServiceContract|\PHPUnit_Framework_MockObject_MockObject
+     */
+    protected function getMockedSearchService($mockedIndices)
+    {
+        $searchService = $this->getMockBuilder(ElasticsearchServiceContract::class)
+                              ->setMethods(['indices'])
+                              ->getMockForAbstractClass();
+
+        $searchService->method('indices')
+                      ->willReturn($mockedIndices);
+
+        return $searchService;
+    }
+}
diff --git a/tests/Pipelines/Stages/CheckIndexExistsStageTest.php b/tests/Pipelines/Stages/CheckIndexExistsStageTest.php
@@ -1,44 +1,30 @@
 <?php
 
-namespace Nord\Lumen\Elasticsearch\Tests\Unit\Search\Pipelines\Stages;
+namespace Nord\Lumen\Elasticsearch\Tests\Pipelines\Stages;
 
-use Elasticsearch\Namespaces\IndicesNamespace;
-use Nord\Lumen\Elasticsearch\Contracts\ElasticsearchServiceContract;
 use Nord\Lumen\Elasticsearch\Pipelines\Payloads\ApplyMigrationPayload;
 use Nord\Lumen\Elasticsearch\Pipelines\Stages\CheckIndexExistsStage;
-use Nord\Lumen\Elasticsearch\Tests\TestCase;
 
 /**
  * Class CheckIndexExistsStageTest
- * @package Nord\Lumen\Elasticsearch\Tests\Unit\Search\Pipelines\Stages
+ * @package Nord\Lumen\Elasticsearch\Tests\Pipelines\Stages
  */
-class CheckIndexExistsStageTest extends TestCase
+class CheckIndexExistsStageTest extends AbstractStageTestCase
 {
 
     /**
      * @expectedException \Nord\Lumen\Elasticsearch\Exceptions\IndexExistsException
      */
     public function testStage()
     {
-        /** @var IndicesNamespace|\PHPUnit_Framework_MockObject_MockObject $indices */
-        $indices = $this->getMockBuilder(IndicesNamespace::class)
-                        ->disableOriginalConstructor()
-                        ->setMethods(['exists'])
-                        ->getMock();
+        $indices = $this->getMockedIndices(['exists']);
 
         $indices->expects($this->once())
                 ->method('exists')
                 ->with(['index' => 'content_1'])
                 ->willReturn(true);
 
-        /** @var ElasticsearchServiceContract|\PHPUnit_Framework_MockObject_MockObject $service */
-        $service = $this->getMockBuilder(ElasticsearchServiceContract::class)
-                        ->setMethods(['indices'])
-                        ->getMockForAbstractClass();
-
-        $service->expects($this->once())
-                ->method('indices')
-                ->willReturn($indices);
+        $service = $this->getMockedSearchService($indices);
 
         $payload = new ApplyMigrationPayload($this->getResourcesBasePath() . '/content.php', 100);
         $payload->setTargetVersionFile('1.php');