diff --git a/docs/static/spec/openapi/logstash-api.yaml b/docs/static/spec/openapi/logstash-api.yaml
index 517a2080c53..4778a0d91ae 100644
--- a/docs/static/spec/openapi/logstash-api.yaml
+++ b/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
diff --git a/logstash-core/lib/logstash/api/errors.rb b/logstash-core/lib/logstash/api/errors.rb
index ad783c63aa5..f15f77c7999 100644
--- a/logstash-core/lib/logstash/api/errors.rb
+++ b/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
diff --git a/logstash-core/lib/logstash/api/modules/root.rb b/logstash-core/lib/logstash/api/modules/root.rb
index 260319e78b1..e6fe62fcc64 100644
--- a/logstash-core/lib/logstash/api/modules/root.rb
+++ b/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
diff --git a/logstash-core/spec/logstash/api/modules/root_spec.rb b/logstash-core/spec/logstash/api/modules/root_spec.rb
index 22fbfe44de5..703277c3a11 100644
--- a/logstash-core/spec/logstash/api/modules/root_spec.rb
+++ b/logstash-core/spec/logstash/api/modules/root_spec.rb
@@ -29,4 +29,276 @@
   end
 
   include_examples "not found"
+
+  describe 'wait_for_status' do
+
+    let(:response) { get request }
+
+    context 'timeout' do
+
+      context 'no timeout provided' do
+
+        let(:request) { "/" }
+
+        include_examples "returns successfully without waiting"
+      end
+
+      context 'timeout is provided' do
+
+        let(:request) { "/?timeout=#{timeout}" }
+
+        context 'timeout does not have units' do
+
+          let(:timeout) { '1' }
+          let(:error_message) { described_class::INVALID_TIMEOUT_MESSAGE % [timeout] }
+
+          include_examples 'bad request response'
+        end
+
+        context 'timeout number is not an integer' do
+
+          let(:timeout) { '1.0s' }
+          let(:error_message) { described_class::INVALID_TIMEOUT_MESSAGE % [timeout] }
+
+          include_examples 'bad request response'
+        end
+
+        context 'timeout is not in the accepted format' do
+
+          let(:timeout) { 'invalid' }
+          let(:error_message) { described_class::INVALID_TIMEOUT_MESSAGE % [timeout] }
+
+          include_examples 'bad request response'
+        end
+
+        context 'valid timeout is provided' do
+
+          context 'no status is provided' do
+
+            let(:timeout) { '1s' }
+
+            include_examples "returns successfully without waiting"
+          end
+
+          context 'status is provided' do
+
+            let(:timeout_num) { 2 }
+            let(:timeout_string) { "#{timeout_num}s"}
+            let(:status) { 'green' }
+            let(:request) { "/?wait_for_status=#{status}&timeout=#{timeout_string}" }
+
+            let(:return_statuses) do
+              [
+                org.logstash.health.Status::RED,
+                org.logstash.health.Status::GREEN
+
+              ]
+            end
+
+            it 'returns status code 200' do
+              expect(response.status).to be 200
+            end
+
+            include_examples "waits until the target status (or better) is reached and returns successfully"
+          end
+        end
+      end
+    end
+
+    context 'status' do
+
+      context 'no status provided' do
+
+        let(:request) { '/'}
+
+        include_examples "returns successfully without waiting"
+      end
+
+      context 'status is provided' do
+
+        context 'status is not valid' do
+
+          let(:status) { 'invalid' }
+          let(:error_message) { described_class::INVALID_HEALTH_STATUS_MESSAGE % [status] }
+          let(:request) { "/?wait_for_status=#{status}&timeout=1s" }
+
+          include_examples 'bad request response'
+        end
+
+        context 'status is valid' do
+
+          context 'no timeout is provided' do
+
+            let(:request) { "/?wait_for_status=green" }
+            let(:error_message) { described_class::TIMEOUT_REQUIRED_WITH_STATUS_MESSAGE }
+
+            include_examples "bad request response"
+          end
+
+          context 'timeout is provided' do
+
+            let(:timeout_num) { 2 }
+            let(:timeout_string) { "#{timeout_num}s"}
+            let(:status) { 'green' }
+            let(:request) { "/?wait_for_status=#{status}&timeout=#{timeout_string}" }
+
+            let(:return_statuses) do
+              [
+                org.logstash.health.Status::RED,
+                org.logstash.health.Status::GREEN
+
+              ]
+            end
+
+            include_examples 'waits until the target status (or better) is reached and returns successfully'
+          end
+        end
+      end
+    end
+
+    context 'timeout and status provided' do
+
+      let(:timeout_num) { 2 }
+      let(:timeout_units) { 's' }
+      let(:timeout_string) { "#{timeout_num}#{timeout_units}"}
+      let(:status) { 'green' }
+      let(:request) { "/?wait_for_status=#{status}&timeout=#{timeout_string}" }
+
+      context "the status doesn't change before the timeout" do
+
+        let(:return_statuses) do
+          [
+            org.logstash.health.Status::RED
+          ]
+        end
+
+        include_examples 'times out waiting for target status (or better)'
+      end
+
+      context 'target status is green' do
+
+        let(:status) { 'green' }
+
+        context 'the status does not change' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::YELLOW
+            ]
+          end
+
+          include_examples 'times out waiting for target status (or better)'
+        end
+
+        context 'the status changes to green' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::GREEN
+            ]
+          end
+
+          include_examples 'waits until the target status (or better) is reached and returns successfully'
+        end
+      end
+
+      context 'target status is yellow' do
+
+        let(:status) { 'yellow' }
+
+        context 'the status does not change' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED
+            ]
+          end
+
+          include_examples 'times out waiting for target status (or better)'
+        end
+
+        context 'the status changes to yellow' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::YELLOW
+            ]
+          end
+
+          include_examples 'waits until the target status (or better) is reached and returns successfully'
+        end
+
+        context 'the status changes to green' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::GREEN
+            ]
+          end
+
+          include_examples 'waits until the target status (or better) is reached and returns successfully'
+        end
+      end
+
+      context 'target status is red' do
+
+        let(:status) { 'red' }
+
+        context 'the status does not change' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED
+            ]
+          end
+
+          include_examples "returns successfully without waiting"
+        end
+
+        context 'the status changes to yellow' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::YELLOW
+            ]
+          end
+
+          include_examples "returns successfully without waiting"
+        end
+
+        context 'the status changes to green' do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED,
+              org.logstash.health.Status::GREEN
+            ]
+          end
+
+          include_examples "returns successfully without waiting"
+        end
+      end
+
+      context 'timeout units is ms' do
+
+        let(:timeout_units) { 'ms' }
+
+        context "the status doesn't change before the timeout" do
+
+          let(:return_statuses) do
+            [
+              org.logstash.health.Status::RED
+            ]
+          end
+
+          include_examples 'times out waiting for target status (or better)'
+        end
+      end
+    end
+  end
 end
