Merge pull request #1 from logstash-plugins/master

9.1.1 changes

Author: ItamarBenjamin

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 9.1.1
+  - Docs: Set the default_codec doc attribute.
+
+## 9.1.0
+  - Set number_of_shards to 1 and document_type to '_doc' for es 7.x clusters #741 #747
+  - Fix usage of upsert and script when update action is interpolated #239
+  - Add metrics to track bulk level and document level responses #585
+
 ## 9.0.3
   - Ignore master-only nodes when using sniffing
 

docs/index.asciidoc

@@ -1,5 +1,6 @@
 :plugin: elasticsearch
 :type: output
+:default_codec: plain
 
 ///////////////////////////////////////////
 START - GENERATED VARIABLES, DO NOT EDIT!
@@ -105,8 +106,8 @@ not reevaluate its DNS value while the keepalive is in effect.
 
 This plugin supports request and response compression. Response compression is enabled by default and 
 for Elasticsearch versions 5.0 and later, the user doesn't have to set any configs in Elasticsearch for 
-it to send back compressed response. For versions before 5.0, `http.compression` must be set to `true` in 
-Elasticsearch[https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-http.html#modules-http] to take advantage of response compression when using this plugin
+it to send back compressed response. For versions before 5.0, `http.compression` must be set to `true` https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-http.html#modules-http[in 
+Elasticsearch] to take advantage of response compression when using this plugin
 
 For requests compression, regardless of the Elasticsearch version, users have to enable `http_compression` 
 setting in their Logstash config file.
@@ -476,6 +477,14 @@ This can be dynamic using the `%{foo}` syntax.
 
 Set script name for scripted update mode
 
+Example:
+[source,ruby]
+    output {
+      elasticsearch {
+        script => "ctx._source.message = params.event.get('message')"
+      }
+    }
+
 [id="plugins-{type}s-{plugin}-script_lang"]
 ===== `script_lang` 
 
@@ -682,3 +691,5 @@ See also https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-in
 
 [id="plugins-{type}s-{plugin}-common-options"]
 include::{include_path}/{type}.asciidoc[]
+
+:default_codec!:

lib/logstash/outputs/elasticsearch.rb

@@ -227,6 +227,7 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   config :http_compression, :validate => :boolean, :default => false
 
   def build_client
+    params["metric"] = metric
     @client ||= ::LogStash::Outputs::ElasticSearch::HttpClientBuilder.build(@logger, @hosts, params)
   end
 

lib/logstash/outputs/elasticsearch/common.rb

@@ -25,6 +25,8 @@ def register
 
       install_template
       check_action_validity
+      @bulk_request_metrics = metric.namespace(:bulk_requests)
+      @document_level_metrics = metric.namespace(:documents)
 
       @logger.info("New Elasticsearch output", :class => self.class.name, :hosts => @hosts.map(&:sanitized).map(&:to_s))
     end
@@ -156,8 +158,16 @@ def submit(actions)
 
       # If the response is nil that means we were in a retry loop
       # and aborted since we're shutting down
+      return if bulk_response.nil?
+
       # If it did return and there are no errors we're good as well
-      return if bulk_response.nil? || !bulk_response["errors"]
+      if bulk_response["errors"]
+        @bulk_request_metrics.increment(:with_errors)
+      else
+        @bulk_request_metrics.increment(:successes)
+        @document_level_metrics.increment(:successes, actions.size)
+        return
+      end
 
       actions_to_retry = []
       bulk_response["items"].each_with_index do |response,idx|
@@ -173,8 +183,10 @@ def submit(actions)
         # - For a mapping error, we send to dead letter queue for a human to intervene at a later point.
         # - For everything else there's mastercard. Yep, and we retry indefinitely. This should fix #572 and other transient network issues
         if DOC_SUCCESS_CODES.include?(status)
+          @document_level_metrics.increment(:successes)
           next
         elsif DOC_CONFLICT_CODE == status
+          @document_level_metrics.increment(:non_retryable_failures)
           @logger.warn "Failed action.", status: status, action: action, response: response if !failure_type_logging_whitelist.include?(failure["type"])
           next
         elsif DOC_DLQ_CODES.include?(status)
@@ -186,9 +198,11 @@ def submit(actions)
           else
             @logger.warn "Could not index event to Elasticsearch.", status: status, action: action, response: response
           end
+          @document_level_metrics.increment(:non_retryable_failures)
           next
         else
           # only log what the user whitelisted
+          @document_level_metrics.increment(:retryable_failures)
           @logger.info "retrying failed action with response code: #{status} (#{failure})" if !failure_type_logging_whitelist.include?(failure["type"])
           actions_to_retry << action
         end
@@ -198,16 +212,19 @@ def submit(actions)
     end
 
     # Determine the correct value for the 'type' field for the given event
