Remove scan option

Use `{ "sort": [ "_doc" ] }` instead.

Clean up the previous stuff.

fixes logstash-plugins#50

Fixes logstash-plugins#51

Author: untergeek

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 4.0.0
+  - Remove `scan` from list of options as this is no longer allowed in
+    Elasticsearch 5.0.
+  - Change default query to sort by \_doc, as this replicates the `scan`
+    behavior
+  - Improve documentation to show sort by \_doc, and how to add it to custom
+    queries.
+    
 ## 3.0.2
   - Relax constraint on logstash-core-plugin-api to >= 1.60 <= 2.99
 

lib/logstash/inputs/elasticsearch.rb

@@ -12,7 +12,7 @@
 #       # Read all documents from Elasticsearch matching the given query
 #       elasticsearch {
 #         hosts => "localhost"
-#         query => '{ "query": { "match": { "statuscode": 200 } } }'
+#         query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }'
 #       }
 #     }
 #
@@ -23,7 +23,8 @@
 #         "match": {
 #           "statuscode": 200
 #         }
-#       }
+#       },
+#       "sort": [ "_doc" ]
 #     }'
 #
 class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
@@ -40,20 +41,16 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   config :index, :validate => :string, :default => "logstash-*"
 
   # The query to be executed. Read the Elasticsearch query DSL documentation
-  # for more info 
+  # for more info
   # https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html
-  config :query, :validate => :string, :default => '{"query": { "match_all": {} } }'
-
-  # Enable the Elasticsearch "scan" search type.  This will disable
-  # sorting but increase speed and performance.
-  config :scan, :validate => :boolean, :default => true
+  config :query, :validate => :string, :default => '{ "sort": [ "_doc" ] }'
 
   # This allows you to set the maximum number of hits returned per scroll.
   config :size, :validate => :number, :default => 1000
 
   # This parameter controls the keepalive time in seconds of the scrolling
   # request and initiates the scrolling process. The timeout applies per
-  # round trip (i.e. between the previous scan scroll request, to the next).
+  # round trip (i.e. between the previous scroll request, to the next).
   config :scroll, :validate => :string, :default => "1m"
 
   # If set, include Elasticsearch document information such as index, type, and
@@ -117,8 +114,6 @@ def register
       :size => @size
     }
 
-    @options[:search_type] = 'scan' if @scan
-
     transport_options = {}
 
     if @user && @password
@@ -146,14 +141,8 @@ def run(output_queue)
     # get first wave of data
     r = @client.search(@options)
 
-    # since 'scan' doesn't return data on the search call, do an extra scroll
-    if @scan
-      r = process_next_scroll(output_queue, r['_scroll_id'])
-      has_hits = r['has_hits']
-    else # not a scan, process the response
-      r['hits']['hits'].each { |hit| push_hit(hit, output_queue) }
-      has_hits = r['hits']['hits'].any?
-    end
+    r['hits']['hits'].each { |hit| push_hit(hit, output_queue) }
+    has_hits = r['hits']['hits'].any?
 
     while has_hits && !stop?
       r = process_next_scroll(output_queue, r['_scroll_id'])

logstash-input-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-elasticsearch'
-  s.version         = '3.0.2'
+  s.version         = '4.0.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Read from an Elasticsearch cluster, based on search query results"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -28,4 +28,3 @@ Gem::Specification.new do |s|
 
   s.add_development_dependency 'logstash-devutils'
 end
-

spec/inputs/elasticsearch_spec.rb

@@ -28,7 +28,6 @@
       input {
         elasticsearch {
           hosts => ["localhost"]
-          scan => false
           query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
         }
       }
@@ -74,63 +73,6 @@
     insist { event.get("message") } == [ "ohayo" ]
   end
 
-  it "should retrieve json event from elasticseach with scan" do
-    config = %q[
-      input {
-        elasticsearch {
-          hosts => ["localhost"]
-          scan => true
-          query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
-        }
-      }
-    ]
-
-    scan_response = {
-      "_scroll_id" => "DcrY3G1xff6SB",
-    }
-
-    scroll_responses = [
-      {
-        "_scroll_id" => "cXVlcnlUaGVuRmV0Y2g",
-        "took" => 27,
-        "timed_out" => false,
-        "_shards" => {
-          "total" => 169,
-          "successful" => 169,
-          "failed" => 0
-        },
-        "hits" => {
-          "total" => 1,
-          "max_score" => 1.0,
-          "hits" => [ {
-            "_index" => "logstash-2014.10.12",
-            "_type" => "logs",
-            "_id" => "C5b2xLQwTZa76jBmHIbwHQ",
-            "_score" => 1.0,
-            "_source" => { "message" => ["ohayo"] }
-          } ]
-        }
-      },
-      {
-        "_scroll_id" => "r453Wc1jh0caLJhSDg",
-        "hits" => { "hits" => [] }
-      }
-    ]
-
-    client = Elasticsearch::Client.new
-    expect(Elasticsearch::Client).to receive(:new).with(any_args).and_return(client)
-    expect(client).to receive(:search).with(any_args).and_return(scan_response)
-    expect(client).to receive(:scroll).with({ :body => "DcrY3G1xff6SB", :scroll => "1m" }).and_return(scroll_responses.first)
-    expect(client).to receive(:scroll).with({ :body=> "cXVlcnlUaGVuRmV0Y2g", :scroll => "1m" }).and_return(scroll_responses.last)
-
-    event = input(config) do |pipeline, queue|
-      queue.pop
-    end
-
-    insist { event }.is_a?(LogStash::Event)
-    insist { event.get("message") } == [ "ohayo" ]
-  end
-
   context "with Elasticsearch document information" do
     let!(:response) do
       {
@@ -150,7 +92,7 @@
             "_type" => "logs",
             "_id" => "C5b2xLQwTZa76jBmHIbwHQ",
             "_score" => 1.0,
-            "_source" => { 
+            "_source" => {
               "message" => ["ohayo"],
               "metadata_with_hash" => { "awesome" => "logstash" },
               "metadata_with_string" => "a string"
@@ -181,7 +123,6 @@
             input {
               elasticsearch {
                 hosts => ["localhost"]
-                scan => false
                 query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
                 docinfo => true
               }
@@ -196,7 +137,6 @@
             input {
               elasticsearch {
                 hosts => ["localhost"]
-                scan => false
                 query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
                 docinfo => true
                 docinfo_target => '#{metadata_field}'
@@ -213,15 +153,14 @@
         expect(event.get("[#{metadata_field}][_id]")).to eq('C5b2xLQwTZa76jBmHIbwHQ')
         expect(event.get("[#{metadata_field}][awesome]")).to eq("logstash")
       end
-      
+
       it 'thows an exception if the `docinfo_target` exist but is not of type hash' do
         metadata_field = 'metadata_with_string'
 
         config_metadata_with_string = %Q[
             input {
               elasticsearch {
                 hosts => ["localhost"]
-                scan => false
                 query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
                 docinfo => true
                 docinfo_target => '#{metadata_field}'
@@ -253,7 +192,6 @@
             input {
               elasticsearch {
                 hosts => ["localhost"]
-                scan => false
                 query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
                 docinfo => true
                 docinfo_target => 'meta'
@@ -275,7 +213,6 @@
             input {
               elasticsearch {
                 hosts => ["localhost"]
-                scan => false
                 query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
                 docinfo => true
                 docinfo_fields => #{fields}
@@ -299,7 +236,6 @@
           input {
             elasticsearch {
               hosts => ["localhost"]
-              scan => false
               query => '{ "query": { "match": { "city_name": "Okinawa" } }, "fields": ["message"] }'
             }
           }