RATS: Dead Letter Queue integration tests (WIP)

Simple test for dead letter queue integration tests:
  Attempt to write invalid entries to elastic search, fail and
  remove invalid field. Verify that mutated entry exists in es

Not for committing - has different jvm.options to improve stability
  to ensure that the tests pass in CI.

Fixes #7882

Fixes #8026

config/jvm.options

@@ -52,9 +52,9 @@
 #-Djna.nosys=true
 
 # Turn on JRuby invokedynamic
--Djruby.compile.invokedynamic=true
+#-Djruby.compile.invokedynamic=true
 # Force Compilation
--Djruby.jit.threshold=0
+#-Djruby.jit.threshold=0
 
 ## heap dumps
 

qa/integration/fixtures/dlq_spec.yml

@@ -0,0 +1,38 @@
+---
+services:
+  - logstash
+  - elasticsearch
+config:
+  input {
+    generator{
+      message => '{"test":"one"}'
+      codec => "json"
+      count => 1000
+    }
+
+    dead_letter_queue {
+      path => "<%=options[:dlq_dir]%>"
+      commit_offsets => true
+    }
+  }
+
+  filter {
+    if ([geoip]) {
+        mutate {
+            remove_field => ["geoip"]
+            add_field => {
+              "mutated" => "true"
+            }
+        }
+    }else{
+      mutate {
+        add_field => {
+          "geoip" => "somewhere"
+        }
+      }
+    }
+  }
+  output {
+    elasticsearch {}
+  }
+teardown_script:

qa/integration/services/logstash_service.rb

@@ -77,6 +77,10 @@ class LogstashService < Service
     end
   end
 
+  def start_background_with_config_settings(config, settings_file)
+    spawn_logstash("-f", "#{config}", "--path.settings", settings_file)
+  end
+
   def start_with_config_string(config)
     spawn_logstash("-e", "#{config} ")
   end

qa/integration/specs/dlq_spec.rb

@@ -0,0 +1,63 @@
+require_relative '../framework/fixture'
+require_relative '../framework/settings'
+require_relative '../services/logstash_service'
+require_relative '../framework/helpers'
+require "logstash/devutils/rspec/spec_helper"
+
+describe "Test Dead Letter Queue" do
+
+  before(:all) {
+    @fixture = Fixture.new(__FILE__)
+  }
+
+  after(:all) {
+    begin
+      es_client = @fixture.get_service("elasticsearch").get_client
+      es_client.indices.delete(index: 'logstash-*') unless es_client.nil?
+    ensure
+      @fixture.teardown
+    end
+  }
+
+  let(:dlq_dir) { Stud::Temporary.directory }
+
+  let(:dlq_config) {
+      {
+          "dead_letter_queue.enable" => true,
+          "path.dead_letter_queue" => dlq_dir,
+          "log.level" => "debug"
+      }
+  }
+
+  let!(:settings_dir) { Stud::Temporary.directory }
+  let!(:config_yaml) { dlq_config.to_yaml }
+  let!(:config_yaml_file) { ::File.join(settings_dir, "logstash.yml") }
+  let(:generator_config_file) { config_to_temp_file(@fixture.config("root",{ :dlq_dir => dlq_dir })) }
+
+
+  before(:each) do
+    IO.write(config_yaml_file, config_yaml)
+  end
+
+  it 'can index 1000 generated documents' do
+    logstash_service = @fixture.get_service("logstash")
+    logstash_service.start_background_with_config_settings(generator_config_file, settings_dir)
+    es_service = @fixture.get_service("elasticsearch")
+    es_client = es_service.get_client
+    # Wait for es client to come up
+    sleep(10)
+    # now we test if all data was indexed by ES, but first refresh manually
+    es_client.indices.refresh
+
+    logstash_service.wait_for_logstash
+    try(50) do
+      result = es_client.search(index: 'logstash-*', size: 0, q: '*')
+      expect(result["hits"]["total"]).to eq(1000)
+    end
+
+    # randomly checked for results and structured fields
+    result = es_client.search(index: 'logstash-*', size: 1, q: '*')
+    s = result["hits"]["hits"][0]["_source"]
+    expect(s["mutated"]).to eq("true")
+  end
+end