feat: add processors.http plugin for enrichment

[daniel-falk]

Summary

This PR adds a processor for HTTP enrichment. It takes a metric as input, makes an HTTP request and merges the response with the original metric before emitting it. This is great for two use cases:

• Enrich metrics with data from an API, e.g. given an IP address, enrich with a field for the geographic location
• Make HTTP requests using dynamic request data, e.g. an IP address that is produced from another plugin

Design rationale

The plugin is essentially a combination of inputs.http and outputs.http.

I have reused as much functionality as possible and kept the style and terminology from the other plugins.

For functionality like merging, this is done in the same style as in the processors.parser plugin.

Even if I don't like it, I kept the enforcement of same input and output data_format as was done in the processors.execd plugin. I think that is a more severe limitation in this plugin (and I'm not happy about it in the execd processor either), but I wanted to keep the PR as small as possible. Later, it would be good to add a PR that allows us to use e.g. the template serializer while selecting e.g. json or json_v2 as the output parser.

Error reporting with status_code and result tags is similar to input.http_response but adapted a bit for this use case. Since this is used for enrichment, I did not want to add any status fields on success, but it does add an http_error field on error. This keeps the enrichment clean. The two tags are added both on success and failure (unless `on_error = "drop") in which case there won't be any output if we didn't get any response.

I'm intentionally not using bufio.Scanner for the default response parsing to avoid the fixed size buffer limit it would produce (in some cases we might need to be able to parse very large responses). It is, however, used to parse error responses since I think it's safe to assume they are smaller than 64KB.

Example workflow

[agent]
  interval = "10s"
  flush_interval = "10s"
  debug = true

# Seed metric to trigger Starlark fan-out into per-IP metrics
[[inputs.mock]]
  metric_name = "trigger"
  [[inputs.mock.constant]]
    name = "seed"
    value = 1

# Expand trigger into one metric per IP address
[[processors.starlark]]
  namepass = ["trigger"]
  source = '''
def apply(metric):
    out = []
    for ip in ip_addresses:
        m = Metric("ip_lookup")
        m.fields["ip_address"] = ip
        m.tags["source"] = "starlark"
        m.time = metric.time
        out.append(m)
    return out
'''
  [processors.starlark.constants]
    ip_addresses = ["209.198.157.145", "8.8.8.8", "1.1.1.1"]

# POST each IP to ip-api.com/batch (one-element batch per metric)
[[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": "<ip_address>"}]
  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"

[[outputs.file]]
  files = ["stdout"]
  data_format = "template"
  template = "{{ dict \"name\" .Name \"tags\" .Tags \"fields\" .Fields | toPrettyJson }}\n"

Running this outputs the following three metrics:

{
  "fields": {
    "as": "AS14593 Space Exploration Technologies Corporation",
    "city": "Oslo",
    "country": "Norway",
    "countryCode": "NO",
    "ip_address": "209.198.157.145",
    "isp": "Space Exploration Technologies Corporation",
    "lat": 59.9138,
    "lon": 10.7522,
    "org": "Starlink Frntdeu1",
    "query": "209.198.157.145",
    "region": "03",
    "regionName": "Oslo County",
    "status": "success",
    "zip": ""
  },
  "name": "ip_lookup",
  "tags": {
    "result": "success",
    "source": "starlark",
    "status_code": "200"
  }
}
{
  "fields": {
    "as": "AS15169 Google LLC",
    "city": "Ashburn",
    "country": "United States",
    "countryCode": "US",
    "ip_address": "8.8.8.8",
    "isp": "Google LLC",
    "lat": 39.03,
    "lon": -77.5,
    "org": "Google Public DNS",
    "query": "8.8.8.8",
    "region": "VA",
    "regionName": "Virginia",
    "status": "success",
    "zip": "20149"
  },
  "name": "ip_lookup",
  "tags": {
    "result": "success",
    "source": "starlark",
    "status_code": "200"
  }
}
{
  "fields": {
    "as": "AS13335 Cloudflare, Inc.",
    "city": "South Brisbane",
    "country": "Australia",
    "countryCode": "AU",
    "ip_address": "1.1.1.1",
    "isp": "Cloudflare, Inc",
    "lat": -27.4766,
    "lon": 153.0166,
    "org": "APNIC and Cloudflare DNS Resolver project",
    "query": "1.1.1.1",
    "region": "QLD",
    "regionName": "Queensland",
    "status": "success",
    "zip": "4101"
  },
  "name": "ip_lookup",
  "tags": {
    "result": "success",
    "source": "starlark",
    "status_code": "200"
  }
}

With drop_original = false we would also get all the unchanged input metrics:

{
  "fields": {
    "ip_address": "209.198.157.145"
  },
  "name": "ip_lookup",
  "tags": {
    "source": "starlark"
  }
}

Error example with invalid host and on_error = "keep":

{
  "fields": {
    "http_error": "Post \"http://i-do-not-exist.com\": dial tcp: lookup i-do-not-exist.com: no such host",
    "ip_address": "1.1.1.1"
  },
  "name": "ip_lookup",
  "tags": {
    "result": "dns_error",
    "source": "starlark"
  }
}

Error when the parsing fails (and on_error = "keep"):

{
  "fields": {
    "http_error": "parsing response failed: invalid character '\u003c' looking for beginning of value",
    "ip_address": "1.1.1.1"
  },
  "name": "ip_lookup",
  "tags": {
    "result": "body_read_error",
    "source": "starlark",
    "status_code": "200"
  }
}

Response code not in success_status_codes (and on_error = "keep"):

{
  "fields": {
    "http_error": "received status code 500 (Internal Server Error), expected any value out of [200]. body: \"There was an error processing the request.\"",
    "ip_address": "1.1.1.1"
  },
  "name": "ip_lookup",
  "tags": {
    "result": "response_status_code_mismatch",
    "source": "starlark",
    "status_code": "500"
  }
}

Checklist

• No AI generated code was used in this PR
• AI generated code used in this PR follows the InfluxData Policy on AI-Generated Code Contributions

TODO:

• Go through all unit tests and make sure they make sense and are easy to understand
• Test the parent merge behavior with multiple metrics in the response

Go isn't my native language, so it would be great if anyone skilled in Go can take a look at the code.

Ideas for future follow-up PRs

• Support different input and output data_format
• Support for GET
• high throughput optimizations
  • Response caching with TTL and cache keys
  • Thread pool for parallel requests
• Make the URL templatable

Later, I would want to be able to do like:

url = "https://{{ .Tag \"host\" }}/api/v1/device/{{ .Field \"device_id\" }}?threshold={{ .Field \"threshold_name\" }}"

or perhaps like this:

[[processors.http]]
  url = "http://{{ .Tag \"host\" }}/api/v1/device/{{ .Field \"device_id\" }}"
  method = "GET"

  data_format = "json" # Doesn't really matter for GET, the request has no body.

  [processors.http.query]
    threshold = '{{ .Field \"threshold_name\" }}'
    otherArg = '{{ .Field "text" | urlquery }}'

The templatable URL and query args probably also applies for the outputs.http plugin too where the same features would make sense (except from GET which would be a bit strange in an output even if I think there are use cases for it..)

Related issues

None: This is a new feature request.

[telegraf-tiger]

Thanks so much for the pull request!
🤝 ✒️ Just a reminder that the CLA has not yet been signed, and we'll need it before merging. Please sign the CLA when you get a chance, then post a comment here saying !signed-cla

[daniel-falk]

!signed-cla

[daniel-falk]

I'm a bit unsure about the AWS and GCP authentication options. I kept them just to stay as similar as possible with the input and output http plugins, but considering that there is no reuse in the code between the plugins (as I understand it, that's intentional?), it might be better to remove it from this PR and add it when someone actually need it?

[telegraf-tiger]

Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

☺️ This pull request doesn't significantly change the Telegraf binary size (less than 1%)

📦 Click here to get additional PR build artifacts

Artifact URLs

. DEB	. RPM	. TAR . GZ	. ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz

[srebhan]

First of all thank you very much for your contribution @daniel-falk! Much appreciated!

I do have some concerns with the general way this is heading to so let me try to understand what you are trying to achieve.

Enrich metrics with data from an API, e.g. given an IP address, enrich with a field for the geographic location

Wouldn't it be better to add a processor doing exactly this, adding IP details like the geo-location? Adding e.g. a processors.geolocation processor is both, less error prone and easier to use for people. You could make it support multiple services etc.

Make HTTP requests using dynamic request data, e.g. an IP address that is produced from another plugin

And this is my main concern: You are building a "general purpose make this API call" thing which tries to introduce coding into a no-code structure. Soon people need something combining multiple calls, e.g. to login first or to do a first API call to lookup some data used in another API call. This will the trigger issues like data-conversion, parsing etc and the plugin complexity will grow as you are already seeing when talking about the data-formats. Such complexity will not be maintainable in the long-run as it tends to creep into a "general interpreter language". :-)

This is also the reason we are rejecting "multi-API call" PRs for the http input plugin. If you need tailored stuff, write a plugin!

However, I do understand the need for "custom" general metric enrichment but then let's solve it in a general way. Currently this can be done via external plugins where the "algorithm" for enriching the metric lives outside of Telegraf.
Another way would be to add GRPC plugins (see #3813) or a plugin with an embedded "wasm runtime" (e.g. using wazero), where the user can write a custom enrichment himself and Telegraf allows to integrate that custom code.

So my suggestion is, if you want geo-location lookup, please write a processor plugin doing exactly this! If you want "general purpose processing of metrics" please come up with a specification or at least an issue describing a plugin allowing to embed custom code.

What do you think?

[daniel-falk]

@srebhan Thanks for your quick response!

I created issue #19250 to describe a way to embed custom interpreted code.

Generally, I do agree with you regarding the processor.geolocation plugin. If that was the only thing we needed, I would build a custom plugin for it. For us, this is just one of many things we need / are doing. Creating a dedicated plugin for it is probably nothing we will prioritize.

With the processor, I saw the potential to create something that is generic enough to kill multiple birds with one stone, but simple enough to make the design and implementation close to trivial.

A generic script runner with I/O support like #19250 would be great for solving bespoke and complex needs, but for many things I think it is overkill. The configuration in the TOML would be much more straight forward and simple with this processors.http plugin than it will be with a generic script runner plugin.

Regarding the processors.execd plugin, that is what we are using now but for us it's really painful. We are running telegraf on small embedded devices. The only interpreted language that is installed is /bin/sh. Creating and maintaining scripts in that is terrible, especially when we need to parse and serialize metrics even for the simplest scripts.

I also do not like that it depends on things being installed in the external system. It makes versioning and portability complex.

GRPC is way to much implementation overhead for this.

The ability to load external plugins from .so files (#1717) is a fantastic addition that I was not aware of. We will definitely start to use that. But it does not resolve this issue since we would still need to build and compile plugins for the right environment and bundle with Telegraf. The same problem would apply to web assembly, I like the idea since it supports multiple source languages and the compiled files are portable accross architectures, but we need something that can be used without compiling it. We just want a simple TOML configuration block to add the request to the workflow.