diff --git a/plugins/inputs/prometheus/README.md b/plugins/inputs/prometheus/README.md
index 5d6dfd69b..d77ea8cad 100644
--- a/plugins/inputs/prometheus/README.md
+++ b/plugins/inputs/prometheus/README.md
@@ -83,6 +83,11 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## Default is 60 seconds.
   # pod_scrape_interval = 60
 
+  ## Content length limit
+  ## When set, telegraf will drop responses with length larger than the configured value.
+  ## Default is "0KB" which means unlimited.
+  # content_length_limit = "0KB"
+
   ## Restricts Kubernetes monitoring to a single namespace
   ##   ex: monitor_kubernetes_pods_namespace = "default"
   # monitor_kubernetes_pods_namespace = ""
diff --git a/plugins/inputs/prometheus/prometheus.go b/plugins/inputs/prometheus/prometheus.go
index d82eeb407..5e2186d81 100644
--- a/plugins/inputs/prometheus/prometheus.go
+++ b/plugins/inputs/prometheus/prometheus.go
@@ -75,7 +75,8 @@ type Prometheus struct {
 
 	HTTPHeaders map[string]string `toml:"http_headers"`
 
-	ResponseTimeout config.Duration `toml:"response_timeout" deprecated:"1.26.0;use 'timeout' instead"`
+	ResponseTimeout    config.Duration `toml:"response_timeout" deprecated:"1.26.0;use 'timeout' instead"`
+	ContentLengthLimit config.Size     `toml:"content_length_limit"`
 
 	MetricVersion int `toml:"metric_version"`
 
@@ -437,9 +438,28 @@ func (p *Prometheus) gatherURL(u URLAndAddress, acc telegraf.Accumulator) error
 		return fmt.Errorf("%q returned HTTP status %q", u.URL, resp.Status)
 	}
 
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("error reading body: %w", err)
+	var body []byte
+	if p.ContentLengthLimit != 0 {
+		limit := int64(p.ContentLengthLimit)
+
+		// To determine whether io.ReadAll() ended due to EOF or reached the specified limit,
+		// read up to the specified limit plus one extra byte, and then make a decision based
+		// on the length of the result.
+		lr := io.LimitReader(resp.Body, limit+1)
+
+		body, err = io.ReadAll(lr)
+		if err != nil {
+			return fmt.Errorf("error reading body: %w", err)
+		}
+		if int64(len(body)) > limit {
+			p.Log.Infof("skipping %s: content length exceeded maximum body size (%d)", u.URL, limit)
+			return nil
+		}
+	} else {
+		body, err = io.ReadAll(resp.Body)
+		if err != nil {
+			return fmt.Errorf("error reading body: %w", err)
+		}
 	}
 
 	// Parse the metrics
diff --git a/plugins/inputs/prometheus/prometheus_test.go b/plugins/inputs/prometheus/prometheus_test.go
index 1631f2692..0ecfa621a 100644
--- a/plugins/inputs/prometheus/prometheus_test.go
+++ b/plugins/inputs/prometheus/prometheus_test.go
@@ -342,6 +342,26 @@ func TestPrometheusGeneratesMetricsSlowEndpointHitTheTimeoutNewConfigParameter(t
 	require.ErrorContains(t, err, "error making HTTP request to \""+ts.URL+"/metrics\"")
 }
 
+func TestPrometheusContentLengthLimit(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, err := fmt.Fprintln(w, sampleTextFormat)
+		require.NoError(t, err)
+	}))
+	defer ts.Close()
+
+	p := &Prometheus{
+		Log:                testutil.Logger{},
+		URLs:               []string{ts.URL},
+		URLTag:             "url",
+		ContentLengthLimit: 1,
+	}
+	require.NoError(t, p.Init())
+
+	var acc testutil.Accumulator
+	require.NoError(t, acc.GatherError(p.Gather))
+	require.Empty(t, acc.Metrics)
+}
+
 func TestPrometheusGeneratesSummaryMetricsV2(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		_, err := fmt.Fprintln(w, sampleSummaryTextFormat)
diff --git a/plugins/inputs/prometheus/sample.conf b/plugins/inputs/prometheus/sample.conf
index b57a15589..51d853b28 100644
--- a/plugins/inputs/prometheus/sample.conf
+++ b/plugins/inputs/prometheus/sample.conf
@@ -66,6 +66,11 @@
   ## Default is 60 seconds.
   # pod_scrape_interval = 60
 
+  ## Content length limit
+  ## When set, telegraf will drop responses with length larger than the configured value.
+  ## Default is "0KB" which means unlimited.
+  # content_length_limit = "0KB"
+
   ## Restricts Kubernetes monitoring to a single namespace
   ##   ex: monitor_kubernetes_pods_namespace = "default"
   # monitor_kubernetes_pods_namespace = ""