diff --git a/logstash-core/spec/support/shared_examples.rb b/logstash-core/spec/support/shared_examples.rb
index 29eb4953b19..0706148590a 100644
--- a/logstash-core/spec/support/shared_examples.rb
+++ b/logstash-core/spec/support/shared_examples.rb
@@ -119,3 +119,72 @@
     expect(LogStash::Json.load(last_response.body)["path"]).not_to be_nil
   end
 end
+
+shared_examples "returns successfully without waiting" do
+
+  it 'returns status code 200' do
+    expect(response.status).to be 200
+  end
+
+  it 'returns immediately' do
+    start_time = Time.now
+    response
+    end_time = Time.now
+    expect(end_time - start_time).to be < 0.5
+  end
+end
+
+shared_examples "bad request response" do
+
+  it 'returns an error response' do
+    expect(response.body).to include(error_message)
+  end
+
+  it 'returns an 400 status' do
+    expect(response.status).to be 400
+  end
+end
+
+shared_examples 'waits until the target status (or better) is reached and returns successfully' do
+
+  before do
+    allow(@agent.health_observer).to receive(:status).and_return(*return_statuses)
+  end
+
+  it 'returns status code 200' do
+    expect(response.status).to be 200
+  end
+
+  it 'checks for the status until the target status (or better) is reached' do
+    start_time = Time.now
+    response
+    end_time = Time.now
+    expect(end_time - start_time).to be <= timeout_num
+  end
+end
+
+shared_examples 'times out waiting for target status (or better)' do
+
+  let(:expected_time_elapsed) do
+    LogStash::Util::TimeValue.from_value(timeout_string).to_nanos.to_f/1_000_000_000
+  end
+
+  before do
+    allow(@agent.health_observer).to receive(:status).and_return(*return_statuses)
+  end
+
+  it 'times out waiting for target status (or better)' do
+    start_time = Time.now
+    response
+    end_time = Time.now
+    expect(end_time - start_time).to be >= expected_time_elapsed
+  end
+
+  it 'returns status code 408' do
+    expect(response.status).to eq 408
+  end
+
+  it 'returns a message saying the request timed out' do
+    expect(response.body).to include(described_class::TIMED_OUT_WAITING_FOR_STATUS_MESSAGE % [status])
+  end
+end