feat(common.socket): parallel parsing #15891

LarsStegman · 2024-09-16T07:34:45Z

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 · 2024-09-16T13:54:38Z

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:

Implementing the override in the input plugin using metric.SetTime() to overwrite the timestamp after parsing.
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 · 2024-09-17T13:00:38Z

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 · 2024-09-19T09:29:48Z

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 · 2024-09-19T09:46:27Z

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 · 2024-09-19T09:49:31Z

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 · 2024-09-19T09:50:49Z

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 · 2024-09-19T16:29:04Z

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 · 2024-09-19T20:10:34Z

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

plugins/common/socket/datagram.go

plugins/common/socket/socket.go

plugins/common/socket/stream.go

srebhan · 2024-09-30T15:36:40Z

plugins/inputs/socket_listener/socket_listener.go

-	onData := func(_ net.Addr, data []byte) {
+	onData := func(_ net.Addr, data []byte, receiveTime time.Time) {
 		metrics, err := sl.parser.Parse(data)
+
 		if err != nil {
 			acc.AddError(err)
 			return
 		}
+
 		if len(metrics) == 0 {
 			once.Do(func() {
 				sl.Log.Debug(internal.NoMetricsCreatedMsg)
 			})
 		}
+
 		for _, m := range metrics {
+			switch sl.TimeSource {
+			case "", "metric":
+			case "receive_time":
+				m.SetTime(receiveTime)
+			}
+


I think this should not be here... :-)

It should, since service plugins do not work with #15917

#15891 (comment)

Yeah but please make this an own PR to separate feature introduction!

LarsStegman

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

plugins/common/socket/datagram.go

plugins/common/socket/stream.go

LarsStegman · 2024-09-30T19:30:35Z

plugins/inputs/socket_listener/socket_listener.go

-	onData := func(_ net.Addr, data []byte) {
+	onData := func(_ net.Addr, data []byte, receiveTime time.Time) {
 		metrics, err := sl.parser.Parse(data)
+
 		if err != nil {
 			acc.AddError(err)
 			return
 		}
+
 		if len(metrics) == 0 {
 			once.Do(func() {
 				sl.Log.Debug(internal.NoMetricsCreatedMsg)
 			})
 		}
+
 		for _, m := range metrics {
+			switch sl.TimeSource {
+			case "", "metric":
+			case "receive_time":
+				m.SetTime(receiveTime)
+			}
+


It should, since service plugins do not work with #15917

#15891 (comment)

plugins/common/socket/socket.go

LarsStegman · 2024-09-30T19:36:20Z

plugins/inputs/socket_listener/socket_listener.go

@@ -20,6 +21,7 @@ var once sync.Once

 type SocketListener struct {
 	ServiceAddress string          `toml:"service_address"`
+	TimeSource     string          `toml:"time_source"`


Is it okay to use the same name as in #15917 ?

It might cause issues if the terms are not the same and checked somewhere. Either make sure there is no collision or (easier) use a different option.

telegraf-tiger · 2024-10-01T09:01:30Z

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 · 2024-10-02T06:48:16Z

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

srebhan marked this pull request as draft September 16, 2024 13:40

srebhan self-assigned this Sep 16, 2024

LarsStegman mentioned this pull request Sep 20, 2024

feat(inputs): Add option to choose the metric time source #15917

Open

1 task

LarsStegman force-pushed the feat/parser-with-timestamp branch from 8f7aa5d to 538a885 Compare September 20, 2024 12:43

LarsStegman changed the title ~~WIP: feat(parsers): support passing receive timestamp to parser~~ feat(inputs.socket_listener): parallel parsing Sep 20, 2024

telegraf-tiger bot added feat Improvement on an existing feature such as adding a new setting/mode to an existing plugin plugin/input 1. Request for new input plugins 2. Issues/PRs that are related to input plugins labels Sep 20, 2024

LarsStegman marked this pull request as ready for review September 20, 2024 12:47

LarsStegman force-pushed the feat/parser-with-timestamp branch 2 times, most recently from c41db77 to 9eaa313 Compare September 21, 2024 06:12

srebhan reviewed Sep 30, 2024

View reviewed changes

LarsStegman commented Sep 30, 2024

View reviewed changes

LarsStegman requested a review from srebhan October 1, 2024 08:35

LarsStegman added 7 commits October 3, 2024 08:46

feat(inputs.socket_listener): async parse

1be7c57

feat(inputs): fix format

c55d54b

feat(inputs): fix race condition

160f112

feat(common.socket): use decoder pool

5e07d86

feat(common.socket): linting

c56a76e

feat(common.socket): no magic max parallel parsers

711fd18

feat(common.socket): return decoder

6e92a4e

LarsStegman added 2 commits October 3, 2024 08:46

feat(common.socket): use supported verb

ba7464f

feat(common.socket): revert timestamping

a22481d

LarsStegman force-pushed the feat/parser-with-timestamp branch from bad5458 to a22481d Compare October 3, 2024 07:20

LarsStegman mentioned this pull request Oct 3, 2024

feat(inputs.socket_listener): Use buffer read time as timestamp #15976

Open

1 task

LarsStegman changed the title ~~feat(inputs.socket_listener): parallel parsing~~ feat(common.socket): parallel parsing Oct 3, 2024

feat(common.socket): remove whitespace only diff

814740d

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(common.socket): parallel parsing #15891

feat(common.socket): parallel parsing #15891

LarsStegman commented Sep 16, 2024

srebhan commented Sep 16, 2024

LarsStegman commented Sep 17, 2024

srebhan commented Sep 19, 2024

LarsStegman commented Sep 19, 2024

srebhan commented Sep 19, 2024

LarsStegman commented Sep 19, 2024 •

edited

Loading

LarsStegman commented Sep 19, 2024

srebhan commented Sep 19, 2024

srebhan left a comment

srebhan Sep 30, 2024

LarsStegman Sep 30, 2024

srebhan Oct 2, 2024

LarsStegman left a comment

LarsStegman Sep 30, 2024

LarsStegman Sep 30, 2024

srebhan Oct 2, 2024

telegraf-tiger bot commented Oct 1, 2024

Artifact URLs

srebhan commented Oct 2, 2024

feat(common.socket): parallel parsing #15891

Are you sure you want to change the base?

feat(common.socket): parallel parsing #15891

Conversation

LarsStegman commented Sep 16, 2024

Summary

Checklist

Related issues

srebhan commented Sep 16, 2024

LarsStegman commented Sep 17, 2024

srebhan commented Sep 19, 2024

LarsStegman commented Sep 19, 2024

srebhan commented Sep 19, 2024

LarsStegman commented Sep 19, 2024 • edited Loading

LarsStegman commented Sep 19, 2024

srebhan commented Sep 19, 2024

srebhan left a comment

Choose a reason for hiding this comment

srebhan Sep 30, 2024

Choose a reason for hiding this comment

LarsStegman Sep 30, 2024

Choose a reason for hiding this comment

srebhan Oct 2, 2024

Choose a reason for hiding this comment

LarsStegman left a comment

Choose a reason for hiding this comment

LarsStegman Sep 30, 2024

Choose a reason for hiding this comment

LarsStegman Sep 30, 2024

Choose a reason for hiding this comment

srebhan Oct 2, 2024

Choose a reason for hiding this comment

telegraf-tiger bot commented Oct 1, 2024

Artifact URLs

srebhan commented Oct 2, 2024

LarsStegman commented Sep 19, 2024 •

edited

Loading