Support wait_for_status and timeout query params on root endpoint

docs/static/spec/openapi/logstash-api.yaml

@@ -66,6 +66,77 @@ tags:
     #   description:
     #   url:
 paths:
+  /:
+    get:
+      summary: Gets basic metadata
+      description: |
+        Shows basic metadata about the running logstash service. This includes build info, pipeline info and the service's status.
+      operationId: root
+      tags:
+        - root
+      parameters:
+        - name: wait_for_status
+          in: query
+          required: false
+          schema:
+            type: string
+          description: One of green, yellow or red. Will wait (until the timeout provided) until the status of the service changes to the one provided or better, i.e. green > yellow > red. The query param, timeout, is required when this query param is provided.
+        - name: timeout
+            in: query
+            required: false
+            schema:
+              type: string
+            description: Period to wait for the status to reach the requested target status. If the target status is not reached before the timeout expires, the request returns status code 503.
+        - $ref: "#/components/parameters/pretty"
+      responses:
+        '200':
+          description: Indicates a successful call
+          content:
+            application/json:
+              schema:
+                - type: object
+                  properties:
+                    host:
+                      type: string
+                    version:
+                      type: string
+                    http_address:
+                      type: string
+                    id:
+                      type: string
+                    name:
+                      type: string
+                    ephemeral_id:
+                      type: string
+                    snapshot:
+                      type: string
+                    status:
+                      type: string
+                    pipeline:
+                      type: object
+                      properties:
+                        workers:
+                          type: integer
+                        batch_size:
+                          type: integer
+                        batch_delay:
+                          type: integer
+              example:
+                host: "logstash-pipelines.example.com"
+                version: "9.3.0"
+                http_address: "127.0.0.1:9600"
+                id: "58df6f7c-eb5c-5d42-bc20-c7b22779aa12"
+                name: "logstash-pipelines"
+                ephermeral_id: "59df6f6c-eb5c-4d42-bc20-c7b44779aa12"
+                snapshot:
+                status: "green"
+                pipeline:
+                  workers: 10
+                  batch_size: 125
+                  batch_delay: 50
+      x-metaTags:
+        - content: Logstash
+          name: product_name
   /_node/jvm:
     get:
       summary: Gets node-level JVM info

logstash-core/lib/logstash/api/errors.rb

@@ -40,5 +40,25 @@ def status_code
         404
       end
     end
+
+    class BadRequest < ApiError
+      def initialize(message = nil)
+        super(message || "Bad Request")
+      end
+
+      def status_code
+        400
+      end
+    end
+
+    class RequestTimeout < ApiError
+      def initialize(message = nil)
+        super(message || "Request Timeout")
+      end
+
+      def status_code
+        408
+      end
+    end
   end
 end

logstash-core/lib/logstash/api/modules/root.rb

@@ -19,9 +19,82 @@ module LogStash
   module Api
     module Modules
       class Root < ::LogStash::Api::Modules::Base
+
+        HEALTH_STATUS = [
+          Java::OrgLogstashHealth::Status::GREEN,
+          Java::OrgLogstashHealth::Status::YELLOW,
+          Java::OrgLogstashHealth::Status::RED
+        ].map(&:external_value)
+
+        INVALID_HEALTH_STATUS_MESSAGE = "Invalid status '%s' provided. The valid statuses are: green, yellow, red."
+        INVALID_TIMEOUT_MESSAGE = "Invalid timeout '%s' provided."
+        TIMEOUT_REQUIRED_WITH_STATUS_MESSAGE = "A timeout must be provided along with a status."
+        TIMED_OUT_WAITING_FOR_STATUS_MESSAGE = "Timed out waiting for status '%s'."
+
         get "/" do
+          input_status = params[:wait_for_status]
+          input_timeout = params[:timeout]
+
+          if input_status && !(target_status = parse_status(input_status))
+            return status_error_response(input_status)
+          end
+
+          if input_timeout && !(timeout_s = parse_timeout_s(input_timeout))
+            return timeout_error_response(input_timeout)
+          end
+
+          if target_status
+            return timeout_required_with_status_response unless timeout_s
+            wait_for_status_and_respond(target_status, timeout_s)
+          else
+            command = factory.build(:system_basic_info)
+            respond_with(command.run)
+          end
+        end
+
+        private
+        def parse_timeout_s(timeout)
+          # If we call #to_seconds directly, the value will be rounded. So call to_nanos, then convert
+          # to a float and divide by 1e9 to get the value in seconds.
+          LogStash::Util::TimeValue.from_value(timeout).to_nanos.to_f/1_000_000_000
+        rescue ArgumentError
+        end
+
+        def parse_status(input_status)
+          target_status = input_status.downcase
+          target_status if HEALTH_STATUS.include?(target_status)
+        end
+
+        def timeout_error_response(timeout)
+          respond_with(BadRequest.new(INVALID_TIMEOUT_MESSAGE % [timeout]))
+        end
+
+        def status_error_response(target_status)
+          respond_with(BadRequest.new(INVALID_HEALTH_STATUS_MESSAGE % [target_status]))
+        end
+
+        def timeout_required_with_status_response
+          respond_with(BadRequest.new(TIMEOUT_REQUIRED_WITH_STATUS_MESSAGE))
+        end
+
+        def wait_for_status_and_respond(target_status, timeout)
+          wait_interval = 0.2 # seconds
+          deadline = Time.now + timeout
+
+          loop do
+            current_status = HEALTH_STATUS.index(agent.health_observer.status.external_value)
+            break if current_status <= HEALTH_STATUS.index(target_status)
+
+            time_remaining = deadline - Time.now
+            if time_remaining <= 0
+              return respond_with(RequestTimeout.new(TIMED_OUT_WAITING_FOR_STATUS_MESSAGE % [target_status]))
+            end
+            sleep((time_remaining <= wait_interval) ? time_remaining : wait_interval)
+            wait_interval = wait_interval * 2
+          end
+
           command = factory.build(:system_basic_info)
-          respond_with command.run
+          respond_with(command.run)
         end
       end
     end