-    DEFAULT_EVENT_TYPE="doc".freeze
+    DEFAULT_EVENT_TYPE_ES6="doc".freeze
+    DEFAULT_EVENT_TYPE_ES7="_doc".freeze
     def get_event_type(event)
       # Set the 'type' value for the index.
       type = if @document_type
                event.sprintf(@document_type)
              else
                if client.maximum_seen_major_version < 6
-                 event.get("type") || DEFAULT_EVENT_TYPE
+                 event.get("type") || DEFAULT_EVENT_TYPE_ES6
+               elsif client.maximum_seen_major_version == 6
+                 DEFAULT_EVENT_TYPE_ES6
                else
-                 DEFAULT_EVENT_TYPE
+                 DEFAULT_EVENT_TYPE_ES7
                end
              end
 
@@ -239,6 +256,7 @@ def safe_bulk(actions)
 
         # We retry until there are no errors! Errors should all go to the retry queue
         sleep_interval = sleep_for_interval(sleep_interval)
+        @bulk_request_metrics.increment(:failures)
         retry unless @stopping.true?
       rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::NoConnectionAvailableError => e
         @logger.error(
@@ -249,8 +267,10 @@ def safe_bulk(actions)
         )
         Stud.stoppable_sleep(sleep_interval) { @stopping.true? }
         sleep_interval = next_sleep_interval(sleep_interval)
+        @bulk_request_metrics.increment(:failures)
         retry unless @stopping.true?
       rescue ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
+        @bulk_request_metrics.increment(:failures)
         log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s}
         log_hash[:body] = e.response_body if @logger.debug? # Generally this is too verbose
         message = "Encountered a retryable error. Will Retry with exponential backoff "
@@ -280,6 +300,7 @@ def safe_bulk(actions)
         @logger.debug("Failed actions for last bad bulk request!", :actions => actions)
 
         sleep_interval = sleep_for_interval(sleep_interval)
+        @bulk_request_metrics.increment(:failures)
         retry unless @stopping.true?
       end
     end

lib/logstash/outputs/elasticsearch/elasticsearch-template-es7x.json

@@ -2,7 +2,8 @@
   "template" : "logstash-*",
   "version" : 60001,
   "settings" : {
-    "index.refresh_interval" : "5s"
+    "index.refresh_interval" : "5s",
+    "number_of_shards": 1
   },
   "mappings" : {
     "_doc" : {

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -50,6 +50,9 @@ class HttpClient
     #   through a special http path, such as using mod_rewrite.
     def initialize(options={})
       @logger = options[:logger]
+      @metric = options[:metric]
+      @bulk_request_metrics = @metric.namespace(:bulk_requests)
+      @bulk_response_metrics = @bulk_request_metrics.namespace(:responses)
 
       # Again, in case we use DEFAULT_OPTIONS in the future, uncomment this.
       # @options = DEFAULT_OPTIONS.merge(options)
@@ -142,9 +145,12 @@ def bulk_send(body_stream)
         body_stream.seek(0)
       end
 
+      @bulk_response_metrics.increment(response.code.to_s)
+
       if response.code != 200
+        url = ::LogStash::Util::SafeURI.new(response.final_url)
         raise ::LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError.new(
-          response.code, @bulk_path, body_stream.to_s, response.body
+          response.code, url, body_stream.to_s, response.body
         )
       end
 
@@ -282,7 +288,8 @@ def build_pool(options)
         :sniffing_path => options[:sniffing_path],
         :healthcheck_path => options[:healthcheck_path],
         :resurrect_delay => options[:resurrect_delay],
-        :url_normalizer => self.method(:host_to_url)
+        :url_normalizer => self.method(:host_to_url),
+        :metric => options[:metric]
       }
       pool_options[:scheme] = self.scheme if self.scheme
 

lib/logstash/outputs/elasticsearch/http_client/pool.rb

@@ -45,6 +45,7 @@ def message
     def initialize(logger, adapter, initial_urls=[], options={})
       @logger = logger
       @adapter = adapter
+      @metric = options[:metric]
       @initial_urls = initial_urls
 
       raise ArgumentError, "No URL Normalizer specified!" unless options[:url_normalizer]
@@ -161,8 +162,8 @@ def sniff!
     # Sniffs and returns the results. Does not update internal URLs!
     def check_sniff
       _, url_meta, resp = perform_request(:get, @sniffing_path)
+      @metric.increment(:sniff_requests)
       parsed = LogStash::Json.load(resp.body)
-      
       nodes = parsed['nodes']
       if !nodes || nodes.empty?
         @logger.warn("Sniff returned no nodes! Will not update hosts.")

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -14,6 +14,7 @@ def self.build(logger, hosts, params)
 
       common_options = {
         :client_settings => client_settings,
+        :metric => params["metric"],
         :resurrect_delay => params["resurrect_delay"]
       }
 

logstash-output-elasticsearch.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-elasticsearch'
-  s.version         = '9.0.3'
+  s.version         = '9.1.1'
   s.licenses        = ['apache-2.0']
   s.summary         = "Stores logs in Elasticsearch"
   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"

spec/integration/outputs/metrics_spec.rb

@@ -0,0 +1,70 @@
+require_relative "../../../spec/es_spec_helper"
+
+describe "metrics", :integration => true do
+  subject! do
+    require "logstash/outputs/elasticsearch"
+    settings = {
+      "manage_template" => false,
+      "hosts" => "#{get_host_port()}"
+    }
+    plugin = LogStash::Outputs::ElasticSearch.new(settings)
+  end
+
+  let(:metric) { subject.metric }
+  let(:bulk_request_metrics) { subject.instance_variable_get(:@bulk_request_metrics) }
+  let(:document_level_metrics) { subject.instance_variable_get(:@document_level_metrics) }
+
+  before :each do
+    require "elasticsearch"
+
+    # Clean ES of data before we start.
+    @es = get_client
+    @es.indices.delete_template(:name => "*")
+
+    # This can fail if there are no indexes, ignore failure.
+    @es.indices.delete(:index => "*") rescue nil
+    #@es.indices.refresh
+    subject.register
+  end
+
+  context "after a succesful bulk insert" do
+    let(:bulk) { [
+      LogStash::Event.new("message" => "sample message here"),
+      LogStash::Event.new("somemessage" => { "message" => "sample nested message here" }),
+      LogStash::Event.new("somevalue" => 100),
+      LogStash::Event.new("somevalue" => 10),
+      LogStash::Event.new("somevalue" => 1),
+      LogStash::Event.new("country" => "us"),
+      LogStash::Event.new("country" => "at"),
+      LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0 ] })
+    ]}
+
+    it "increases successful bulk request metric" do
+      expect(bulk_request_metrics).to receive(:increment).with(:successes).once
+      subject.multi_receive(bulk)
+    end
+
+    it "increases number of successful inserted documents" do
+      expect(document_level_metrics).to receive(:increment).with(:successes, bulk.size).once
+      subject.multi_receive(bulk)
+    end
+  end
+
+  context "after a bulk insert that generates errors" do
+    let(:bulk) { [
+      LogStash::Event.new("message" => "sample message here"),
+      LogStash::Event.new("message" => { "message" => "sample nested message here" }),
+    ]}
+    it "increases bulk request with error metric" do
+      expect(bulk_request_metrics).to receive(:increment).with(:with_errors).once
+      expect(bulk_request_metrics).to_not receive(:increment).with(:successes)
+      subject.multi_receive(bulk)
+    end
+
+    it "increases number of successful and non retryable documents" do
+      expect(document_level_metrics).to receive(:increment).with(:non_retryable_failures).once
+      expect(document_level_metrics).to receive(:increment).with(:successes).once
+      subject.multi_receive(bulk)
+    end
+  end
+end

