feat(inputs.prometheus): Add option to limit body length (#14661)

Co-authored-by: Thomas Casteleyn <thomas.casteleyn@me.com>
Co-authored-by: Josh Powers <powersj@fastmail.com>
This commit is contained in:
Hiroshi Hayakawa 2024-02-02 21:25:09 +09:00 committed by GitHub
parent f9f2adf3ae
commit 534289f3b4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 54 additions and 4 deletions

View File

@ -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 = ""

View File

@ -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

View File

@ -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)

View File

@ -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 = ""