Add max_message_len in kafka_consumer input #2636

nfirvine · 2017-04-07T21:25:29Z

Partially fixes #2606

danielnelson

Looks good, I just have a few minor things.

danielnelson · 2017-04-07T22:15:17Z

plugins/inputs/kafka_consumer/README.md

@@ -28,6 +28,9 @@ from the same topic in parallel.
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"
+
+  ## Maximum length of a message to consume, in bytes (default 0/unlimited); larger messages are dropped


Wrap at 78 chars

danielnelson · 2017-04-07T22:15:26Z

plugins/inputs/kafka_consumer/README.md

@@ -28,6 +28,9 @@ from the same topic in parallel.
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"
+
+  ## Maximum length of a message to consume, in bytes (default 0/unlimited); larger messages are dropped
+  max_message_len = 65536


max_message_len = 64 * 1024

You can do math in an TOML file? TIL

Oh whoops, I'm clearly just not paying much attention.

danielnelson · 2017-04-07T22:16:11Z

plugins/inputs/kafka_consumer/kafka_consumer.go

  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"
+
+  ## Maximum length of a message to consume, in bytes (default 0/unlimited); larger messages are dropped
+  max_message_len = 65536


Same comments as above

danielnelson · 2017-04-07T22:16:31Z

plugins/inputs/kafka_consumer/kafka_consumer.go

