spec: improve ls2ls spec (#17114) (#17117)

* spec: improve ls2ls spec

 - fixes upstream/downstream convention
   - upstream: the sending logstash (has an LS output)
   - downstream: the receiving logstash (has an LS input)
 - helper `run_logstash_instance` yields the `LogstashService` instance
   and handles the teardown.
 - pass the pipeline id and node name to the LS instances via command line
   flags to make logging easier to differentiate
 - use the generator input's sequence id to ensure that the _actual_ events
   generated are received by the downstream pipeline

* start with port-offset 100

Co-authored-by: Mashhur <[email protected]>

---------

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

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

Author: github-actions[bot]

qa/integration/fixtures/logstash_to_logstash_spec.yml

@@ -14,7 +14,7 @@ config:
       file {
         path => '<%=options[:output_file_path]%>'
         flush_interval => 0
-        codec => line { format => "%{message}" }
+        codec => line { format => "%{[event][sequence]}:%{message}" }
       }
     }
   basic_ls_output: |-

qa/integration/specs/logstash_to_logstash_spec.rb

@@ -43,20 +43,26 @@ def get_temp_path_dir
     tmp_data_path
   end
 
-  def run_logstash_instance(config_name, options = {})
-    @next_api_port_offset = (@next_api_port_offset||0).next.modulo(1000) # cycle through 1000 possibles
+  def run_logstash_instance(config_name, options = {}, &block)
+    @next_api_port_offset = (@next_api_port_offset||100).next.modulo(1000) # cycle through 1000 possibles
     api_port = 9600 + @next_api_port_offset
 
     # to avoid LogstashService's clean-from-tarball default behaviour, we need
     # to tell it where our LOGSTASH_HOME is in the existing service
     existing_fixture_logstash_home = @fixture.get_service("logstash").logstash_home
     logstash_service = LogstashService.new(@fixture.settings.override("ls_home_abs_path" => existing_fixture_logstash_home), api_port)
 
-    logstash_service.spawn_logstash("--path.config", config_to_temp_file(@fixture.config(config_name, options)),
+    logstash_service.spawn_logstash("--node.name", config_name,
+                                    "--pipeline.id", config_name,
+                                    "--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
+
+    yield logstash_service
+
+  ensure
+    logstash_service&.teardown
   end
 
   def wait_for_logstash(service)
@@ -86,26 +92,25 @@ def wait_for_logstash(service)
     }
 
     it "successfully send events" do
-      upstream_logstash_service = run_logstash_instance(input_config_name, all_config_options)
-      downstream_logstash_service = run_logstash_instance(output_config_name, all_config_options)
+      run_logstash_instance(input_config_name, all_config_options) do |downstream_logstash_service|
+        run_logstash_instance(output_config_name, all_config_options) do |upstream_logstash_service|
 
-      try(num_retries) do
-        event_stats = upstream_logstash_service.monitoring_api.event_stats
-        if event_stats
-          expect(event_stats["in"]).to eq(num_events)
-        end
-      end
+          try(num_retries) do
+            downstream_event_stats = downstream_logstash_service.monitoring_api.event_stats
 
-      upstream_logstash_service.teardown
-      downstream_logstash_service.teardown
+            expect(downstream_event_stats).to include({"in" => num_events}), lambda { "expected #{num_events} events to have been received by downstream"}
+          end
 
-      # make sure received events are in the file
-      file_output_path = File.join(upstream_logstash_service.logstash_home, output_file_path_with_datetime)
-      expect(File).to exist(file_output_path), "Logstash to Logstash output file: #{file_output_path} does not exist"
-      count = File.foreach(file_output_path).inject(0) { |c, _| c + 1 }
-      expect(count).to eq(num_events)
+          # make sure received events are in the file
+          file_output_path = File.join(downstream_logstash_service.logstash_home, output_file_path_with_datetime)
+          expect(File).to exist(file_output_path), "Logstash to Logstash output file: #{file_output_path} does not exist"
+          actual_lines = File.read(file_output_path).lines.to_a
+          expected_lines = (0...num_events).map { |sequence| "#{sequence}:Hello world!\n" }
+          expect(actual_lines).to match_array(expected_lines)
 
-      File.delete(file_output_path)
+          File.delete(file_output_path)
+        end
+      end
     end
   end