Add new integration tests

Fixes #6031

qa/integration/fixtures/01_reload_config_spec.yml

@@ -0,0 +1,38 @@
+---
+name: Metrics test
+services:
+  - logstash
+config:
+  initial: |-
+    input {
+      tcp {
+        port => '<%=options[:port]%>'
+      }
+    }
+    output {
+      file {
+         path => '<%=options[:file]%>'
+         flush_interval => 0
+         codec => line { format => "%{message}" }
+      }
+    }
+  reload: |-
+    input {
+      tcp {
+        port => '<%=options[:port]%>'
+      }
+    }
+    filter {
+      grok {
+        match => {
+          "message" => "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}"
+        }
+      }
+    }
+    output {
+      file {
+         path => '<%=options[:file]%>'
+         flush_interval => 0
+         codec => json
+      }
+    }

qa/integration/fixtures/02_logstash_bin_smoke_spec.yml

@@ -0,0 +1,10 @@
+---
+name: Metrics test
+services:
+  - logstash
+config: |-
+ input {
+    tcp {
+      port => '<%=options[:port]%>'
+    }
+  } 

qa/integration/framework/helpers.rb

@@ -0,0 +1,31 @@
+# encoding: utf-8
+# Helper module for all tests
+
+def wait_for_port(port, retry_attempts)
+  tries = retry_attempts
+  while tries > 0
+    if is_port_open?(port)
+      break
+    else
+      sleep 1
+    end
+    tries -= 1
+  end
+end
+
+def is_port_open?(port)
+  begin
+    s = TCPSocket.open("localhost", port)
+    s.close
+    return true
+  rescue Errno::ECONNREFUSED, Errno::EHOSTUNREACH
+    return false
+  end
+end
+
+def config_to_temp_file(config)
+  f = Stud::Temporary.file
+  f.write(config)
+  f.close
+  f.path
+end

qa/integration/specs/01_reload_config_spec.rb

@@ -0,0 +1,85 @@
+require_relative '../framework/fixture'
+require_relative '../framework/settings'
+require_relative '../services/logstash_service'
+require_relative '../framework/helpers'
+require "logstash/devutils/rspec/spec_helper"
+require "socket"
+require "json"
+
+describe "Config reload" do
+  before(:all) {
+    @fixture = Fixture.new(__FILE__)
+  }
+
+  after(:all) {
+    @fixture.teardown
+  }
+  
+  let(:timeout_seconds) { 5 }
+  let(:initial_port) { 9198 }
+  let(:reload_port) { 9199 }
+  let(:retry_attempts) { 10 }
+  let(:output_file1) { Stud::Temporary.file.path }
+  let(:output_file2) { Stud::Temporary.file.path }
+  let(:sample_data) { '74.125.176.147 - - [11/Sep/2014:21:50:37 +0000] "GET /?flav=rss20 HTTP/1.1" 200 29941 "-" "FeedBurner/1.0 (http://www.FeedBurner.com)"' }
+  
+  let(:initial_config_file) { config_to_temp_file(@fixture.config("initial", { :port => initial_port, :file => output_file1 })) }
+  let(:reload_config_file) { config_to_temp_file(@fixture.config("reload", { :port => reload_port, :file => output_file2 })) }
+  
+  
+  def send_data(port, data)
+    socket = TCPSocket.new("127.0.0.1", port)
+    socket.puts(data)
+    socket.flush
+    socket.close
+  end
+
+  it "can reload config with new TCP port and grok pattern" do
+    logstash_service = @fixture.get_service("logstash")
+    logstash_service.spawn_logstash("-f", "#{initial_config_file}", "--config.reload.automatic", "true")
+    logstash_service.wait_for_logstash
+    wait_for_port(initial_port, retry_attempts)
+    
+    # try sending events with this
+    send_data(initial_port, sample_data)
+    Stud.try(retry_attempts.times, RSpec::Expectations::ExpectationNotMetError) do
+      expect(IO.read(output_file1).gsub("\n", "")).to eq(sample_data)
+    end
+    
+    # check metrics
+    result = logstash_service.monitoring_api.event_stats
+    expect(result["in"]).to eq(1)
+    expect(result["out"]).to eq(1)
+    
+    # do a reload
+    logstash_service.reload_config(initial_config_file, reload_config_file)
+
+    logstash_service.wait_for_logstash
+    wait_for_port(reload_port, retry_attempts)
+    
+    # make sure old socket is closed
+    expect(is_port_open?(initial_port)).to be false
+    
+    send_data(reload_port, sample_data)
+    Stud.try(retry_attempts.times, RSpec::Expectations::ExpectationNotMetError) do
+      expect(IO.read(output_file2).blank?).to be false
+    end
+    
+    # check metrics. It should be reset
+    result = logstash_service.monitoring_api.event_stats
+    expect(result["in"]).to eq(1)
+    expect(result["out"]).to eq(1)
+    
+    # check reload stats
+    reload_stats = logstash_service.monitoring_api.pipeline_stats["reloads"]
+    expect(reload_stats["successes"]).to eq(1)
+    expect(reload_stats["failures"]).to eq(0)
+    expect(reload_stats["last_success_timestamp"].blank?).to be false
+    expect(reload_stats["last_error"]).to eq(nil)
+    
+    # parse the results and validate
+    re = JSON.load(File.new(output_file2))
+    expect(re["clientip"]).to eq("74.125.176.147")
+    expect(re["response"]).to eq(200)
+  end
+end

qa/integration/specs/02_logstash_bin_smoke_spec.rb

@@ -0,0 +1,52 @@
+require_relative '../framework/fixture'
+require_relative '../framework/settings'
+require_relative '../services/logstash_service'
+require_relative '../framework/helpers'
+require "logstash/devutils/rspec/spec_helper"
+require "yaml"
+
+describe "Monitoring API" do
+  before(:all) {
+    @fixture = Fixture.new(__FILE__)
+  }
+
+  after(:all) {
+    @fixture.teardown
+  }
+  
+  let(:retry_attempts) { 10 }
+  let(:config1) { config_to_temp_file(@fixture.config("root", { :port => '9980' })) }
+  let(:config2) { config_to_temp_file(@fixture.config("root", { :port => '9981' })) }
+
+  it "a single instance can start with http server and port 9600" do
+    logstash_service = @fixture.get_service("logstash")
+    logstash_service.start_with_stdin
+    Stud.try(retry_attempts.times, RSpec::Expectations::ExpectationNotMetError) do
+      expect(is_port_open?(9600)).to be true
+    end
+    logstash_service.teardown
+  end
+  
+  it "multiple instances can be started on the same box" do
+    ls1 = @fixture.get_service("logstash")
+    ls1.spawn_logstash("-f", config1)
+    Stud.try(retry_attempts.times, RSpec::Expectations::ExpectationNotMetError) do
+      expect(is_port_open?(9600)).to be true
+    end
+
+    # bring up new LS instance
+    ls2 = LogstashService.new(@fixture.settings)
+    ls2.spawn_logstash("-f", config2)
+    Stud.try(retry_attempts.times, RSpec::Expectations::ExpectationNotMetError) do
+      expect(is_port_open?(9601)).to be true
+    end
+
+    expect(ls1.process_id).not_to eq(ls2.process_id)
+  end
+  
+  it "gets the right version" do
+    ls = @fixture.get_service("logstash")
+    expected = YAML.load_file(LogstashService::LS_VERSION_FILE)
+    expect(ls.get_version.strip).to eq("logstash #{expected['logstash']}")
+  end
+end