replace direct access of hidden indices with system indices api

x-pack/lib/config_management/elasticsearch_source.rb

@@ -18,7 +18,6 @@ class ElasticsearchSource < LogStash::Config::Source::Base
 
       class RemoteConfigError < LogStash::Error; end
 
-      PIPELINE_INDEX = ".logstash"
       # exclude basic
       VALID_LICENSES = %w(trial standard gold platinum enterprise)
       FEATURE_INTERNAL = 'management'
@@ -50,6 +49,21 @@ def config_conflict?
         false
       end
 
+      # decide using system indices api (7.10+) or legacy api (< 7.10) base on elasticsearch server version
+      def pipeline_fetcher_factory
+        response = client.get("/")
+
+        if response["error"]
+          raise RemoteConfigError, "Cannot find elasticsearch version, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
+        end
+
+        logger.debug("Elasticsearch version ", response["version"]["number"])
+        version_number = response["version"]["number"].split(".")
+        first = version_number[0].to_i
+        second = version_number[1].to_i
+        (first >= 8 || (first == 7 && second >= 10))? SystemIndicesFetcher.new: LegacyHiddenIndicesFetcher.new
+      end
+
       def pipeline_configs
         logger.trace("Fetch remote config pipeline", :pipeline_ids => pipeline_ids)
 
@@ -63,33 +77,27 @@ def pipeline_configs
           end
         end
 
-        response = fetch_config(pipeline_ids)
+        fetcher = pipeline_fetcher_factory
+        response = fetcher.fetch_config(pipeline_ids, client)
 
         if response["error"]
           raise RemoteConfigError, "Cannot find find configuration for pipeline_id: #{pipeline_ids}, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
         end
 
-        if response["docs"].nil?
-          logger.debug("Server returned an unknown or malformed document structure", :response => response)
-          raise RemoteConfigError, "Elasticsearch returned an unknown or malformed document structure"
-        end
+        response = fetcher.format_response(response)
 
-        # Cache pipelines to handle the case where a remote configuration error can render a pipeline unusable
-        # it is not reloadable
-        @cached_pipelines = response["docs"].collect do |document|
-          get_pipeline(document)
+        @cached_pipelines = pipeline_ids.collect do |pid|
+          get_pipeline(pid, response, fetcher)
         end.compact
       end
 
-      def get_pipeline(response)
-        pipeline_id = response["_id"]
-
-        if response["found"] == false
+      def get_pipeline(pipeline_id, response, fetcher)
+        if response.has_key?(pipeline_id) == false
           logger.debug("Could not find a remote configuration for a specific `pipeline_id`", :pipeline_id => pipeline_id)
           return nil
         end
 
-        config_string = response.fetch("_source", {})["pipeline"]
+        config_string = fetcher.get_single_pipeline_setting(response, pipeline_id)["pipeline"]
 
         raise RemoteConfigError, "Empty configuration for pipeline_id: #{pipeline_id}" if config_string.nil? || config_string.empty?
 
@@ -100,7 +108,7 @@ def get_pipeline(response)
         settings.set("pipeline.id", pipeline_id)
 
         # override global settings with pipeline settings from ES, if any
-        pipeline_settings = response["_source"]["pipeline_settings"]
+        pipeline_settings = fetcher.get_single_pipeline_setting(response, pipeline_id)["pipeline_settings"]
         unless pipeline_settings.nil?
           pipeline_settings.each do |setting, value|
             if SUPPORTED_PIPELINE_SETTINGS.include? setting
@@ -127,15 +135,6 @@ def build_client
         es.build_client
       end
 
