Add feature flag support for RATS (#6328)

* Add feature flag support for RATS

Adds feature flag support for RATS integration tests. First feature to use it
is persistent_queues. You can enable a FF in travis by using the travis matrix
and adding a env variable called FEATURE_FLAG=<feature_name>.

To use feature flag, you need a logstash.yml file which has this feature turned on.
Unfortunately --path.settings needs an entire dir where logstash.yml is located, not
just the logstash.yml file, so we need to checkin the fixture which has the feature yml
and log4j.

For example, you can now set FEATURE_FLAG=persistent_queues which means travis will
run tests twice, one with this feature enabled and once without it. When the feature is enaabled
the logstash.yml located in qa/integration/fixtures/persistent_queues is picked up whenever LS is started in tests

.travis.yml

@@ -9,7 +9,8 @@ rvm:
 jdk:
   - oraclejdk8
 env:
-  - INTEGRATION=true  
+  - INTEGRATION=true
+  - INTEGRATION=true FEATURE_FLAG=persistent_queues
 before_install:
   # Force bundler 1.12.5 because version 1.13 has issues, see https://github.com/fastlane/fastlane/issues/6065#issuecomment-246044617
   - gem uninstall -i /home/travis/.rvm/gems/jruby-1.7.25@global bundler

qa/integration/fixtures/persistent_queues/log4j2.properties

@@ -0,0 +1,83 @@
+status = error
+name = LogstashPropertiesConfig
+
+appender.console.type = Console
+appender.console.name = plain_console
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
+
+appender.json_console.type = Console
+appender.json_console.name = json_console
+appender.json_console.layout.type = JSONLayout
+appender.json_console.layout.compact = true
+appender.json_console.layout.eventEol = true
+
+appender.rolling.type = RollingFile
+appender.rolling.name = plain_rolling
+appender.rolling.fileName = ${sys:ls.logs}/logstash-${sys:ls.log.format}.log
+appender.rolling.filePattern = ${sys:ls.logs}/logstash-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
+appender.rolling.policies.type = Policies
+appender.rolling.policies.time.type = TimeBasedTriggeringPolicy
+appender.rolling.policies.time.interval = 1
+appender.rolling.policies.time.modulate = true
+appender.rolling.layout.type = PatternLayout
+appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %-.10000m%n
+
+appender.json_rolling.type = RollingFile
+appender.json_rolling.name = json_rolling
+appender.json_rolling.fileName = ${sys:ls.logs}/logstash-${sys:ls.log.format}.log
+appender.json_rolling.filePattern = ${sys:ls.logs}/logstash-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
+appender.json_rolling.policies.type = Policies
+appender.json_rolling.policies.time.type = TimeBasedTriggeringPolicy
+appender.json_rolling.policies.time.interval = 1
+appender.json_rolling.policies.time.modulate = true
+appender.json_rolling.layout.type = JSONLayout
+appender.json_rolling.layout.compact = true
+appender.json_rolling.layout.eventEol = true
+
+
+rootLogger.level = ${sys:ls.log.level}
+rootLogger.appenderRef.console.ref = ${sys:ls.log.format}_console
+rootLogger.appenderRef.rolling.ref = ${sys:ls.log.format}_rolling
+
+# Slowlog
+
+appender.console_slowlog.type = Console
+appender.console_slowlog.name = plain_console_slowlog
+appender.console_slowlog.layout.type = PatternLayout
+appender.console_slowlog.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %m%n
+
+appender.json_console_slowlog.type = Console
+appender.json_console_slowlog.name = json_console_slowlog
+appender.json_console_slowlog.layout.type = JSONLayout
+appender.json_console_slowlog.layout.compact = true
+appender.json_console_slowlog.layout.eventEol = true
+
+appender.rolling_slowlog.type = RollingFile
+appender.rolling_slowlog.name = plain_rolling_slowlog
+appender.rolling_slowlog.fileName = ${sys:ls.logs}/logstash-slowlog-${sys:ls.log.format}.log
+appender.rolling_slowlog.filePattern = ${sys:ls.logs}/logstash-slowlog-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
+appender.rolling_slowlog.policies.type = Policies
+appender.rolling_slowlog.policies.time.type = TimeBasedTriggeringPolicy
+appender.rolling_slowlog.policies.time.interval = 1
+appender.rolling_slowlog.policies.time.modulate = true
+appender.rolling_slowlog.layout.type = PatternLayout
+appender.rolling_slowlog.layout.pattern = [%d{ISO8601}][%-5p][%-25c] %.10000m%n
+
+appender.json_rolling_slowlog.type = RollingFile
+appender.json_rolling_slowlog.name = json_rolling_slowlog
+appender.json_rolling_slowlog.fileName = ${sys:ls.logs}/logstash-slowlog-${sys:ls.log.format}.log
+appender.json_rolling_slowlog.filePattern = ${sys:ls.logs}/logstash-slowlog-${sys:ls.log.format}-%d{yyyy-MM-dd}.log
+appender.json_rolling_slowlog.policies.type = Policies
+appender.json_rolling_slowlog.policies.time.type = TimeBasedTriggeringPolicy
+appender.json_rolling_slowlog.policies.time.interval = 1
+appender.json_rolling_slowlog.policies.time.modulate = true
+appender.json_rolling_slowlog.layout.type = JSONLayout
+appender.json_rolling_slowlog.layout.compact = true
+appender.json_rolling_slowlog.layout.eventEol = true
+
+logger.slowlog.name = slowlog
+logger.slowlog.level = trace
+logger.slowlog.appenderRef.console_slowlog.ref = ${sys:ls.log.format}_console_slowlog
+logger.slowlog.appenderRef.rolling_slowlog.ref = ${sys:ls.log.format}_rolling_slowlog
+logger.slowlog.additivity = false

qa/integration/fixtures/persistent_queues/logstash.yml

@@ -0,0 +1 @@
+queue.type: persisted

qa/integration/framework/settings.rb

@@ -12,7 +12,7 @@ class TestSettings
     test_name = File.basename(test_file_path, ".*" )
     @tests_settings_file = File.join(FIXTURES_DIR, "#{test_name}.yml")
     # Global suite settings
-    @suite_settings = YAML.load_file(SUITE_SETTINGS_FILE)
+    @suite_settings = YAML.load(ERB.new(File.new(SUITE_SETTINGS_FILE).read).result)
     # Per test settings, where one can override stuff and define test specific config
     @test_settings = YAML.load_file(@tests_settings_file)
     
@@ -50,4 +50,12 @@ class TestSettings
   def is_set?(key)
     @suite_settings.key?(key) || @test_settings.key?(key)
   end
+
+  def feature_flag
+    @suite_settings["feature_flag"].to_s.strip
+  end
+
+  def feature_config_dir
+    feature_flag.empty? ? nil: File.join(FIXTURES_DIR, feature)
+  end
 end

qa/integration/services/logstash_service.rb

@@ -13,6 +13,7 @@ class LogstashService < Service
   LS_BUILD_DIR = File.join(LS_ROOT_DIR, "build")
   LS_BIN = File.join("bin", "logstash")
   LS_CONFIG_FILE = File.join("config", "logstash.yml")
+  SETTINGS_CLI_FLAG = "--path.settings"
 
   STDIN_CONFIG = "input {stdin {}} output { }"
   RETRY_ATTEMPTS = 10
@@ -20,7 +21,7 @@ class LogstashService < Service
   @process = nil
   
   attr_reader :logstash_home
-  attr_reader :application_settings_file
+  attr_reader :default_settings_file
   attr_writer :env_variables
 
   def initialize(settings)
@@ -42,7 +43,7 @@ class LogstashService < Service
       raise "Logstash binary not found in path #{@logstash_home}" unless File.file? @logstash_bin
     end
     
-    @application_settings_file = File.join(@logstash_home, LS_CONFIG_FILE)
+    @default_settings_file = File.join(@logstash_home, LS_CONFIG_FILE)
     @monitoring_api = MonitoringAPI.new
   end
 
@@ -86,7 +87,7 @@ class LogstashService < Service
     Bundler.with_clean_env do
       out = Tempfile.new("duplex")
       out.sync = true
-      @process = ChildProcess.build(@logstash_bin, "-e", STDIN_CONFIG)
+      @process = build_child_process("-e", STDIN_CONFIG)
       # pipe STDOUT and STDERR to a file
       @process.io.stdout = @process.io.stderr = out
       @process.duplex = true
@@ -104,9 +105,8 @@ class LogstashService < Service
 
   # Spawn LS as a child process
   def spawn_logstash(*args)
-    puts "Starting Logstash #{@logstash_bin} #{args}" 
     Bundler.with_clean_env do
-      @process = ChildProcess.build(@logstash_bin, *args)
+      @process = build_child_process(*args)
       @env_variables.map { |k, v|  @process.environment[k] = v} unless @env_variables.nil?
       @process.io.inherit!
       @process.start
@@ -115,6 +115,20 @@ class LogstashService < Service
     end
   end
 
+  def build_child_process(*args)
+    feature_config_dir = @settings.feature_config_dir
+    # if we are using a feature flag and special settings dir to enable it, use it
+    # If some tests is explicitly using --path.settings, ignore doing this, because the tests
+    # chose to overwrite it.
+    if feature_config_dir && !args.include?(SETTINGS_CLI_FLAG)
+      args << "--path.settings"
+      args << feature_config_dir
+      puts "Found feature flag. Starting LS using --path.settings #{feature_config_dir}"
+    end
+    puts "Starting Logstash: #{@logstash_bin} #{args}"
+    ChildProcess.build(@logstash_bin, *args)
+  end
+
   def teardown
     if !@process.nil?
       # todo: put this in a sleep-wait loop to kill it force kill
@@ -172,6 +186,15 @@ class LogstashService < Service
     @process.pid
   end
 
+  def application_settings_file
+    feature_config_dir = @settings.feature_config_dir
+    unless feature_config_dir
+      @default_settings_file
+    else
+      File.join(feature_config_dir, "logstash.yml")
+    end
+  end
+
   def plugin_cli
     PluginCli.new(@logstash_home)
   end

qa/integration/specs/kafka_input_spec.rb

@@ -2,9 +2,12 @@ require_relative '../framework/fixture'
 require_relative '../framework/settings'
 require_relative '../services/logstash_service'
 require "rspec/wait"
+require "logstash/devutils/rspec/spec_helper"
 
 describe "Test Kafka Input" do
-  let(:timeout_seconds) { 30 }
+  let(:num_retries) { 60 }
+  let(:num_events) { 37 }
+
   before(:all) {
     @fixture = Fixture.new(__FILE__)
   }
@@ -17,9 +20,14 @@ describe "Test Kafka Input" do
     logstash_service = @fixture.get_service("logstash")
     logstash_service.start_background(@fixture.config)
 
-    wait(timeout_seconds).for { @fixture.output_exists? }.to be true
-    expect(@fixture.output_equals_expected?).to be true
-      lambda { "Expected File output to match what was ingested into Kafka." }
+    try(num_retries) do
+      expect(@fixture.output_exists?).to be true
+    end
+
+    try(num_retries) do
+      count = File.foreach(@fixture.actual_output).inject(0) {|c, line| c+1}
+      expect(count).to eq(num_events)
+    end
   end
 
 end

qa/integration/suite.yml

@@ -3,4 +3,5 @@
 verbose_mode: false  
 # Typically we use the binaries in LS_HOME/build. If you want to QA a LS in different location, 
 # use the absolute path below  
-#ls_home_abs_path: /tmp/logstash-5.0.0-alpha6
+#ls_home_abs_path: /tmp/logstash-5.0.0-alpha6
+feature_flag: <%= ENV['FEATURE_FLAG'] %>