diff --git a/.buildkite/scripts/benchmark/config/logstash.yml b/.buildkite/scripts/benchmark/config/logstash.yml
index 5b228a64edd..59d64888934 100644
--- a/.buildkite/scripts/benchmark/config/logstash.yml
+++ b/.buildkite/scripts/benchmark/config/logstash.yml
@@ -3,6 +3,7 @@ pipeline.workers: ${WORKER}
 pipeline.batch.size: ${BATCH_SIZE}
 queue.type: ${QTYPE}
 
+xpack.monitoring.allow_legacy_collection: true
 xpack.monitoring.enabled: true
 xpack.monitoring.elasticsearch.username: ${MONITOR_ES_USER}
 xpack.monitoring.elasticsearch.password: ${MONITOR_ES_PW}
diff --git a/config/logstash.yml b/config/logstash.yml
index 0ae8a0abb82..c80f38efade 100644
--- a/config/logstash.yml
+++ b/config/logstash.yml
@@ -338,6 +338,8 @@
 #
 # X-Pack Monitoring
 # https://www.elastic.co/guide/en/logstash/current/monitoring-logstash.html
+# Flag to allow the legacy internal monitoring (default: false)
+#xpack.monitoring.allow_legacy_collection: false
 #xpack.monitoring.enabled: false
 #xpack.monitoring.elasticsearch.username: logstash_system
 #xpack.monitoring.elasticsearch.password: password
