feat(common.socket): Allow parallel parsing with a pool of workers

[LarsStegman]

Summary

This is a proof of concept for the issue I raised in #15884. I feel like there might be a better way to implement this, than my way. I was thinking of something like the Options pattern, but I am not sure about this.

I am very open to discuss this! @srebhan do you have time to discuss this? I am not even sure if it is something you're open to add to Telegraf, but I think it is worth it considering the increased performance we see in our application.

Checklist

• No AI generated code was used in this PR

Related issues

resolves #15884

[srebhan]

Hey! Thanks for being so active and contributing your findings and fixes! We recently added something similar to the kafka consumer plugin allowing to override the timestamp provided by parsing.
I see two options:

1. Implementing the override in the input plugin using metric.SetTime() to overwrite the timestamp after parsing.
2. Implementing the override as a general option for inputs in the running_input model.

Option two was discussed a while ago but we never found the time to implement it. Let me know if you are interested in this and I will detail more on what needs to be done.

[LarsStegman]

You're welcome, it helps me too, since I now only need to maintain some proprietary plugins. Maintaining a fork that touches a lot of the internals is very time consuming. And most importantly, I enjoy contributing. I am also glad that my company allows me to.

I'm up to working on it. I wasn't really aware of how the internals for Telegraf worked, since I worked mostly on plugins. This weekend of profiling and optimizing really made things a lot clearer. I think I see how it would be implemented on the running_input, but please elaborate.

Regarding moving parsing for inputs.socket_listener to a separate goroutine. Should I make that a separate PR after the override has been added to the running_input? Is that something that stands a chance of getting merged?

[srebhan]

I think I see how it would be implemented on the running_input, but please elaborate.

The idea is to add a general timesource option with the choices

• metric: use the timestamp of the metric as provided by the input plugin e.g. through parsing (default)
• collection-start: use the timestamp when gathering started for all metrics
• collection-end: use the timestamp when gathering finished for all metrics

to all inputs by adding a new config option (and parse it) to the inputs.

Then you adapt the metric using the start time (stored in the model struct) or end time (stored in the model struct) and call SetTime() on all metrics, or do nothing dependent on the config option.

Does this make things more clear?

Regarding moving parsing for inputs.socket_listener to a separate goroutine. Should I make that a separate PR after the override has been added to the running_input? Is that something that stands a chance of getting merged?

I think we might accept this if there is a compelling argumentation why this is needed and what the benefit will be. I guess you are already having those numbers as you are asking... ;-)
One thing, please use a thread pool like github.com/alitto/pond (see e.g. the snmp_lookup processor) to allow the user to keep control over the number of threads spawned as otherwise it might be easily possible to DoS machines...

[LarsStegman]

Yes, that makes it clear and is about what I expected. I'll try to work on it tomorrow!

I'll also work on the async parsing.

[srebhan]

Just to be clear, async parsing should be an own PR! Please also double check that parser calls are thread-safe! I remember that we had this a while back with Avro but I want to urge you to double check! :-)

[LarsStegman]

Yes, that was my plan as well :) Easier to get things merged when the changes are small and isolated.

I thought the "contract" requires calls to Parse to be thread safe, right?

[LarsStegman]

The idea is to add a general timesource option with the choices

• metric: use the timestamp of the metric as provided by the input plugin e.g. through parsing (default)
• collection-start: use the timestamp when gathering started for all metrics
• collection-end: use the timestamp when gathering finished for all metrics

I am working on implementing it, but I am a little uncertain how this would work for Service plugins. Those often don't do anything in Gather, so overriding it with a collection-start/end time is useless.

I was thinking of maybe adding the same feature to the RunningParser, but even then there is no guarantee this is the correct time. The time when the Parse function is called, does not need to be the same as the time the original data comes in. For example, I read the UDP buffer, and then send the data into a goroutine to be parsed. The goroutine is not guaranteed to be started immediately.

[srebhan]

@LarsStegman right, service plugins will need to take care of timestamp overriding themselves...

[srebhan]

Thanks @LarsStegman for your contribution! A few comments from my side.

[LarsStegman]

I answered some questions and gave some explanations. I have not yet implemented any changes, I will do that tomorrow.

[srebhan]

Please move the time-source part to another PR and keep the parallelization in here!

[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]

@LarsStegman one more question because we had issues in the past... Did you test if taking down telegraf still works reliably with parallel workers?

[LarsStegman]

With the first implementation with just creating new goroutines, it did not. I am not 100% sure with the pool implementation. I don't remember seeing issues, but I will test it tomorrow. I can imagine some things breaking, because the channel is already closed when parsing is finished or something. I should probably add some wait group things or something.

Edit: looking at the code again, there shouldn't be any problems, because I already wait until the pool is done. I will test it tomorrow to be sure, though.

[LarsStegman]

@srebhan yeah, no problems. It stops gracefully.

[srebhan]

Thanks for all your effort @LarsStegman!