-			for _, metric := range metrics {
-				k.acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time())
+			if k.MaxMessageLen != 0 && len(msg.Value) > k.MaxMessageLen {
+				k.acc.AddError(fmt.Errorf("D! Kafka message longer than max_message_len (%d)", k.MaxMessageLen))


Lets make this a W! level. AddError should put the plugin name so possibly the Kafka part is redundant, but make sure to add some information about what we are going to do. Maybe it should be "W! Message longer than max_message_len (%d); dropping message".

I was trying to avoid dumping the message to the logs, since it could be very long, and thereby fill/ruin the logs. At least for me, it can make them very difficult to read.

Yeah, it's just the max length here though, not the message.

Actually I think AddError does not take a level prefix at all, and always logs at E!

Yep, looks like. I'll remove all level tags then

danielnelson · 2017-04-07T22:16:52Z

plugins/inputs/kafka_consumer/kafka_consumer_test.go

@@ -62,6 +63,23 @@ func TestRunParserInvalidMsg(t *testing.T) {
 	assert.Equal(t, acc.NFields(), 0)
 }

+// Test that


Remove comment

nfirvine · 2017-04-07T22:54:25Z

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -133,14 +137,17 @@ func (k *Kafka) receiver() {
 				k.acc.AddError(fmt.Errorf("Kafka Consumer Error: %s\n", err))


Was thinking of adding a tag here, probably 'E!'. Make sense? And should it end in a newline?

Kind of scope creep, but thought I'd do it while I was in the area

No it shouldn't have a newline, but AddError logs only at E! level.

nfirvine · 2017-04-07T22:56:01Z

plugins/inputs/kafka_consumer/kafka_consumer.go

+			} else {
+				metrics, err := k.parser.Parse(msg.Value)
+				if err != nil {
+					k.acc.AddError(fmt.Errorf("E! Kafka Message Parse Error\nmessage: %s\nerror: %s",


This logging of message parsing failures in the default logging level is IMO a bad idea and I think (for reasons I listed elsewhere should be a D!)

Idea being that you would notice metrics are being dropped from the gather_metrics internal stat, and restart in debug mode to learn more.

I agree, we should remove the logging of the message.

Turns out AddError forces E!-level, which makes this a much bigger change. I'm willing to just do E!

danielnelson · 2017-04-07T23:23:14Z

plugins/inputs/kafka_consumer/README.md

+
+  ## Maximum length of a message to consume, in bytes (default 0/unlimited);
+  ## larger messages are dropped
+  max_message_len = 64 * 1024


I commented back a bit on this, but in case it is lost in the "outdated comments". You can't do this, I just wasn't paying attention to this being TOML in my original comment.

phemmer · 2017-04-09T05:56:01Z

plugins/inputs/kafka_consumer/kafka_consumer.go

+				}
+				for _, metric := range metrics {
+					k.acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time())
+				}


https://github.com/golang/go/wiki/CodeReviewComments#indent-error-flow
You can avoid having to increase the indentation on all this code by doing

if k.MaxMessageLen != 0 && len(msg.Value) > k.MaxMessageLen { k.acc.AddError(fmt.Errorf("Message longer than max_message_len (%d)", k.MaxMessageLen)) continue }

I think it still needs to call the bit below to ack the message.

Right, that was my concern.

phemmer · 2017-04-09T05:57:09Z

plugins/inputs/kafka_consumer/kafka_consumer.go

-			for _, metric := range metrics {
-				k.acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time())
+			if k.MaxMessageLen != 0 && len(msg.Value) > k.MaxMessageLen {
+				k.acc.AddError(fmt.Errorf("Message longer than max_message_len (%d)", k.MaxMessageLen))


The error should probably contain how long the message is, not what the limit is. The user should know the limit as they configured it. But if this error is popping up, they might not know the length of the messages that are trying to come through.

danielnelson

Can you add a line to the CHANGELOG?

danielnelson · 2017-04-11T00:32:58Z

plugins/inputs/kafka_consumer/kafka_consumer.go

+				}
+				for _, metric := range metrics {
+					k.acc.AddFields(metric.Name(), metric.Fields(), metric.Tags(), metric.Time())
+				}


I think it still needs to call the bit below to ack the message.

nfirvine force-pushed the limit-kafka-message-len branch from 5c8187d to 0d70d9d Compare April 7, 2017 21:51

danielnelson suggested changes Apr 7, 2017

View reviewed changes

nfirvine commented Apr 7, 2017

View reviewed changes

danielnelson reviewed Apr 7, 2017

View reviewed changes

phemmer reviewed Apr 9, 2017

View reviewed changes

danielnelson reviewed Apr 11, 2017

View reviewed changes

danielnelson approved these changes Apr 11, 2017

View reviewed changes

Nick Irvine added 6 commits April 11, 2017 11:46

Add max_message_len in kafka_consumer input

6fed7b0

fixes from code review

d9c71b4

remove log level tags when using AddError

907cd7b

Don't do math in TOML

b05537a

Add message length to error message

d1f559f

Update changelog with PR

9207ddf

nfirvine force-pushed the limit-kafka-message-len branch from 13ec6f2 to 9207ddf Compare April 11, 2017 18:46

danielnelson added this to the 1.3.0 milestone Apr 11, 2017

danielnelson merged commit 0193cbe into influxdata:master Apr 11, 2017

nfirvine deleted the limit-kafka-message-len branch April 11, 2017 19:20

vlamug pushed a commit to vlamug/telegraf that referenced this pull request May 30, 2017

Add max_message_len in kafka_consumer input (influxdata#2636)

c173f06

maxunt pushed a commit that referenced this pull request Jun 26, 2018

Add max_message_len in kafka_consumer input (#2636)

198ef8d

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add max_message_len in kafka_consumer input #2636

Add max_message_len in kafka_consumer input #2636

nfirvine commented Apr 7, 2017

danielnelson left a comment

danielnelson Apr 7, 2017

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017

danielnelson Apr 7, 2017

danielnelson Apr 7, 2017

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017

danielnelson Apr 7, 2017

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017 •

edited

Loading

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017 •

edited

Loading

danielnelson Apr 7, 2017

nfirvine Apr 7, 2017

danielnelson Apr 7, 2017

phemmer Apr 9, 2017

danielnelson Apr 11, 2017

nfirvine Apr 11, 2017

phemmer Apr 9, 2017

danielnelson left a comment

danielnelson Apr 11, 2017

		@@ -133,14 +137,17 @@ func (k *Kafka) receiver() {
		k.acc.AddError(fmt.Errorf("Kafka Consumer Error: %s\n", err))

Add max_message_len in kafka_consumer input #2636

Add max_message_len in kafka_consumer input #2636

Conversation

nfirvine commented Apr 7, 2017

danielnelson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nfirvine Apr 7, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nfirvine Apr 7, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

danielnelson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nfirvine Apr 7, 2017 •

edited

Loading

nfirvine Apr 7, 2017 •

edited

Loading