-      def fetch_config(pipeline_ids)
-        request_body_string = LogStash::Json.dump({ "docs" => pipeline_ids.collect { |pipeline_id| { "_id" => pipeline_id } } })
-        client.post(config_path, {}, request_body_string)
-      end
-
-      def config_path
-        "#{PIPELINE_INDEX}/_mget"
-      end
-
       def populate_license_state(xpack_info)
         if xpack_info.failed?
           {
@@ -193,5 +192,55 @@ def client
         @client ||= build_client
       end
     end
+
+    class SystemIndicesFetcher
+      include LogStash::Util::Loggable
+
+      SYSTEM_INDICES_API_PATH = "_logstash/pipeline"
+
+      def fetch_config(pipeline_ids, client)
+        path_ids = pipeline_ids.join(",")
+        client.get("#{SYSTEM_INDICES_API_PATH}/#{path_ids}")
+      end
+
+      def format_response(response)
+        # for backwards compatibility
+        response
+      end
+
+      def get_single_pipeline_setting(response, pipeline_id)
+        response.fetch(pipeline_id, {})
+      end
+    end
+
+    # TODO clean up LegacyHiddenIndicesFetcher when 7.9.* is deprecated
+    class LegacyHiddenIndicesFetcher
+      include LogStash::Util::Loggable
+
+      PIPELINE_INDEX = ".logstash"
+
+      def fetch_config(pipeline_ids, client)
+        request_body_string = LogStash::Json.dump({ "docs" => pipeline_ids.collect { |pipeline_id| { "_id" => pipeline_id } } })
+        client.post("#{PIPELINE_INDEX}/_mget", {}, request_body_string)
+      end
+
+      # transform legacy response to be similar to system indices response
+      def format_response(response)
+        if response["docs"].nil?
+          logger.debug("Server returned an unknown or malformed document structure", :response => response)
+          raise ElasticsearchSource::RemoteConfigError, "Elasticsearch returned an unknown or malformed document structure"
+        end
+
+        response["docs"].map { |pipeline|
+          {pipeline["_id"] => pipeline} if pipeline.fetch("found", false)
+        }.compact
+        .reduce({}, :merge)
+      end
+
+      def get_single_pipeline_setting(response, pipeline_id)
+        response.fetch(pipeline_id, {}).fetch("_source", {})
+      end
+    end
+
   end
 end

x-pack/qa/integration/management/multiple_pipelines_spec.rb

@@ -17,7 +17,7 @@
       "hello" => nil
     }
 
-    cleanup_elasticsearch(".logstash*")
+    cleanup_system_indices(@pipelines.keys)
     cleanup_elasticsearch(".monitoring-logstash*")
 
     @pipelines.each do |pipeline_id, config|

x-pack/qa/integration/management/read_configuration_spec.rb

@@ -31,7 +31,7 @@ def logstash_options(pipeline_id, wait_condition)
   def start_services(elasticsearch_options, logstash_options)
     @elasticsearch_service = elasticsearch(elasticsearch_options)
 
-    cleanup_elasticsearch(".logstash*")
+    cleanup_system_indices([PIPELINE_ID])
 
     config = "input { generator { count => 100 } tcp { port => 6000 } } output { null {} }"
     push_elasticsearch_config(PIPELINE_ID, config)

x-pack/qa/integration/support/helpers.rb

@@ -8,6 +8,7 @@
 require "fileutils"
 require "stud/try"
 require "open3"
+require "time"
 
 VERSIONS_YML_PATH = File.join(File.dirname(__FILE__), "..", "..", "..", "..", "versions.yml")
 VERSION_PATH = File.join(File.dirname(__FILE__), "..", "..", "..", "VERSION")
@@ -94,15 +95,37 @@ def elasticsearch_client(options = { :url => "http://elastic:#{elastic_password}
   Elasticsearch::Client.new(options)
 end
 
+def es_version
+  response = elasticsearch_client.perform_request(:get, "")
+  response.body["version"]["number"].gsub(/(\d+\.\d+)\..+/, '\1').to_f
+end
+
 def push_elasticsearch_config(pipeline_id, config)
-  elasticsearch_client.index :index => '.logstash', :type => "_doc", id: pipeline_id, :body => { :pipeline => config }
+  if es_version >= 7.10
+    elasticsearch_client.perform_request(:put, "_logstash/pipeline/#{pipeline_id}", {},
+      { :pipeline => config, :username => "log.stash", :pipeline_metadata => {:version => "1" },
+              :pipeline_settings => {"pipeline.batch.delay": "50"}, :last_modified => Time.now.utc.iso8601})
+  else
+    elasticsearch_client.index :index => '.logstash', :type => "_doc", id: pipeline_id, :body => { :pipeline => config }
+  end
 end
 
 def cleanup_elasticsearch(index = MONITORING_INDEXES)
   elasticsearch_client.indices.delete :index => index
   elasticsearch_client.indices.refresh
 end
 
+def cleanup_system_indices(pipeline_ids)
+  pipeline_ids.each do |id|
+    begin
+      elasticsearch_client.perform_request(:delete, "_logstash/pipeline/#{id}")
+    rescue Elasticsearch::Transport::Transport::Errors::NotFound => e
+      puts ".logstash can be empty #{e.message}"
+    end
+  end
+  elasticsearch_client.indices.refresh
+end
+
 def logstash_command_append(cmd, argument, value)
   if cmd !~ /#{Regexp.escape(argument)}/
     cmd << " #{argument} #{value}"
@@ -136,4 +159,4 @@ def verify_response!(cmd, response)
   unless response.successful?
     raise "Something went wrong when installing xpack,\ncmd: #{cmd}\nresponse: #{response}"
   end
-end
+end