diff --git a/docker/data/logstash/env2yaml/env2yaml.go b/docker/data/logstash/env2yaml/env2yaml.go
index 588f1bb41e2..e0680957a1e 100644
--- a/docker/data/logstash/env2yaml/env2yaml.go
+++ b/docker/data/logstash/env2yaml/env2yaml.go
@@ -17,7 +17,6 @@ package main
 import (
 	"errors"
 	"fmt"
-	"gopkg.in/yaml.v2"
 	"io/ioutil"
 	"log"
 	"os"
@@ -82,6 +81,7 @@ var validSettings = []string{
 	"api.auth.basic.password_policy.include.symbol",
 	"allow_superuser",
 	"monitoring.cluster_uuid",
+	"xpack.monitoring.allow_legacy_collection",
 	"xpack.monitoring.enabled",
 	"xpack.monitoring.collection.interval",
 	"xpack.monitoring.elasticsearch.hosts",
diff --git a/docs/static/monitoring/monitoring-internal-legacy.asciidoc b/docs/static/monitoring/monitoring-internal-legacy.asciidoc
index 7b1730baef0..bb12923d8e5 100644
--- a/docs/static/monitoring/monitoring-internal-legacy.asciidoc
+++ b/docs/static/monitoring/monitoring-internal-legacy.asciidoc
@@ -7,6 +7,11 @@
 
 deprecated[7.9.0]
 
+NOTE: Starting from version 9.0, legacy internal collection is behind a feature flag and is turned off by default. 
+Set `xpack.monitoring.allow_legacy_collection` to `true` to allow access to the feature.
+
+Using <<monitoring-with-ea,{agent} for monitoring>> is a better alternative for most {ls} deployments. 
+
 ==== Components for legacy collection
 
 Monitoring {ls} with legacy collection uses these components:
@@ -57,7 +62,7 @@ monitoring cluster will show the Logstash metrics under the _monitoring_ cluster
 
 --
 
-. Verify that the `xpack.monitoring.collection.enabled` setting is `true` on the
+. Verify that the `xpack.monitoring.allow_legacy_collection` and `xpack.monitoring.collection.enabled` settings are `true` on the
 production cluster. If that setting is `false`, the collection of monitoring data
 is disabled in {es} and data is ignored from all other sources.
 
diff --git a/logstash-core/lib/logstash/api/commands/default_metadata.rb b/logstash-core/lib/logstash/api/commands/default_metadata.rb
index 635e3e5f43a..7c8f309db80 100644
--- a/logstash-core/lib/logstash/api/commands/default_metadata.rb
+++ b/logstash-core/lib/logstash/api/commands/default_metadata.rb
@@ -71,6 +71,8 @@ def http_address
 
         private
         def enabled_xpack_monitoring?
+          LogStash::SETTINGS.registered?("xpack.monitoring.allow_legacy_collection") &&
+          LogStash::SETTINGS.get_value("xpack.monitoring.allow_legacy_collection") &&
           LogStash::SETTINGS.registered?("xpack.monitoring.enabled") &&
           LogStash::SETTINGS.get("xpack.monitoring.enabled")
         end
diff --git a/logstash-core/spec/logstash/api/commands/default_metadata_spec.rb b/logstash-core/spec/logstash/api/commands/default_metadata_spec.rb
index acdbc8a5091..c93a7737794 100644
--- a/logstash-core/spec/logstash/api/commands/default_metadata_spec.rb
+++ b/logstash-core/spec/logstash/api/commands/default_metadata_spec.rb
@@ -54,13 +54,22 @@ def registerIfNot(setting)
         )
     end
 
-    it "check monitoring exist when monitoring is enabled" do
+    it "check monitoring section exist when legacy monitoring is enabled and allowed" do
+      LogStash::SETTINGS.set_value("xpack.monitoring.allow_legacy_collection", true)
       LogStash::SETTINGS.set_value("xpack.monitoring.enabled", true)
       expect(report.keys).to include(
         :monitoring
         )
     end
 
+    it "check monitoring section does not appear when legacy monitoring is not allowed but enabled" do
+      LogStash::SETTINGS.set_value("xpack.monitoring.allow_legacy_collection", false)
+      LogStash::SETTINGS.set_value("xpack.monitoring.enabled", true)
+      expect(report.keys).not_to include(
+                               :monitoring
+                             )
+    end
+
     it "check monitoring does not appear when not enabled and nor cluster_uuid is defined" do
       LogStash::SETTINGS.set_value("xpack.monitoring.enabled", false)
       LogStash::SETTINGS.get_setting("monitoring.cluster_uuid").reset
diff --git a/x-pack/lib/monitoring/monitoring.rb b/x-pack/lib/monitoring/monitoring.rb
index fac0b7565bf..bdc6d51f11d 100644
--- a/x-pack/lib/monitoring/monitoring.rb
+++ b/x-pack/lib/monitoring/monitoring.rb
@@ -155,8 +155,10 @@ def after_agent(runner)
       # For versions prior to 6.3 the default value of "xpack.monitoring.enabled" was true
       # For versions 6.3+ the default of "xpack.monitoring.enabled" is false.
       # To help keep passivity, assume that if "xpack.monitoring.elasticsearch.hosts" has been set that monitoring should be enabled.
-      # return true if xpack.monitoring.enabled=true (explicitly) or xpack.monitoring.elasticsearch.hosts is configured
+      # return true if xpack.monitoring.allow_legacy_collection=true and xpack.monitoring.enabled=true (explicitly) or xpack.monitoring.elasticsearch.hosts is configured
       def monitoring_enabled?(settings)
+        log_warn_if_legacy_is_enabled_and_not_allowed(settings)
+        return false unless settings.get_value("xpack.monitoring.allow_legacy_collection")
         return settings.get_value("monitoring.enabled") if settings.set?("monitoring.enabled")
         return settings.get_value("xpack.monitoring.enabled") if settings.set?("xpack.monitoring.enabled")
 
@@ -170,6 +172,15 @@ def monitoring_enabled?(settings)
         end
       end
 
+      def log_warn_if_legacy_is_enabled_and_not_allowed(settings)
+        allowed = settings.get_value("xpack.monitoring.allow_legacy_collection")
+        legacy_monitoring_enabled = (settings.get_value("xpack.monitoring.enabled") || settings.get_value("monitoring.enabled"))
+        if !allowed && legacy_monitoring_enabled
+          logger.warn("You have enabled legacy internal monitoring. However, starting from version 9.0, this feature is deactivated and behind a feature flag. Set `xpack.monitoring.allow_legacy_collection` to `true` to allow access to the feature.")
+        end
+      end
+      private :log_warn_if_legacy_is_enabled_and_not_allowed
+
       def setup_metrics_pipeline
         settings = LogStash::SETTINGS.clone
 
@@ -195,7 +206,8 @@ def generate_pipeline_config(settings)
           raise ArgumentError.new("\"xpack.monitoring.enabled\" is configured while also \"monitoring.enabled\"")
         end
 
-        if any_set?(settings, /^xpack.monitoring/) && any_set?(settings, /^monitoring./)
+        if any_set?(settings, /^xpack.monitoring/, "xpack.monitoring.allow_legacy_collection") &&
+          any_set?(settings, /^monitoring./)
           raise ArgumentError.new("\"xpack.monitoring.*\" settings can't be configured while using \"monitoring.*\"")
         end
 
@@ -225,8 +237,8 @@ def retrieve_collection_settings(settings, prefix = "")
         opt
       end
 
-      def any_set?(settings, regexp)
-        !settings.get_subset(regexp).to_hash.keys.select { |k| settings.set?(k)}.empty?
+      def any_set?(settings, regexp, to_avoid = [])
+        !settings.get_subset(regexp).to_hash.keys.select{ |k| !to_avoid.include?(k)}.select { |k| settings.set?(k)}.empty?
       end
     end
 
@@ -259,6 +271,9 @@ def additionals_settings(settings)
 
     private
     def register_monitoring_settings(settings, prefix = "")
+      if prefix == "xpack."
+        settings.register(LogStash::Setting::Boolean.new("xpack.monitoring.allow_legacy_collection", false))
+      end
       settings.register(LogStash::Setting::Boolean.new("#{prefix}monitoring.enabled", false))
       settings.register(LogStash::Setting::ArrayCoercible.new("#{prefix}monitoring.elasticsearch.hosts", String, ["http://localhost:9200"]))
       settings.register(LogStash::Setting::TimeValue.new("#{prefix}monitoring.collection.interval", "10s"))
diff --git a/x-pack/qa/integration/management/multiple_pipelines_spec.rb b/x-pack/qa/integration/management/multiple_pipelines_spec.rb
index 0511da0bfac..a7b66de6f3a 100644
--- a/x-pack/qa/integration/management/multiple_pipelines_spec.rb
+++ b/x-pack/qa/integration/management/multiple_pipelines_spec.rb
@@ -33,6 +33,7 @@
         "xpack.management.elasticsearch.hosts" => ["http://localhost:9200"],
         "xpack.management.elasticsearch.username" => "elastic",
         "xpack.management.elasticsearch.password" => elastic_password,
+        "xpack.monitoring.allow_legacy_collection" => true,
         "xpack.monitoring.elasticsearch.username" => "elastic",
         "xpack.monitoring.elasticsearch.password" => elastic_password
 
diff --git a/x-pack/qa/integration/monitoring/direct_shipping_spec.rb b/x-pack/qa/integration/monitoring/direct_shipping_spec.rb
index bded6b36100..183f91906ed 100644
--- a/x-pack/qa/integration/monitoring/direct_shipping_spec.rb
+++ b/x-pack/qa/integration/monitoring/direct_shipping_spec.rb
@@ -15,6 +15,7 @@
 
     @logstash_service = logstash_with_empty_default("bin/logstash -e '#{config}' -w 1  --pipeline.id #{SecureRandom.hex(8)}", {
       :settings => {
+        "xpack.monitoring.allow_legacy_collection" => true,
         "monitoring.enabled" => true,
         "monitoring.elasticsearch.hosts" => ["http://localhost:9200", "http://localhost:9200"],
         "monitoring.collection.interval" => "1s",
diff --git a/x-pack/qa/integration/monitoring/es_documents_structure_validation_spec.rb b/x-pack/qa/integration/monitoring/es_documents_structure_validation_spec.rb
index 9b3c31057cb..f471153180a 100644
--- a/x-pack/qa/integration/monitoring/es_documents_structure_validation_spec.rb
+++ b/x-pack/qa/integration/monitoring/es_documents_structure_validation_spec.rb
@@ -85,6 +85,7 @@ def start_monitoring_logstash(config, prefix = "")
   end
   logstash_with_empty_default("bin/logstash -e '#{config}' -w 1", {
     :settings => {
+      "xpack.monitoring.allow_legacy_collection" => true,
       "#{mon_prefix}monitoring.enabled" => true,
       "#{mon_prefix}monitoring.elasticsearch.hosts" => ["http://localhost:9200", "http://localhost:9200"],
       "#{mon_prefix}monitoring.collection.interval" => "1s",
diff --git a/x-pack/qa/integration/monitoring/multiple_host_defined_spec.rb b/x-pack/qa/integration/monitoring/multiple_host_defined_spec.rb
index 235c92de4e3..67e7d4f652f 100644
--- a/x-pack/qa/integration/monitoring/multiple_host_defined_spec.rb
+++ b/x-pack/qa/integration/monitoring/multiple_host_defined_spec.rb
@@ -14,6 +14,7 @@
 
     @logstash_service = logstash("bin/logstash -e '#{config}' -w 1", {
       :settings => {
+        "xpack.monitoring.allow_legacy_collection" => true,
         "xpack.monitoring.elasticsearch.hosts" => ["http://localhost:9200", "http://localhost:9200"],
         "xpack.monitoring.collection.interval" => "1s",
         "xpack.monitoring.elasticsearch.username" => "elastic",
diff --git a/x-pack/qa/integration/monitoring/no_ssl_create_monitoring_indexes_spec.rb b/x-pack/qa/integration/monitoring/no_ssl_create_monitoring_indexes_spec.rb
index 0213d675a63..17b494bfaf0 100644
--- a/x-pack/qa/integration/monitoring/no_ssl_create_monitoring_indexes_spec.rb
+++ b/x-pack/qa/integration/monitoring/no_ssl_create_monitoring_indexes_spec.rb
@@ -16,6 +16,7 @@
 
     @logstash_service = logstash("bin/logstash -e '#{config}' -w 1", {
       :settings => {
+        "xpack.monitoring.allow_legacy_collection" => true,
         "xpack.monitoring.collection.interval" => "1s",
         "xpack.monitoring.elasticsearch.username" => "elastic",
         "xpack.monitoring.elasticsearch.password" => elastic_password
diff --git a/x-pack/qa/integration/monitoring/persisted_queue_is_enabled_spec.rb b/x-pack/qa/integration/monitoring/persisted_queue_is_enabled_spec.rb
index 1ed1ca84614..126d5efccc6 100644
--- a/x-pack/qa/integration/monitoring/persisted_queue_is_enabled_spec.rb
+++ b/x-pack/qa/integration/monitoring/persisted_queue_is_enabled_spec.rb
@@ -14,6 +14,7 @@
 
     @logstash_service = logstash("bin/logstash -e '#{config}' -w 1", {
       :settings => {
+        "xpack.monitoring.allow_legacy_collection" => true,
         "xpack.monitoring.elasticsearch.hosts" => ["http://localhost:9200", "http://localhost:9200"],
         "xpack.monitoring.collection.interval" => "1s",
         "queue.type" => "persisted",
diff --git a/x-pack/spec/monitoring/pipeline_register_hook_spec.rb b/x-pack/spec/monitoring/pipeline_register_hook_spec.rb
index 98723413e81..e2074baaeb0 100644
--- a/x-pack/spec/monitoring/pipeline_register_hook_spec.rb
+++ b/x-pack/spec/monitoring/pipeline_register_hook_spec.rb
@@ -24,6 +24,18 @@
   }
 
   context 'validate monitoring settings' do
+    it "it's not enabled with just xpack.monitoring.enabled set to true" do
+      expect(subject.monitoring_enabled?(settings)).to be_falsey
+    end
+
+    context 'when legacy monitoring is allowed and xpack monitoring is enabled' do
+      let(:settings) { super().merge({"xpack.monitoring.allow_legacy_collection" => true}) }
+
+      it "then internal monitoring should be effectively enabled" do
+        expect(subject.monitoring_enabled?(settings)).to be_truthy
+      end
+    end
+
     it "work without any monitoring settings" do
       settings.set_value("xpack.monitoring.enabled", true)
       expect(subject.generate_pipeline_config(settings)).to be_truthy