tests: ls2ls delay checking until events have been processed (#17167) (#17250)

* tests: ls2ls delay checking until events have been processed

* Make sure upstream sends expected number of events before checking the expectation with downstream. Remove unnecessary or duplicated logics from the spec.

* Add exception handling in `wait_for_rest_api` to make wait for LS REST API retriable.

---------

Co-authored-by: Mashhur <[email protected]>
Co-authored-by: Mashhur <[email protected]>
(cherry picked from commit 73ffa24)

Co-authored-by: Ry Biesemeyer <[email protected]>

Author: mergify[bot]

qa/integration/services/logstash_service.rb

@@ -219,7 +219,9 @@ def is_port_open?
   # check REST API is responsive
   def rest_active?
     result = monitoring_api.node_info
-    started = !result.nil?
+    !result.nil?
+  rescue
+    return false
   end
 
   def monitoring_api

qa/integration/specs/logstash_to_logstash_spec.rb

@@ -27,12 +27,9 @@
 
   before(:all) {
     @fixture = Fixture.new(__FILE__)
-    # backup original setting file since we change API port number, and restore after all tests
-    FileUtils.cp(@fixture.get_service('logstash').application_settings_file, "#{@fixture.get_service('logstash').application_settings_file}.original")
   }
 
   after(:all) {
-    FileUtils.mv("#{@fixture.get_service('logstash').application_settings_file}.original", @fixture.get_service('logstash').application_settings_file)
     @fixture.teardown
   }
 
@@ -57,26 +54,10 @@ def run_logstash_instance(config_name, options = {}, &block)
                                     "--path.config", config_to_temp_file(@fixture.config(config_name, options)),
                                     "--path.data", get_temp_path_dir,
                                     "--api.http.port", api_port.to_s)
-    wait_for_logstash(logstash_service)
-
+    logstash_service.wait_for_rest_api
     yield logstash_service
-
   ensure
-    logstash_service&.teardown
-  end
-
-  def wait_for_logstash(service)
-    wait_in_seconds = 60
-    while wait_in_seconds > 0 do
-      begin
-        return if service.rest_active?
-      rescue => e
-        puts "Exception: #{e.message}"
-        wait_in_seconds -= 1
-        sleep 1
-      end
-    end
-    raise "Logstash is not responsive after 60 seconds."
+    logstash_service.teardown
   end
 
   let(:num_retries) { 60 }
@@ -97,8 +78,8 @@ def wait_for_logstash(service)
 
           try(num_retries) do
             downstream_event_stats = downstream_logstash_service.monitoring_api.event_stats
-
-            expect(downstream_event_stats).to include({"in" => num_events}), lambda { "expected #{num_events} events to have been received by downstream"}
+            expect(downstream_event_stats).to include({"in" => num_events}), lambda { "expected #{num_events} events to have been received by downstream" }
+            expect(downstream_event_stats).to include({"out" => num_events}), lambda { "expected #{num_events} events to have been processed by downstream" }
           end
 
           # make sure received events are in the file
@@ -122,4 +103,4 @@ def wait_for_logstash(service)
     include_examples "send events"
   end
 
-end
+end