Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixed race condition in the Wavefront parser #5764

Merged
merged 3 commits into from
Apr 25, 2019
Merged

Fixed race condition in the Wavefront parser #5764

merged 3 commits into from
Apr 25, 2019

Conversation

prydin
Copy link
Contributor

@prydin prydin commented Apr 24, 2019

Required for all PRs:

  • Signed CLA.
  • Associated README.md updated.
  • Has appropriate unit tests.

The Wavefront parser had some statefulness to it making it panic when receiving heavy traffic from multiple sources. Solved it by moving stateful objects to a separate struct and using a sync.Pool to make sure they are reused but never shared across goroutines.

Copy link
Contributor

@danielnelson danielnelson left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BTW, someday I want to make it so the parsers don't need to be threadsafe and can be stateful. I messed around with this on the tail plugin a bit, we needed the parser to know what file it was parsing in order to name metrics. Ended up with this interface as a quick fix:

type ParserFuncInput interface {
	// GetParser returns a new parser.
	SetParserFunc(fn ParserFunc)
}

I think it's not quite the interface we want long term, but what we really need is a way for plugins to get shared things from Telegraf (perhaps a logger or other objects besides just the parser). On inputs we could stick things into the accumulator, but for outputs/processors it doesn't really work.

@@ -61,6 +99,8 @@ func (p *PointParser) Parse(buf []byte) ([]telegraf.Metric, error) {
buf = append(buf, []byte("\n")...)
}

//log.Printf("D! [parsers.wavefront] Received data: %s", string(buf))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you remove this?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done!

wp := &WavefrontParser{defaultTags: defaultTags}
wp.parsers = &sync.Pool{
New: func() interface{} {
log.Printf("D! [parsers.wavefront] Creating new parser for %s", wp)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(This is the ci test failure) I would just remove the for %s part completely. Default tags never change unless Telegraf is reloaded and are shared on all inputs.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done!

@danielnelson danielnelson added this to the 1.11.0 milestone Apr 24, 2019
@danielnelson danielnelson added the fix pr to fix corresponding bug label Apr 24, 2019
@danielnelson danielnelson modified the milestones: 1.11.0, 1.10.4 Apr 24, 2019
@danielnelson danielnelson merged commit f32b064 into influxdata:master Apr 25, 2019
danielnelson pushed a commit that referenced this pull request Apr 25, 2019
hwaastad pushed a commit to hwaastad/telegraf that referenced this pull request Jun 13, 2019
bitcharmer pushed a commit to bitcharmer/telegraf that referenced this pull request Oct 18, 2019
athoune pushed a commit to bearstech/telegraf that referenced this pull request Apr 17, 2020
idohalevi pushed a commit to idohalevi/telegraf that referenced this pull request Sep 29, 2020
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
fix pr to fix corresponding bug
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants