[EXT] Refactored and improved the "Reindex" extension

This change improves the patch cbfa9d4, which added a Reindex extension
to the extension library.

* The code of the `Reindex` class has been changed to separate the initialization
  and the actual executing of the reindex action

* The arguments to the class have been changed to be aligned with the "Reindex" API
  in the core Elasticsearch client (see 1cf5fb9)

* The unit tests have been restructured and amended

* Dedicated integration tests have been added to verify the behaviour against
  a real Elasticsearch cluster

Example:

    reindex = Elasticsearch::Extensions::Reindex.new \
                source: { index: 'test1', client: source_client },
                target: { index: 'test2', client: target_client },
                transform: lambda { |doc| doc['_source']['category'].upcase! },
                batch_size: 100,
                refresh: true

Related: cbfa9d4, 1cf5fb9

Closes #270

Author: karmi

elasticsearch-extensions/lib/elasticsearch/extensions.rb

@@ -1,6 +1,7 @@
+# encoding: utf-8
+
 require 'elasticsearch'
 require 'elasticsearch/extensions/version'
-require 'elasticsearch/extensions/reindex'
 
 module Elasticsearch
   module Extensions

elasticsearch-extensions/lib/elasticsearch/extensions/reindex.rb

@@ -1,65 +1,158 @@
+# encoding: utf-8
+
 module Elasticsearch
   module Extensions
-    # Reindex using the scroll api. This moves data (not mappings) from one index
-    # to another. The target index can be on a different cluster.
-    #
-    # This is useful when updating mappings on existing fields in an index (eg with
-    # new analyzers).
-    #
-    # @example Reindex all documents under a new index name
+
+    # This module allows copying documents from one index/cluster to another one
     #
-    # Elasticsearch::Extensions::Reindex.new client: client, src_index: 'foo', target_index: 'bar'
+    # When required together with the client, it will add the `reindex` method
     #
-    # @see https://www.elastic.co/guide/en/elasticsearch/guide/current/reindex.html
+    # @see Reindex::Reindex.initialize
+    # @see Reindex::Reindex#perform
     #
-    # @option arguments [Client] :client (*Required*)
-    # @option arguments [String] :src_index (*Required*)
-    # @option arguments [String] :target_index (*Required*)
-    # @option arguments [Client] :target_client
-    # @option arguments [Int] :chunk_size
-    # @option arguments [String] :period period to ask es to keep scroll buffer open '5m'
+    # @see http://www.rubydoc.info/gems/elasticsearch-api/Elasticsearch/API/Actions#reindex-instance_method
     #
-    class Reindex
-      def initialize(opts = {})
-        raise ArgumentError, "Required argument 'client' missing"  unless opts[:client]
-        raise ArgumentError, "Required argument 'src_index' missing"  unless opts[:src_index]
-        raise ArgumentError, "Required argument 'target_index' missing" unless opts[:target_index]
-
-        valid_params = [
-          :client,
-          :src_index,
-          :target_index,
-          :target_client,
-          :chunk_size,
-          :period
-        ]
-
-        default_params = {
-          chunk_size: 500,
-          period: '5m'
-        }
-
-        opts.each { |k, v| raise ArgumentError unless valid_params.include?(k) }
-        params = default_params.merge(opts)
-        client = params[:client]
-        target_client = params[:target_client] || client
-
-        r = client.search(index: params[:src_index],
-                          search_type: 'scan',
-                          scroll: params[:period],
-                          size: params[:chunk_size])
-
-        while r = client.scroll(scroll_id: r['_scroll_id'], scroll: params[:period]) do
-          docs = r['hits']['hits']
-          break if docs.empty?
-          body = docs.map do |doc|
-            doc['_index'] = params[:target_index]
-            doc['data'] = doc['_source']
-            doc.delete('_score')
-            doc.delete('_source')
-            { index: doc }
+    module Reindex
+
+      # Initialize a new instance of the Reindex class (shortcut)
+      #
+      # @see Reindex::Reindex.initialize
+      #
+      def new(arguments={})
+        Reindex.new(arguments)
+      end; extend self
+
+      module API
+        # Copy documents from one index into another and refresh the target index
+        #
+        # @example
+        #     client.reindex source: { index: 'test1' }, target: { index: 'test2' }, refresh: true
+        #
+        # The method allows all the options as {Reindex::Reindex.new}.
+        #
+        # This method will be mixed into the Elasticsearch client's API, if available.
+        #
+        def reindex(arguments={})
+          arguments[:source] ||= {}
+          arguments[:source][:client] = self
+          Reindex.new(arguments).perform
+        end
+      end
+
+      # Include the `reindex` method in the API and client, if available
+      Elasticsearch::API::Actions.__send__ :include, API if defined?(Elasticsearch::API::Actions)
+      Elasticsearch::Transport::Client.__send__ :include, API if defined?(Elasticsearch::Transport::Client) && defined?(Elasticsearch::API)
+
+      # Copy documents from one index into another
+      #
+      # @example Copy documents to another index
+      #
+      #   client  = Elasticsearch::Client.new
+      #   reindex = Elasticsearch::Extensions::Reindex.new \
+      #               source: { index: 'test1', client: client },
+      #               target: { index: 'test2' }
+      #
+      #   reindex.perform
+      #
+      # @example Copy documents to a different cluster
+      #
+      #     source_client  = Elasticsearch::Client.new url: 'http://localhost:9200'
+      #     target_client  = Elasticsearch::Client.new url: 'http://localhost:9250'
+      #
+      #     reindex = Elasticsearch::Extensions::Reindex.new \
+      #                 source: { index: 'test', client: source_client },
+      #                 target: { index: 'test', client: target_client }
+      #     reindex.perform
+      #
+      # @example Transform the documents during re-indexing
+      #
+      #     reindex = Elasticsearch::Extensions::Reindex.new \
+      #                 source: { index: 'test1', client: client },
+      #                 target: { index: 'test2' },
+      #                 transform: lambda { |doc| doc['_source']['category'].upcase! }
+      #
+      # The reindexing process works by "scrolling" an index and sending
+      # batches via the "Bulk" API to the target index/cluster
+      #
+      # @option arguments [String] :source The source index/cluster definition (*Required*)
+      # @option arguments [String] :target The target index/cluster definition (*Required*)
+      # @option arguments [Integer] :batch_size The size of the batch for scroll operation (Default: 1000)
+      # @option arguments [String] :scroll The timeout for the scroll operation (Default: 5min)
+      # @option arguments [Boolean] :refresh Whether to refresh the target index after
+      #                                      the operation is completed (Default: false)
+      # @option arguments [Proc] :transform A block which will be executed for each document
+      #
+      # Be aware, that if you want to change the target index settings and/or mappings,
+      # you have to do so in advance by using the "Indices Create" API.
+      #
+      # Note, that there is a native "Reindex" API in Elasticsearch 2.3.x and higer versions,
+      # which will be more performant than the Ruby version.
+      #
+      # @see http://www.rubydoc.info/gems/elasticsearch-api/Elasticsearch/API/Actions#reindex-instance_method
+      #
+      class Reindex
+        attr_reader :arguments
+
+        def initialize(arguments={})
+          [
+            [:source, :index],
+            [:source, :client],
+            [:target, :index]
+          ].each do |required_option|
+            value = required_option.reduce(arguments) { |sum, o| sum = sum[o] ? sum[o] : {}  }
+
+            raise ArgumentError,
+                  "Required argument '#{Hash[*required_option]}' missing" if \
+                  value.respond_to?(:empty?) ? value.empty? : value.nil?
+          end
+
+          @arguments = {
+            batch_size: 1000,
+            scroll: '5m',
+            refresh: false
+          }.merge(arguments)
+
+          arguments[:target][:client] ||= arguments[:source][:client]
+        end
+
+        # Performs the operation
+        #
+        # @return [Hash] A Hash with the information about the operation outcome
+        #
+        def perform
+          output = { errors: 0 }
+
+          response = arguments[:source][:client].search(
+            index: arguments[:source][:index],
+            scroll: arguments[:scroll],
+            size: arguments[:batch_size],
+            search_type: 'scan',
+            fields: ['_source', '_parent', '_routing', '_timestamp']
+          )
+
+          while response = arguments[:source][:client].scroll(scroll_id: response['_scroll_id'], scroll: arguments[:scroll]) do
+            documents = response['hits']['hits']
+            break if documents.empty?
+
+            bulk = documents.map do |doc|
+              doc['_index'] = arguments[:target][:index]
+
+              arguments[:target][:transform].call(doc) if arguments[:target][:transform]
+
+              doc['data'] = doc['_source']
+              doc.delete('_score')
+              doc.delete('_source')
+
+              { index: doc }
+            end
+
+            bulk_response = arguments[:target][:client].bulk body: bulk
+            output[:errors] += bulk_response['items'].select { |k, v| k.values.first['error'] }.size
           end
-          target_client.bulk body: body
+
+          arguments[:target][:client].indices.refresh index: arguments[:target][:index] if arguments[:refresh]
+
+          output
         end
       end
     end

elasticsearch-extensions/test/reindex/integration/reindex_test.rb

@@ -0,0 +1,80 @@
+require 'test_helper'
+require 'elasticsearch/extensions/reindex'
+
+class Elasticsearch::Extensions::ReindexIntegrationTest < Elasticsearch::Test::IntegrationTestCase
+  context "The Reindex extension" do
+    setup do
+      @port = (ENV['TEST_CLUSTER_PORT'] || 9250).to_i
+
+      @logger =  ::Logger.new(STDERR)
+      @logger.formatter = proc do |severity, datetime, progname, msg|
+        color = case severity
+          when /INFO/ then :green
+          when /ERROR|WARN|FATAL/ then :red
+          when /DEBUG/ then :cyan
+          else :white
+        end
+        ANSI.ansi(severity[0] + ' ', color, :faint) + ANSI.ansi(msg, :white, :faint) + "\n"
+      end
+
+      @client = Elasticsearch::Client.new host: "localhost:#{@port}", logger: @logger
+      @client.indices.delete index: '_all'
+
+      @client.index index: 'test1', type: 'd', id: 1, body: { title: 'TEST 1', category: 'one' }
+      @client.index index: 'test1', type: 'd', id: 2, body: { title: 'TEST 2', category: 'two' }
+      @client.index index: 'test1', type: 'd', id: 3, body: { title: 'TEST 3', category: 'three' }
+      @client.indices.refresh index: 'test1'
+
+      @client.cluster.health wait_for_status: 'yellow'
+    end
+
+    should "copy documents from one index to another" do
+      reindex = Elasticsearch::Extensions::Reindex.new \
+                  source: { index: 'test1', client: @client },
+                  target: { index: 'test2' },
+                  refresh: true
+
+      result = reindex.perform
+
+      assert_equal 0, result[:errors]
+      assert_equal 3, @client.search(index: 'test2')['hits']['total']
+    end
+
+    should "transform documents with a lambda" do
+      reindex = Elasticsearch::Extensions::Reindex.new \
+                  source: { index: 'test1', client: @client },
+                  target: { index: 'test2', transform: lambda { |d| d['_source']['category'].upcase! } },
+                  refresh: true
+
+      result = reindex.perform
+
+      assert_equal 0, result[:errors]
+      assert_equal 3, @client.search(index: 'test2')['hits']['total']
+      assert_equal 'ONE', @client.get(index: 'test2', type: 'd', id: 1)['_source']['category']
+    end
+
+    should "return the number of errors" do
+      @client.indices.create index: 'test3', body: { mappings: { d: { properties: { category: { type: 'integer' } }}}}
+      @client.cluster.health wait_for_status: 'yellow'
+
+      reindex = Elasticsearch::Extensions::Reindex.new \
+                  source: { index: 'test1', client: @client },
+                  target: { index: 'test3', transform: lambda { |d| d['_source']['category'].upcase!; d } },
+                  refresh: true
+
+      result = reindex.perform
+
+      assert_equal 3, result[:errors]
+      assert_equal 0, @client.search(index: 'test3')['hits']['total']
+    end
+
+    should "reindex via the API integration" do
+      @client.reindex source: { index: 'test1' }, target: { index: 'test4' }
+
+      @client.indices.refresh index: 'test4'
+
+      assert_equal 3, @client.search(index: 'test4')['hits']['total']
+    end
+  end
+
+end