diff --git a/plugins/processors/all/http.go b/plugins/processors/all/http.go new file mode 100644 index 0000000000000..3b8e7c08ef9bd --- /dev/null +++ b/plugins/processors/all/http.go @@ -0,0 +1,5 @@ +//go:build !custom || processors || processors.http + +package all + +import _ "github.com/influxdata/telegraf/plugins/processors/http" // register plugin diff --git a/plugins/processors/http/README.md b/plugins/processors/http/README.md new file mode 100644 index 0000000000000..1132e81e6556a --- /dev/null +++ b/plugins/processors/http/README.md @@ -0,0 +1,241 @@ +# HTTP Processor Plugin + +This plugin sends each incoming metric to an HTTP endpoint, parses the +response using the configured parser, and emits transformed metrics. It is +useful for enrichment and transformation workflows that require external HTTP +services or to call HTTP APIs where the request data is derived from the incoming +metrics. + +⭐ Telegraf v0.0.0 +🏷️ transformation +💻 all + +## Global configuration options + +Plugins support additional global and plugin configuration settings for tasks +such as modifying metrics, tags, and fields, creating aliases, and configuring +plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details. + +[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins + +## Configuration + +```toml @sample.conf +# Enrich or replace metrics by sending them to an HTTP endpoint +[[processors.http]] + ## Required HTTP endpoint to call for each metric + url = "https://api.example.com/enrich" + + ## HTTP method, one of: "POST", "PUT", or "PATCH" + # method = "POST" + + ## If true, incoming metrics are not emitted. + # drop_original = false + + ## Merge Behavior + ## Possible options are: + ## - none: keep the newly parsed metrics as-is + ## - override: emit a single metric with all tags and fields of newly parsed + ## merged but retaining the first timestamp. If drop_original is + ## false, all metrics are merged into the original metric + ## NOTE: Existing field or tag values will be overridden. + ## - override-with-timestamp: same as "override", but the timestamp is set + ## based on the new metrics if present + ## - parent: emit one metric per newly parsed metric with each newly parsed + ## metric is merged individually into the parent metric keeping the parent + ## timestamp + ## - parent-with-timestamp: same as "parent", but the timestamp is set + ## based on the new metric if present + # merge = "none" + + ## What to do when the HTTP request or response parsing fails. + ## - keep: pass the original metric through; adds status_code tag and + ## http_error field when an HTTP response was received (non-success status + ## or response parsing failure), or http_error only when no response + ## - drop: discard the metric + # on_error = "keep" + + ## List of acceptable HTTP response status codes + # success_status_codes = [200] + + ## HTTP Content-Encoding for the request body, can be set to "gzip" to + ## compress body or "identity" to apply no encoding. + # content_encoding = "identity" + + ## HTTP Basic Auth credentials + # username = "username" + # password = "pa$$word" + + ## Google API Auth + # google_application_credentials = "/etc/telegraf/example_secret.json" + + ## Amount of time allowed to complete the HTTP request + # timeout = "5s" + + ## HTTP connection settings + # idle_conn_timeout = "0s" + # max_idle_conn = 0 + # max_idle_conn_per_host = 0 + # response_timeout = "0s" + + ## Use the local address for connecting, assigned by the OS by default + # local_address = "" + + ## Optional proxy settings + # use_system_proxy = false + # http_proxy_url = "" + + ## Optional TLS settings + ## Set to true/false to enforce TLS being enabled/disabled. If not set, + ## enable TLS only if any of the other options are specified. + # tls_enable = + ## Trusted root certificates for server + # tls_ca = "/path/to/cafile" + ## Used for TLS client certificate authentication + # tls_cert = "/path/to/certfile" + ## Used for TLS client certificate authentication + # tls_key = "/path/to/keyfile" + ## Password for the key file if it is encrypted + # tls_key_pwd = "" + ## Send the specified TLS server name via SNI + # tls_server_name = "kubernetes.example.com" + ## Minimal TLS version to accept by the client + # tls_min_version = "TLS12" + ## List of ciphers to accept, by default all secure ciphers will be accepted + ## See https://pkg.go.dev/crypto/tls#pkg-constants for supported values. + ## Use "all", "secure" and "insecure" to add all support ciphers, secure + ## suites or insecure suites respectively. + # tls_cipher_suites = ["secure"] + ## Renegotiation method, "never", "once" or "freely" + # tls_renegotiation_method = "never" + ## Use TLS but skip chain & host verification + # insecure_skip_verify = false + + ## OAuth2 Client Credentials. The options 'client_id', 'client_secret', and 'token_url' are required to use OAuth2. + # client_id = "clientid" + # client_secret = "secret" + # token_url = "https://indentityprovider/oauth2/v1/token" + # audience = "" + # scopes = ["urn:opc:idm:__myscopes__"] + + ## Optional Cookie authentication + # cookie_auth_url = "https://localhost/authMe" + # cookie_auth_method = "POST" + # cookie_auth_username = "username" + # cookie_auth_password = "pa$$word" + # cookie_auth_headers = { Content-Type = "application/json", X-MY-HEADER = "hello" } + # cookie_auth_body = '{"username": "user", "password": "pa$$word", "authenticate": "me"}' + ## cookie_auth_renewal not set or set to "0" will auth once and never renew the cookie + # cookie_auth_renewal = "0s" + + ## Amazon Region + #region = "us-east-1" + + ## Amazon Credentials + ## Amazon Credentials are not built unless the following aws_service + ## setting is set to a non-empty string. + #aws_service = "execute-api" + + ## Credentials are loaded in the following order + ## 1) Web identity provider credentials via STS if role_arn and web_identity_token_file are specified + ## 2) Assumed credentials via STS if role_arn is specified + ## 3) explicit credentials from 'access_key' and 'secret_key' + ## 4) shared profile from 'profile' + ## 5) environment variables + ## 6) shared credentials file + ## 7) EC2 Instance Profile + #access_key = "" + #secret_key = "" + #token = "" + #role_arn = "" + #web_identity_token_file = "" + #role_session_name = "" + #profile = "" + #shared_credential_file = "" + + ## Data format used for both request serialization and response parsing. + ## The format must support both serialization and parsing (e.g. "influx", "json"). + ## Each data format has its own unique set of configuration options, read + ## more about them here: + ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md + data_format = "json" + + ## NOTE: Due to the way TOML is parsed, tables must be at the END of the + ## plugin definition, otherwise additional config options are read as part of + ## the table + + ## Additional HTTP headers + # [processors.http.headers] + # Content-Type = "application/json" + # Authorization = "Bearer ..." +``` + +### Data format requirements + +The configured `data_format` must exist as both a Telegraf serializer and +parser because the processor uses the same format for the request body and +response body. + +Examples: + +- Supported: `influx`, `json`, and other formats available in both parser and + serializer registries. +- Rejected: serializer-only or parser-only formats. + +## HTTP metadata + +The processor annotates metrics with request status information where applicable: + +- tags: + - `status_code` (response status code) + - `result` ([see below](#result)) +- fields: + - `http_error` (string, set only on failure when `on_error = "keep"`) + +On success, emitted metrics include `status_code` and `result=success`. + +On failure with `on_error = "keep"`, the original metric is passed through with +`http_error` and `result`. The `status_code` tag is set when an HTTP response +was received. + +### `result` + +Upon finishing the HTTP request for each metric, the plugin sets the `result` +tag to describe the outcome: + +- `success`: acceptable status code and response parsed successfully +- `processing_error`: request serialization or parser instantiation failed +- `body_read_error`: response body could not be parsed, or the parser returned + no metrics +- `connection_failed`: network error not otherwise classified below +- `timeout`: request timed out +- `dns_error`: DNS lookup failed +- `response_status_code_mismatch`: status code not listed in + `success_status_codes` + +## Example + +```toml +[[processors.http]] + namepass = ["ip_lookup"] + url = "http://ip-api.com/batch" + method = "POST" + data_format = "json" + merge = "parent" + drop_original = true + on_error = "keep" + timeout = "15s" + + # Reshape Telegraf JSON message to API request body... + # ip-api batch format: [{"query": ""}] + json_transformation = '[{"query": fields.ip_address}]' + + # List the expected string fields in the API response. + json_string_fields = [ + "query", "status", "country", "countryCode", "region", + "regionName", "city", "zip", "isp", "org", "as", + ] + + [processors.http.headers] + Content-Type = "application/json" +``` diff --git a/plugins/processors/http/http.go b/plugins/processors/http/http.go new file mode 100644 index 0000000000000..145b249ab7f78 --- /dev/null +++ b/plugins/processors/http/http.go @@ -0,0 +1,529 @@ +//go:generate ../../../tools/config_includer/generator +//go:generate ../../../tools/readme_config_includer/generator +package http + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + _ "embed" + "encoding/hex" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + aws_signer "github.com/aws/aws-sdk-go-v2/aws/signer/v4" + "golang.org/x/oauth2" + "google.golang.org/api/idtoken" + + "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/internal" + "github.com/influxdata/telegraf/models" + common_aws "github.com/influxdata/telegraf/plugins/common/aws" + common_gcp "github.com/influxdata/telegraf/plugins/common/gcp" + common_http "github.com/influxdata/telegraf/plugins/common/http" + "github.com/influxdata/telegraf/plugins/processors" +) + +//go:embed sample.conf +var sampleConfig string + +const ( + defaultContentType = "text/plain; charset=utf-8" + defaultMethod = http.MethodPost + defaultOnError = "keep" + tagStatusCode = "status_code" + tagResult = "result" + fieldHTTPError = "http_error" +) + +type HTTP struct { + URL string `toml:"url"` + Method string `toml:"method"` + DropOriginal bool `toml:"drop_original"` + Merge string `toml:"merge"` + OnError string `toml:"on_error"` + ContentEncoding string `toml:"content_encoding"` + + Username config.Secret `toml:"username"` + Password config.Secret `toml:"password"` + Headers map[string]*config.Secret `toml:"headers"` + + SuccessStatusCodes []int `toml:"success_status_codes"` + + AwsService string `toml:"aws_service"` + CredentialsFile string `toml:"google_application_credentials"` + + common_http.HTTPClientConfig + common_aws.CredentialConfig + + Log telegraf.Logger `toml:"-"` + + client *http.Client + serializer telegraf.Serializer + parserFunc telegraf.ParserFunc + awsCfg *aws.Config + oauth2Token *oauth2.Token +} + +func (*HTTP) SampleConfig() string { + return sampleConfig +} + +func (h *HTTP) SetSerializer(serializer telegraf.Serializer) { + h.serializer = serializer +} + +func (h *HTTP) SetParserFunc(fn telegraf.ParserFunc) { + h.parserFunc = fn +} + +func (h *HTTP) Init() error { + if h.URL == "" { + return errors.New("url is required") + } + if h.serializer == nil { + return errors.New("serializer not configured") + } + if h.parserFunc == nil { + return errors.New("parser not configured") + } + + switch h.Merge { + case "": + h.Merge = "none" + case "none", "override", "parent": + case "override-with-timestamp", "parent-with-timestamp": + default: + return fmt.Errorf("unrecognized merge value: %s", h.Merge) + } + + switch h.OnError { + case "": + h.OnError = defaultOnError + case "keep", "drop": + default: + return fmt.Errorf("invalid on_error %q", h.OnError) + } + + if h.AwsService != "" { + cfg, err := h.CredentialConfig.Credentials() + if err == nil { + h.awsCfg = &cfg + } + } + + if h.Method == "" { + h.Method = defaultMethod + } + h.Method = strings.ToUpper(h.Method) + if h.Method == http.MethodGet { + return fmt.Errorf("invalid method [%s] %s: GET is not supported", h.URL, h.Method) + } + if h.Method != http.MethodPost && h.Method != http.MethodPut && h.Method != http.MethodPatch { + return fmt.Errorf("invalid method [%s] %s", h.URL, h.Method) + } + + if len(h.SuccessStatusCodes) == 0 { + h.SuccessStatusCodes = []int{http.StatusOK} + } + + ctx := context.Background() + client, err := h.HTTPClientConfig.CreateClient(ctx, h.Log) + if err != nil { + return err + } + h.client = client + + return nil +} + +type httpResult struct { + statusCode int + body []byte +} + +type requestError struct { + result string + resp *httpResult + err error +} + +func (e *requestError) Error() string { + return e.err.Error() +} + +func (e *requestError) Unwrap() error { + return e.err +} + +func requestErr(result string, resp *httpResult, err error) error { + return &requestError{ + result: result, + resp: resp, + err: err, + } +} + +func (h *HTTP) Apply(in ...telegraf.Metric) []telegraf.Metric { + results := make([]telegraf.Metric, 0, len(in)) + for _, metric := range in { + out, err := h.processMetric(metric) + if err != nil { + h.Log.Errorf("%v", err) + if h.OnError == "drop" { + metric.Drop() + continue + } + annotateErrorMetric(metric, err) + results = append(results, metric) + continue + } + results = append(results, out...) + } + return results +} + +func (h *HTTP) processMetric(metric telegraf.Metric) ([]telegraf.Metric, error) { + reqBody, err := h.serializer.Serialize(metric) + if err != nil { + return nil, requestErr("processing_error", nil, fmt.Errorf("serializing metric failed: %w", err)) + } + + resp, err := h.doRequest(reqBody) + if err != nil { + return nil, err + } + + parser, err := h.parserFunc() + if err != nil { + return nil, requestErr("processing_error", resp, fmt.Errorf("instantiating parser failed: %w", err)) + } + h.configureParser(parser) + + responseMetrics, err := parser.Parse(resp.body) + if err != nil { + return nil, requestErr("body_read_error", resp, fmt.Errorf("parsing response failed: %w", err)) + } + if len(responseMetrics) == 0 { + return nil, requestErr("body_read_error", resp, errors.New("parser returned no metrics")) + } + + for _, m := range responseMetrics { + if m.Name() == "" || m.Name() == "http" { + m.SetName(metric.Name()) + } + } + + var newMetrics []telegraf.Metric + if !h.DropOriginal { + newMetrics = append(newMetrics, metric) + } else { + metric.Drop() + } + newMetrics = append(newMetrics, responseMetrics...) + + var results []telegraf.Metric + switch h.Merge { + case "override": + results = []telegraf.Metric{mergeAll(newMetrics[0], newMetrics[1:], false)} + case "override-with-timestamp": + results = []telegraf.Metric{mergeAll(newMetrics[0], newMetrics[1:], true)} + case "parent": + results = mergeIndividual(metric, responseMetrics, false) + if !h.DropOriginal { + results = append([]telegraf.Metric{metric}, results...) + } + case "parent-with-timestamp": + results = mergeIndividual(metric, responseMetrics, true) + if !h.DropOriginal { + results = append([]telegraf.Metric{metric}, results...) + } + default: + results = newMetrics + } + + addSuccessMetadata(results, resp) + return results, nil +} + +func (h *HTTP) configureParser(parser telegraf.Parser) { + if h.Merge != "override-with-timestamp" && h.Merge != "parent-with-timestamp" { + return + } + + unwrapped := parser + if rp, ok := parser.(*models.RunningParser); ok { + unwrapped = rp.Parser + } + if ptfp, ok := unwrapped.(telegraf.ParserTimeFuncPlugin); ok { + ptfp.SetTimeFunc(func() time.Time { return time.Time{} }) + } else { + h.Log.Warnf("Parser will always create a timestamp in merge-mode %q!", h.Merge) + } +} + +func (h *HTTP) doRequest(reqBody []byte) (*httpResult, error) { + var reqBodyBuffer io.Reader = bytes.NewBuffer(reqBody) + + if h.ContentEncoding == "gzip" { + rc := internal.CompressWithGzip(reqBodyBuffer) + defer rc.Close() + reqBodyBuffer = rc + } + + var payloadHash *string + if h.awsCfg != nil { + buf := new(bytes.Buffer) + if _, err := io.Copy(buf, reqBodyBuffer); err != nil { + return nil, err + } + + sum := sha256.Sum256(buf.Bytes()) + reqBodyBuffer = buf + + hash := hex.EncodeToString(sum[:]) + payloadHash = &hash + } + + req, err := http.NewRequest(h.Method, h.URL, reqBodyBuffer) + if err != nil { + return nil, err + } + + if h.awsCfg != nil { + signer := aws_signer.NewSigner() + ctx := context.Background() + + credentials, err := h.awsCfg.Credentials.Retrieve(ctx) + if err != nil { + return nil, err + } + + if err := signer.SignHTTP(ctx, credentials, req, *payloadHash, h.AwsService, h.Region, time.Now().UTC()); err != nil { + return nil, err + } + } + + if !h.Username.Empty() || !h.Password.Empty() { + username, err := h.Username.Get() + if err != nil { + return nil, fmt.Errorf("getting username failed: %w", err) + } + password, err := h.Password.Get() + if err != nil { + username.Destroy() + return nil, fmt.Errorf("getting password failed: %w", err) + } + req.SetBasicAuth(username.String(), password.String()) + username.Destroy() + password.Destroy() + } + + if h.CredentialsFile != "" { + token, err := h.getAccessToken(context.Background(), h.URL) + if err != nil { + return nil, err + } + token.SetAuthHeader(req) + } + + req.Header.Set("User-Agent", internal.ProductToken()) + req.Header.Set("Content-Type", defaultContentType) + if h.ContentEncoding == "gzip" { + req.Header.Set("Content-Encoding", "gzip") + } + + for k, v := range h.Headers { + secret, err := v.Get() + if err != nil { + return nil, err + } + + headerVal := secret.String() + if strings.EqualFold(k, "host") { + req.Host = headerVal + } else { + req.Header.Set(k, headerVal) + } + + secret.Destroy() + } + + resp, err := h.client.Do(req) + if err != nil { + return nil, requestErr(classifyNetworkError(err), nil, err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, requestErr("body_read_error", &httpResult{statusCode: resp.StatusCode}, + fmt.Errorf("reading body failed: %w", err)) + } + + result := &httpResult{ + statusCode: resp.StatusCode, + body: body, + } + + responseHasSuccessCode := false + for _, statusCode := range h.SuccessStatusCodes { + if resp.StatusCode == statusCode { + responseHasSuccessCode = true + break + } + } + if !responseHasSuccessCode { + errorLine := firstLine(body) + return result, requestErr("response_status_code_mismatch", result, fmt.Errorf( + "received status code %d (%s), expected any value out of %v. body: %s", + resp.StatusCode, + http.StatusText(resp.StatusCode), + h.SuccessStatusCodes, + errorLine, + )) + } + + return result, nil +} + +func firstLine(body []byte) string { + scanner := bufio.NewScanner(bytes.NewReader(body)) + if scanner.Scan() { + return scanner.Text() + } + return "" +} + +func addSuccessMetadata(metrics []telegraf.Metric, resp *httpResult) { + for _, m := range metrics { + setResult(m, "success") + if resp != nil && resp.statusCode > 0 { + m.AddTag(tagStatusCode, strconv.Itoa(resp.statusCode)) + } + } +} + +func annotateErrorMetric(metric telegraf.Metric, err error) { + result := "connection_failed" + var resp *httpResult + + var re *requestError + if errors.As(err, &re) { + result = re.result + resp = re.resp + } else if networkResult := classifyNetworkError(err); networkResult != "" { + result = networkResult + } + + if resp != nil && resp.statusCode > 0 { + metric.AddTag(tagStatusCode, strconv.Itoa(resp.statusCode)) + } + metric.AddField(fieldHTTPError, err.Error()) + setResult(metric, result) +} + +func setResult(metric telegraf.Metric, result string) { + metric.AddTag(tagResult, result) +} + +func classifyNetworkError(err error) string { + var timeoutErr net.Error + if errors.As(err, &timeoutErr) && timeoutErr.Timeout() { + return "timeout" + } + + var urlErr *url.Error + if errors.As(err, &urlErr) { + var opErr *net.OpError + if errors.As(urlErr, &opErr) { + var dnsErr *net.DNSError + if errors.As(opErr, &dnsErr) { + return "dns_error" + } + } + } + + return "connection_failed" +} + +func mergeIndividual(base telegraf.Metric, metrics []telegraf.Metric, mergeTime bool) []telegraf.Metric { + result := make([]telegraf.Metric, 0, len(metrics)) + for _, metric := range metrics { + out := base.Copy() + for _, field := range metric.FieldList() { + out.AddField(field.Key, field.Value) + } + for _, tag := range metric.TagList() { + out.AddTag(tag.Key, tag.Value) + } + out.SetName(metric.Name()) + if mergeTime && !metric.Time().IsZero() { + out.SetTime(metric.Time()) + } + result = append(result, out) + } + + return result +} + +func mergeAll(base telegraf.Metric, metrics []telegraf.Metric, mergeTime bool) telegraf.Metric { + for _, metric := range metrics { + for _, field := range metric.FieldList() { + base.AddField(field.Key, field.Value) + } + for _, tag := range metric.TagList() { + base.AddTag(tag.Key, tag.Value) + } + base.SetName(metric.Name()) + + if mergeTime && !metric.Time().IsZero() { + base.SetTime(metric.Time()) + } + } + return base +} + +func (h *HTTP) getAccessToken(ctx context.Context, audience string) (*oauth2.Token, error) { + if h.oauth2Token.Valid() { + return h.oauth2Token, nil + } + + credType, err := common_gcp.ParseCredentialType(h.CredentialsFile) + if err != nil { + return nil, fmt.Errorf("unable to parse credentials file type: %w", err) + } + + ts, err := idtoken.NewTokenSource(ctx, audience, idtoken.WithAuthCredentialsFile(idtoken.CredentialsType(credType), h.CredentialsFile)) + if err != nil { + return nil, fmt.Errorf("error creating oauth2 token source: %w", err) + } + + token, err := ts.Token() + if err != nil { + return nil, fmt.Errorf("error fetching oauth2 token: %w", err) + } + + h.oauth2Token = token + + return token, nil +} + +func init() { + processors.Add("http", func() telegraf.Processor { + return &HTTP{ + Method: defaultMethod, + OnError: defaultOnError, + } + }) +} diff --git a/plugins/processors/http/http_test.go b/plugins/processors/http/http_test.go new file mode 100644 index 0000000000000..cf541c5caeb73 --- /dev/null +++ b/plugins/processors/http/http_test.go @@ -0,0 +1,709 @@ +package http_test + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/config" + "github.com/influxdata/telegraf/metric" + "github.com/influxdata/telegraf/plugins/parsers/influx" + "github.com/influxdata/telegraf/plugins/parsers/json" + "github.com/influxdata/telegraf/plugins/processors" + httpplugin "github.com/influxdata/telegraf/plugins/processors/http" + influxserializer "github.com/influxdata/telegraf/plugins/serializers/influx" + jsonserializer "github.com/influxdata/telegraf/plugins/serializers/json" + "github.com/influxdata/telegraf/testutil" +) + +const ( + metricName = "metricName" + simpleJSON = ` +{ + "a": 1.2 +} +` +) + +func getMetric() telegraf.Metric { + return metric.New( + "cpu", + map[string]string{}, + map[string]interface{}{ + "value": 42.0, + }, + time.Unix(0, 0), + ) +} + +func writeBody(t *testing.T, w http.ResponseWriter, body string) { + if _, err := w.Write([]byte(body)); err != nil { + t.Error(err) + } +} + +func newPlugin(t *testing.T, serverURL string, opts func(*httpplugin.HTTP)) *httpplugin.HTTP { + t.Helper() + + serializer := &jsonserializer.Serializer{} + require.NoError(t, serializer.Init()) + + plugin := &httpplugin.HTTP{ + URL: serverURL, + Merge: "override", + Log: testutil.Logger{}, + } + if opts != nil { + opts(plugin) + } + plugin.SetSerializer(serializer) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{MetricName: metricName} + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + return plugin +} + +func TestOverrideMergeJSONResponse(t *testing.T) { + var requestBody string + var requestMethod string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + return + } + requestBody = string(body) + requestMethod = r.Method + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Method = http.MethodPost + p.Merge = "override" + p.OnError = "keep" + }) + require.NoError(t, plugin.Init()) + + input := metric.New( + metricName, + map[string]string{}, + map[string]interface{}{"value": 42.0}, + time.Unix(0, 0), + ) + + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Contains(t, requestBody, `"value":42`) + require.Equal(t, http.MethodPost, requestMethod) + + got := results[0] + require.Equal(t, metricName, got.Name()) + require.Equal(t, time.Unix(0, 0), got.Time()) + require.InDelta(t, 42.0, got.Fields()["value"], testutil.DefaultDelta) + require.InDelta(t, 1.2, got.Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "200", got.Tags()["status_code"]) + require.Equal(t, "success", got.Tags()["result"]) + _, hasError := got.GetField("http_error") + require.False(t, hasError) +} + +func TestDropOriginalReturnsParsedMetrics(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.DropOriginal = true + p.Merge = "none" + }) + require.NoError(t, plugin.Init()) + + input := metric.New( + metricName, + map[string]string{"keep": "no"}, + map[string]interface{}{"value": 42.0}, + time.Unix(0, 0), + ) + + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Equal(t, metricName, results[0].Name()) + require.InDelta(t, 1.2, results[0].Fields()["a"], testutil.DefaultDelta) + _, hasTag := results[0].GetTag("keep") + require.False(t, hasTag) +} + +func TestParentMergeKeepsOriginal(t *testing.T) { + const responseJSON = `[{"a":1.2,"group":"one"},{"a":3.4,"group":"two"}]` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, responseJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Merge = "parent" + }) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{ + MetricName: metricName, + StringFields: []string{"group"}, + } + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + + input := metric.New( + metricName, + map[string]string{"source": "starlark"}, + map[string]interface{}{"value": 42.0}, + time.Unix(0, 0), + ) + + results := plugin.Apply(input) + require.Len(t, results, 3) + require.Equal(t, input, results[0]) + require.Equal(t, "starlark", results[0].Tags()["source"]) + require.InDelta(t, 42.0, results[0].Fields()["value"], testutil.DefaultDelta) + _, hasA := results[0].GetField("a") + require.False(t, hasA) + + require.InDelta(t, 1.2, results[1].Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "one", results[1].Fields()["group"]) + require.InDelta(t, 42.0, results[1].Fields()["value"], testutil.DefaultDelta) + require.Equal(t, "starlark", results[1].Tags()["source"]) + + require.InDelta(t, 3.4, results[2].Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "two", results[2].Fields()["group"]) +} + +func TestParentMergeDropsOriginal(t *testing.T) { + const responseJSON = `[{"a":1.2,"group":"one"},{"a":3.4,"group":"two"}]` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, responseJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Merge = "parent" + p.DropOriginal = true + }) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{ + MetricName: metricName, + StringFields: []string{"group"}, + } + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + + input := metric.New( + metricName, + map[string]string{"source": "starlark"}, + map[string]interface{}{"value": 42.0}, + time.Unix(0, 0), + ) + + results := plugin.Apply(input) + require.Len(t, results, 2) + require.InDelta(t, 1.2, results[0].Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "one", results[0].Fields()["group"]) + require.InDelta(t, 3.4, results[1].Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "two", results[1].Fields()["group"]) +} + +func TestOnErrorKeep(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + writeBody(t, w, "boom") + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + }) + + input := getMetric() + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Equal(t, input, results[0]) + require.Equal(t, "500", results[0].Tags()["status_code"]) + require.Equal(t, "response_status_code_mismatch", results[0].Tags()["result"]) + require.Contains(t, results[0].Fields()["http_error"], "received status code 500") +} + +func TestOnErrorDrop(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "drop" + }) + + results := plugin.Apply(getMetric()) + require.Empty(t, results) +} + +func TestRejectGETMethod(t *testing.T) { + plugin := &httpplugin.HTTP{ + URL: "http://example.com", + Method: http.MethodGet, + Log: testutil.Logger{}, + } + plugin.SetSerializer(&jsonserializer.Serializer{}) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + return &json.Parser{MetricName: metricName}, nil + }) + + err := plugin.Init() + require.Error(t, err) + require.Contains(t, err.Error(), "GET is not supported") +} + +func TestSuccessStatusCodes(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.SuccessStatusCodes = []int{201} + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.InDelta(t, 1.2, results[0].Fields()["a"], testutil.DefaultDelta) + require.Equal(t, "201", results[0].Tags()["status_code"]) + require.Equal(t, "success", results[0].Tags()["result"]) +} + +func TestOverrideOverwritesConflictingFieldsAndTags(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, `{"field":"new","env":"prod"}`) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Merge = "override" + }) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{ + MetricName: metricName, + StringFields: []string{"field", "env"}, + TagKeys: []string{"env"}, + } + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + + input := metric.New( + metricName, + map[string]string{"env": "dev"}, + map[string]interface{}{"field": "old"}, + time.Unix(0, 0), + ) + + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Equal(t, "new", results[0].Fields()["field"]) + require.Equal(t, "prod", results[0].Tags()["env"]) +} + +func TestConfigLoad(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + cfg := ` +[[processors.http]] + url = "` + server.URL + `" + data_format = "json" + merge = "override" +` + c := config.NewConfig() + require.NoError(t, c.LoadConfigData([]byte(cfg), "testdata.toml")) + require.Len(t, c.Processors, 1) + require.Len(t, c.AggProcessors, 1) + + for _, rp := range append(c.Processors, c.AggProcessors...) { + proc := rp.Processor.(processors.HasUnwrap) + unwrapped, ok := proc.Unwrap().(*httpplugin.HTTP) + require.True(t, ok) + require.Equal(t, server.URL, unwrapped.URL) + require.Equal(t, "override", unwrapped.Merge) + } +} + +func TestConfigRejectsTemplateDataFormat(t *testing.T) { + cfg := ` +[[processors.http]] + url = "http://example.com" + data_format = "template" + template = "x" +` + c := config.NewConfig() + err := c.LoadConfigData([]byte(cfg), "testdata.toml") + require.Error(t, err) + require.Contains(t, err.Error(), "parser not found") +} + +func TestPUTAndPATCHMethods(t *testing.T) { + methods := []string{http.MethodPut, http.MethodPatch} + for _, method := range methods { + t.Run(method, func(t *testing.T) { + var gotMethod string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Method = method + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, method, gotMethod) + }) + } +} + +func TestCustomHeaders(t *testing.T) { + const header = "X-Custom" + const value = "test-value" + var gotHeader string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHeader = r.Header.Get(header) + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + secret := config.NewSecret([]byte(value)) + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Headers = map[string]*config.Secret{header: &secret} + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, value, gotHeader) +} + +func TestBearerAuthViaHeaders(t *testing.T) { + const token = "secret-token" + var gotAuth string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + auth := config.NewSecret([]byte("Bearer " + token)) + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Headers = map[string]*config.Secret{"Authorization": &auth} + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, "Bearer "+token, gotAuth) +} + +func TestInvalidMerge(t *testing.T) { + plugin := &httpplugin.HTTP{ + URL: "http://example.com", + Merge: "append", + Log: testutil.Logger{}, + } + plugin.SetSerializer(&jsonserializer.Serializer{}) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + return &json.Parser{MetricName: metricName}, nil + }) + require.Error(t, plugin.Init()) +} + +func TestParseFailureOnErrorKeep(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, "not-json") + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + }) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{MetricName: metricName, Strict: true} + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + + input := getMetric() + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Equal(t, "cpu", results[0].Name()) + require.Equal(t, "200", results[0].Tags()["status_code"]) + require.Equal(t, "body_read_error", results[0].Tags()["result"]) + require.Contains(t, results[0].Fields()["http_error"], "parsing response failed") +} + +func TestOnErrorKeepEmptyParserResult(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, `[]`) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, "200", results[0].Tags()["status_code"]) + require.Equal(t, "body_read_error", results[0].Tags()["result"]) + require.Contains(t, results[0].Fields()["http_error"], "parser returned no metrics") +} + +func TestOnErrorKeepSerializationFailure(t *testing.T) { + plugin := &httpplugin.HTTP{ + URL: "http://example.com", + OnError: "keep", + Log: testutil.Logger{}, + } + plugin.SetSerializer(&failSerializer{}) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + return &json.Parser{MetricName: metricName}, nil + }) + require.NoError(t, plugin.Init()) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, "processing_error", results[0].Tags()["result"]) + require.Contains(t, results[0].Fields()["http_error"], "serializing metric failed") + _, hasStatus := results[0].GetTag("status_code") + require.False(t, hasStatus) +} + +func TestOnErrorKeepParserInstantiationFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + }) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + return nil, errors.New("parser init failed") + }) + require.NoError(t, plugin.Init()) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, "200", results[0].Tags()["status_code"]) + require.Equal(t, "processing_error", results[0].Tags()["result"]) + require.Contains(t, results[0].Fields()["http_error"], "instantiating parser failed") +} + +type failSerializer struct{} + +func (*failSerializer) Init() error { return nil } + +func (*failSerializer) Serialize(_ telegraf.Metric) ([]byte, error) { + return nil, errors.New("serialize failed") +} + +func (*failSerializer) SerializeBatch(_ []telegraf.Metric) ([]byte, error) { + return nil, errors.New("serialize failed") +} + +func TestOnErrorKeepTimeout(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { + time.Sleep(200 * time.Millisecond) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + p.Timeout = config.Duration(50 * time.Millisecond) + }) + + results := plugin.Apply(getMetric()) + require.Len(t, results, 1) + require.Equal(t, "timeout", results[0].Tags()["result"]) + _, hasStatus := results[0].GetTag("status_code") + require.False(t, hasStatus) +} + +func TestInfluxDataFormatRoundTrip(t *testing.T) { + var requestBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + return + } + requestBody = string(body) + writeBody(t, w, "metricName value=42 123\n") + })) + defer server.Close() + + serializer := &influxserializer.Serializer{} + require.NoError(t, serializer.Init()) + + plugin := &httpplugin.HTTP{ + URL: server.URL, + Merge: "override", + Log: testutil.Logger{}, + } + plugin.SetSerializer(serializer) + plugin.SetParserFunc(func() (telegraf.Parser, error) { + p := &influx.Parser{} + err := p.Init() + return p, err + }) + require.NoError(t, plugin.Init()) + + input := metric.New(metricName, nil, map[string]interface{}{"value": 42.0}, time.Unix(0, 0)) + results := plugin.Apply(input) + require.Len(t, results, 1) + require.Contains(t, requestBody, metricName) + require.InDelta(t, 42.0, results[0].Fields()["value"], testutil.DefaultDelta) +} + +func TestTrackingOnErrorKeep(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "keep" + }) + + var mu sync.Mutex + delivered := make([]telegraf.DeliveryInfo, 0, 1) + notify := func(di telegraf.DeliveryInfo) { + mu.Lock() + defer mu.Unlock() + delivered = append(delivered, di) + } + + input, _ := metric.WithTracking(getMetric(), notify) + results := plugin.Apply(input) + require.Len(t, results, 1) + results[0].Accept() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(delivered) == 1 && delivered[0].Delivered() + }, time.Second, 10*time.Millisecond) +} + +func TestTrackingOnErrorDrop(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.OnError = "drop" + }) + + var mu sync.Mutex + delivered := make([]telegraf.DeliveryInfo, 0, 1) + notify := func(di telegraf.DeliveryInfo) { + mu.Lock() + defer mu.Unlock() + delivered = append(delivered, di) + } + + input, _ := metric.WithTracking(getMetric(), notify) + results := plugin.Apply(input) + require.Empty(t, results) + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(delivered) == 1 + }, time.Second, 10*time.Millisecond) +} + +func TestTrackingOverrideMerge(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.Merge = "override" + }) + + var mu sync.Mutex + delivered := make([]telegraf.DeliveryInfo, 0, 1) + notify := func(di telegraf.DeliveryInfo) { + mu.Lock() + defer mu.Unlock() + delivered = append(delivered, di) + } + + input, _ := metric.WithTracking(getMetric(), notify) + results := plugin.Apply(input) + require.Len(t, results, 1) + results[0].Accept() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(delivered) == 1 && delivered[0].Delivered() + }, time.Second, 10*time.Millisecond) +} + +func TestTrackingDropOriginal(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + writeBody(t, w, simpleJSON) + })) + defer server.Close() + + plugin := newPlugin(t, server.URL, func(p *httpplugin.HTTP) { + p.DropOriginal = true + p.Merge = "none" + }) + + var mu sync.Mutex + delivered := make([]telegraf.DeliveryInfo, 0, 1) + notify := func(di telegraf.DeliveryInfo) { + mu.Lock() + defer mu.Unlock() + delivered = append(delivered, di) + } + + input, _ := metric.WithTracking(getMetric(), notify) + results := plugin.Apply(input) + require.Len(t, results, 1) + results[0].Accept() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(delivered) == 1 + }, time.Second, 10*time.Millisecond) +} diff --git a/plugins/processors/http/sample.conf b/plugins/processors/http/sample.conf new file mode 100644 index 0000000000000..6f34ee9361812 --- /dev/null +++ b/plugins/processors/http/sample.conf @@ -0,0 +1,147 @@ +# Enrich or replace metrics by sending them to an HTTP endpoint +[[processors.http]] + ## Required HTTP endpoint to call for each metric + url = "https://api.example.com/enrich" + + ## HTTP method, one of: "POST", "PUT", or "PATCH" + # method = "POST" + + ## If true, incoming metrics are not emitted. + # drop_original = false + + ## Merge Behavior + ## Possible options are: + ## - none: keep the newly parsed metrics as-is + ## - override: emit a single metric with all tags and fields of newly parsed + ## merged but retaining the first timestamp. If drop_original is + ## false, all metrics are merged into the original metric + ## NOTE: Existing field or tag values will be overridden. + ## - override-with-timestamp: same as "override", but the timestamp is set + ## based on the new metrics if present + ## - parent: emit one metric per newly parsed metric with each newly parsed + ## metric is merged individually into the parent metric keeping the parent + ## timestamp + ## - parent-with-timestamp: same as "parent", but the timestamp is set + ## based on the new metric if present + # merge = "none" + + ## What to do when the HTTP request or response parsing fails. + ## - keep: pass the original metric through; adds status_code tag and + ## http_error field when an HTTP response was received (non-success status + ## or response parsing failure), or http_error only when no response + ## - drop: discard the metric + # on_error = "keep" + + ## List of acceptable HTTP response status codes + # success_status_codes = [200] + + ## HTTP Content-Encoding for the request body, can be set to "gzip" to + ## compress body or "identity" to apply no encoding. + # content_encoding = "identity" + + ## HTTP Basic Auth credentials + # username = "username" + # password = "pa$$word" + + ## Google API Auth + # google_application_credentials = "/etc/telegraf/example_secret.json" + + ## Amount of time allowed to complete the HTTP request + # timeout = "5s" + + ## HTTP connection settings + # idle_conn_timeout = "0s" + # max_idle_conn = 0 + # max_idle_conn_per_host = 0 + # response_timeout = "0s" + + ## Use the local address for connecting, assigned by the OS by default + # local_address = "" + + ## Optional proxy settings + # use_system_proxy = false + # http_proxy_url = "" + + ## Optional TLS settings + ## Set to true/false to enforce TLS being enabled/disabled. If not set, + ## enable TLS only if any of the other options are specified. + # tls_enable = + ## Trusted root certificates for server + # tls_ca = "/path/to/cafile" + ## Used for TLS client certificate authentication + # tls_cert = "/path/to/certfile" + ## Used for TLS client certificate authentication + # tls_key = "/path/to/keyfile" + ## Password for the key file if it is encrypted + # tls_key_pwd = "" + ## Send the specified TLS server name via SNI + # tls_server_name = "kubernetes.example.com" + ## Minimal TLS version to accept by the client + # tls_min_version = "TLS12" + ## List of ciphers to accept, by default all secure ciphers will be accepted + ## See https://pkg.go.dev/crypto/tls#pkg-constants for supported values. + ## Use "all", "secure" and "insecure" to add all support ciphers, secure + ## suites or insecure suites respectively. + # tls_cipher_suites = ["secure"] + ## Renegotiation method, "never", "once" or "freely" + # tls_renegotiation_method = "never" + ## Use TLS but skip chain & host verification + # insecure_skip_verify = false + + ## OAuth2 Client Credentials. The options 'client_id', 'client_secret', and 'token_url' are required to use OAuth2. + # client_id = "clientid" + # client_secret = "secret" + # token_url = "https://indentityprovider/oauth2/v1/token" + # audience = "" + # scopes = ["urn:opc:idm:__myscopes__"] + + ## Optional Cookie authentication + # cookie_auth_url = "https://localhost/authMe" + # cookie_auth_method = "POST" + # cookie_auth_username = "username" + # cookie_auth_password = "pa$$word" + # cookie_auth_headers = { Content-Type = "application/json", X-MY-HEADER = "hello" } + # cookie_auth_body = '{"username": "user", "password": "pa$$word", "authenticate": "me"}' + ## cookie_auth_renewal not set or set to "0" will auth once and never renew the cookie + # cookie_auth_renewal = "0s" + + ## Amazon Region + #region = "us-east-1" + + ## Amazon Credentials + ## Amazon Credentials are not built unless the following aws_service + ## setting is set to a non-empty string. + #aws_service = "execute-api" + + ## Credentials are loaded in the following order + ## 1) Web identity provider credentials via STS if role_arn and web_identity_token_file are specified + ## 2) Assumed credentials via STS if role_arn is specified + ## 3) explicit credentials from 'access_key' and 'secret_key' + ## 4) shared profile from 'profile' + ## 5) environment variables + ## 6) shared credentials file + ## 7) EC2 Instance Profile + #access_key = "" + #secret_key = "" + #token = "" + #role_arn = "" + #web_identity_token_file = "" + #role_session_name = "" + #profile = "" + #shared_credential_file = "" + + ## Data format used for both request serialization and response parsing. + ## The format must support both serialization and parsing (e.g. "influx", "json"). + ## Each data format has its own unique set of configuration options, read + ## more about them here: + ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md + data_format = "json" + + ## NOTE: Due to the way TOML is parsed, tables must be at the END of the + ## plugin definition, otherwise additional config options are read as part of + ## the table + + ## Additional HTTP headers + # [processors.http.headers] + # Content-Type = "application/json" + # Authorization = "Bearer ..." diff --git a/plugins/processors/http/sample.conf.in b/plugins/processors/http/sample.conf.in new file mode 100644 index 0000000000000..5e34dad472f50 --- /dev/null +++ b/plugins/processors/http/sample.conf.in @@ -0,0 +1,90 @@ +# Enrich or replace metrics by sending them to an HTTP endpoint +[[processors.http]] + ## Required HTTP endpoint to call for each metric + url = "https://api.example.com/enrich" + + ## HTTP method, one of: "POST", "PUT", or "PATCH" + # method = "POST" + + ## If true, incoming metrics are not emitted. + # drop_original = false + + ## Merge Behavior + ## Possible options are: + ## - none: keep the newly parsed metrics as-is + ## - override: emit a single metric with all tags and fields of newly parsed + ## merged but retaining the first timestamp. If drop_original is + ## false, all metrics are merged into the original metric + ## NOTE: Existing field or tag values will be overridden. + ## - override-with-timestamp: same as "override", but the timestamp is set + ## based on the new metrics if present + ## - parent: emit one metric per newly parsed metric with each newly parsed + ## metric is merged individually into the parent metric keeping the parent + ## timestamp + ## - parent-with-timestamp: same as "parent", but the timestamp is set + ## based on the new metric if present + # merge = "none" + + ## What to do when the HTTP request or response parsing fails. + ## - keep: pass the original metric through; adds status_code tag and + ## http_error field when an HTTP response was received (non-success status + ## or response parsing failure), or http_error only when no response + ## - drop: discard the metric + # on_error = "keep" + + ## List of acceptable HTTP response status codes + # success_status_codes = [200] + + ## HTTP Content-Encoding for the request body, can be set to "gzip" to + ## compress body or "identity" to apply no encoding. + # content_encoding = "identity" + + ## HTTP Basic Auth credentials + # username = "username" + # password = "pa$$word" + + ## Google API Auth + # google_application_credentials = "/etc/telegraf/example_secret.json" + +{{template "/plugins/common/http/client.conf"}} + + ## Amazon Region + #region = "us-east-1" + + ## Amazon Credentials + ## Amazon Credentials are not built unless the following aws_service + ## setting is set to a non-empty string. + #aws_service = "execute-api" + + ## Credentials are loaded in the following order + ## 1) Web identity provider credentials via STS if role_arn and web_identity_token_file are specified + ## 2) Assumed credentials via STS if role_arn is specified + ## 3) explicit credentials from 'access_key' and 'secret_key' + ## 4) shared profile from 'profile' + ## 5) environment variables + ## 6) shared credentials file + ## 7) EC2 Instance Profile + #access_key = "" + #secret_key = "" + #token = "" + #role_arn = "" + #web_identity_token_file = "" + #role_session_name = "" + #profile = "" + #shared_credential_file = "" + + ## Data format used for both request serialization and response parsing. + ## The format must support both serialization and parsing (e.g. "influx", "json"). + ## Each data format has its own unique set of configuration options, read + ## more about them here: + ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md + data_format = "json" + + ## NOTE: Due to the way TOML is parsed, tables must be at the END of the + ## plugin definition, otherwise additional config options are read as part of + ## the table + + ## Additional HTTP headers + # [processors.http.headers] + # Content-Type = "application/json" + # Authorization = "Bearer ..."