spec/integration/outputs/sniffer_spec.rb

@@ -7,7 +7,13 @@
   let(:logger) { Cabin::Channel.get }
   let(:adapter) { LogStash::Outputs::ElasticSearch::HttpClient::ManticoreAdapter.new(logger) }
   let(:initial_urls) { [::LogStash::Util::SafeURI.new("http://#{get_host_port}")] }
-  let(:options) { {:resurrect_delay => 2, :url_normalizer => proc {|u| u}} } # Shorten the delay a bit to speed up tests
+  let(:options) do
+    {
+      :resurrect_delay => 2, # Shorten the delay a bit to speed up tests
+      :url_normalizer => proc {|u| u},
+      :metric => ::LogStash::Instrument::NullMetric.new(:dummy).namespace(:alsodummy)
+    }
+  end
 
   subject { LogStash::Outputs::ElasticSearch::HttpClient::Pool.new(logger, adapter, initial_urls, options) }
 

spec/unit/outputs/elasticsearch/http_client_spec.rb

@@ -7,7 +7,8 @@
   let(:base_options) do
     opts = {
       :hosts => [::LogStash::Util::SafeURI.new("127.0.0.1")],
-      :logger => Cabin::Channel.get
+      :logger => Cabin::Channel.get,
+      :metric => ::LogStash::Instrument::NamespacedNullMetric.new(:dummy_metric)
     }
 
     if !ssl.nil? # Shortcut to set this

spec/unit/outputs/elasticsearch_spec.rb

@@ -41,12 +41,15 @@
     let(:manticore_url) { manticore_urls.first }
 
     describe "getting a document type" do
-      it "should default to 'doc'" do
-        expect(subject.send(:get_event_type, LogStash::Event.new)).to eql("doc")
-      end
-
       context "if document_type isn't set" do
         let(:options) { super.merge("document_type" => nil)}
+        context "for 7.x elasticsearch clusters" do
+          let(:maximum_seen_major_version) { 7 }
+          it "should return '_doc'" do
+            expect(subject.send(:get_event_type, LogStash::Event.new("type" => "foo"))).to eql("_doc")
+          end
+        end
+
         context "for 6.x elasticsearch clusters" do
           let(:maximum_seen_major_version) { 6 }
           it "should return 'doc'" do