CPM handle 404 response gracefully with user-friendly log (#17052) (#17098)

Starting from es-output 12.0.2, a 404 response is treated as an error. Previously, central pipeline management considered 404 as an empty pipeline, not an error.

This commit restores the expected behavior by handling 404 gracefully and logs a user-friendly message.
It also removes the redundant cache of pipeline in CPM

Fixes: #17035
(cherry picked from commit e896cd7)

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

Author: github-actions[bot]

x-pack/lib/config_management/elasticsearch_source.rb

@@ -59,20 +59,19 @@ def get_pipeline_fetcher(es_version)
       def pipeline_configs
         logger.trace("Fetch remote config pipeline", :pipeline_ids => pipeline_ids)
 
-        begin
-          license_check(true)
-        rescue LogStash::LicenseChecker::LicenseError => e
-          if @cached_pipelines.nil?
-            raise e
-          else
-            return @cached_pipelines
-          end
-        end
+        license_check(true)
         es_version = get_es_version
         fetcher = get_pipeline_fetcher(es_version)
-        fetcher.fetch_config(es_version, pipeline_ids, client)
 
-        @cached_pipelines = fetcher.get_pipeline_ids.collect do |pid|
+        begin
+          fetcher.fetch_config(es_version, pipeline_ids, client)
+        rescue LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError => e
+          # es-output 12.0.2 throws 404 as error, but we want to handle it as empty config
+          return [] if e.response_code == 404
+          raise e
+        end
+
+        fetcher.get_pipeline_ids.collect do |pid|
           get_pipeline(pid, fetcher)
         end.compact
       end

x-pack/spec/config_management/elasticsearch_source_spec.rb

@@ -103,23 +103,7 @@
      "status" => 400}
   }
 
-  let(:elasticsearch_8_err_response) {
-    {"error" =>
-         {"root_cause" =>
-              [{"type" => "index_not_found_exception",
-                "reason" => "no such index [.logstash]",
-                "resource.type" => "index_expression",
-                "resource.id" => ".logstash",
-                "index_uuid" => "_na_",
-                "index" => ".logstash"}],
-          "type" => "index_not_found_exception",
-          "reason" => "no such index [.logstash]",
-          "resource.type" => "index_expression",
-          "resource.id" => ".logstash",
-          "index_uuid" => "_na_",
-          "index" => ".logstash"},
-     "status" => 404}
-  }
+  let(:elasticsearch_8_err_response) { {"error" => "Incorrect HTTP method for uri", "status" => 405} }
 
   before do
     extension.additionals_settings(system_settings)
@@ -466,13 +450,13 @@
                }]
       }.to_json
     end
-    let(:es_path) { ".logstash/_mget" }
+    let(:legacy_api) { ".logstash/_mget" }
     let(:request_body_string) { LogStash::Json.dump({ "docs" => [{ "_id" => pipeline_id }] }) }
 
     before do
       allow(mock_client).to receive(:get).with(system_indices_url_regex).and_return(LogStash::Json.load(elasticsearch_response))
       allow(mock_client).to receive(:get).with("/").and_return(es_version_response)
-      allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
+      allow(mock_client).to receive(:post).with(legacy_api, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
       allow(mock_license_client).to receive(:get).with('_xpack').and_return(valid_xpack_response)
       allow(mock_license_client).to receive(:get).with('/').and_return(cluster_info(LOGSTASH_VERSION))
 
@@ -493,7 +477,7 @@
             before :each do
               expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
               allow_any_instance_of(described_class).to receive(:logger).and_return(logger_stub)
-              allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
+              allow(mock_client).to receive(:post).with(legacy_api, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
             end
 
             let(:config) { "input { generator {} } filter { mutate {} } output { }" }
@@ -734,6 +718,7 @@
 
     describe "when exception occur" do
       let(:elasticsearch_response) { "" }
+      let(:bad_response_404) { LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError.new(404, "url", "request_body", "response_body") }
 
       before do
         expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
@@ -747,9 +732,21 @@
 
       it "raises the exception upstream in [7.9]" do
         allow(mock_client).to receive(:get).with("/").and_return(es_version_7_9_response)
-        expect(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_raise("Something bad")
+        expect(mock_client).to receive(:post).with(legacy_api, {}, request_body_string).and_raise("Something bad")
         expect { subject.pipeline_configs }.to raise_error /Something bad/
       end
+
+      it "returns empty pipeline when ES client raise BadResponseCodeError in [8]" do
+        allow(mock_client).to receive(:get).with("/").and_return(es_version_response)
+        expect(mock_client).to receive(:get).with(system_indices_url_regex).and_raise(bad_response_404)
+        expect(subject.pipeline_configs).to be_empty
+      end
+
+      it "returns empty pipeline when ES client raise BadResponseCodeError in [7.9]" do
+        allow(mock_client).to receive(:get).with("/").and_return(es_version_7_9_response)
+        expect(mock_client).to receive(:post).with(legacy_api, {}, request_body_string).and_raise(bad_response_404)
+        expect(subject.pipeline_configs).to be_empty
+      end
     end
   end