package kafka_consumer import ( "context" "strings" "testing" "github.com/Shopify/sarama" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/plugins/parsers" "github.com/influxdata/telegraf/testutil" "github.com/stretchr/testify/assert" ) const ( testMsg = "cpu_load_short,host=server01 value=23422.0 1422568543702900257\n" testMsgGraphite = "cpu.load.short.graphite 23422 1454780029" testMsgJSON = "{\"a\": 5, \"b\": {\"c\": 6}}\n" invalidMsg = "cpu_load_short,host=server01 1422568543702900257\n" ) type TestConsumer struct { errors chan error messages chan *sarama.ConsumerMessage } func (c *TestConsumer) Errors() <-chan error { return c.errors } func (c *TestConsumer) Messages() <-chan *sarama.ConsumerMessage { return c.messages } func (c *TestConsumer) MarkOffset(msg *sarama.ConsumerMessage, metadata string) { } func (c *TestConsumer) Close() error { return nil } func (c *TestConsumer) Inject(msg *sarama.ConsumerMessage) { c.messages <- msg } func newTestKafka() (*Kafka, *TestConsumer) { consumer := &TestConsumer{ errors: make(chan error), messages: make(chan *sarama.ConsumerMessage, 1000), } k := Kafka{ cluster: consumer, ConsumerGroup: "test", Topics: []string{"telegraf"}, Brokers: []string{"localhost:9092"}, Offset: "oldest", MaxUndeliveredMessages: defaultMaxUndeliveredMessages, doNotCommitMsgs: true, messages: make(map[telegraf.TrackingID]*sarama.ConsumerMessage), } return &k, consumer } // Test that the parser parses kafka messages into points func TestRunParser(t *testing.T) { k, consumer := newTestKafka() acc := testutil.Accumulator{} ctx := context.Background() k.parser, _ = parsers.NewInfluxParser() go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(testMsg)) acc.Wait(1) assert.Equal(t, acc.NFields(), 1) } // Test that the parser ignores invalid messages func TestRunParserInvalidMsg(t *testing.T) { k, consumer := newTestKafka() acc := testutil.Accumulator{} ctx := context.Background() k.parser, _ = parsers.NewInfluxParser() go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(invalidMsg)) acc.WaitError(1) assert.Equal(t, acc.NFields(), 0) } // Test that overlong messages are dropped func TestDropOverlongMsg(t *testing.T) { const maxMessageLen = 64 * 1024 k, consumer := newTestKafka() k.MaxMessageLen = maxMessageLen acc := testutil.Accumulator{} ctx := context.Background() overlongMsg := strings.Repeat("v", maxMessageLen+1) go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(overlongMsg)) acc.WaitError(1) assert.Equal(t, acc.NFields(), 0) } // Test that the parser parses kafka messages into points func TestRunParserAndGather(t *testing.T) { k, consumer := newTestKafka() acc := testutil.Accumulator{} ctx := context.Background() k.parser, _ = parsers.NewInfluxParser() go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(testMsg)) acc.Wait(1) acc.GatherError(k.Gather) assert.Equal(t, acc.NFields(), 1) acc.AssertContainsFields(t, "cpu_load_short", map[string]interface{}{"value": float64(23422)}) } // Test that the parser parses kafka messages into points func TestRunParserAndGatherGraphite(t *testing.T) { k, consumer := newTestKafka() acc := testutil.Accumulator{} ctx := context.Background() k.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil) go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(testMsgGraphite)) acc.Wait(1) acc.GatherError(k.Gather) assert.Equal(t, acc.NFields(), 1) acc.AssertContainsFields(t, "cpu_load_short_graphite", map[string]interface{}{"value": float64(23422)}) } // Test that the parser parses kafka messages into points func TestRunParserAndGatherJSON(t *testing.T) { k, consumer := newTestKafka() acc := testutil.Accumulator{} ctx := context.Background() k.parser, _ = parsers.NewParser(&parsers.Config{ DataFormat: "json", MetricName: "kafka_json_test", }) go k.receiver(ctx, &acc) consumer.Inject(saramaMsg(testMsgJSON)) acc.Wait(1) acc.GatherError(k.Gather) assert.Equal(t, acc.NFields(), 2) acc.AssertContainsFields(t, "kafka_json_test", map[string]interface{}{ "a": float64(5), "b_c": float64(6), }) } func saramaMsg(val string) *sarama.ConsumerMessage { return &sarama.ConsumerMessage{ Key: nil, Value: []byte(val), Offset: 0, Partition: 0, } }