I am new to Telegraf and I’m currently working with an external plugin. I’ve forked the internal Kafka Consumer plugin and followed the documentation to build an external plugin using the shim.

However, when I run the plugin using execd, I encounter issues with obtaining a reference to telegraf.parser.

Below is my configuration:

[agent]
  interval = "3s"
  flush_interval = "3s"

[[outputs.file]]
  files = ["stdout"]
[[inputs.execd]]
command = ["/home/user/telegraf/shared-plugins/kafka-consumer2", "--config", "/home/user/telegraf/shared-plugins/kafka-consumer2.conf"]
data_format = "value"
data_type = "string"
signal = "none"

And the error messgae:

2024-05-08T08:43:15Z E! [inputs.execd] stderr: "panic: runtime error: invalid memory address or nil pointer dereference"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "[signal SIGSEGV: segmentation violation code=0x1 addr=0x18 pc=0x913bf5]"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: ""
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "goroutine 42 [running]:"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "github.com/colataco/telegraf-plugins/plugins/inputs/ose_kafka_consumer.(*ConsumerGroupHandler).Handle(0xc00010f220, {0xc29f90, 0xc0003a2120}, 0xc0003ac320)"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "t/home/taco/go/pkg/mod/github.com/colataco/[email protected]/plugins/inputs/ose_kafka_consumer/ose_kafka_consumer.go:493 +0x195"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "github.com/colataco/telegraf-plugins/plugins/inputs/ose_kafka_consumer.(*ConsumerGroupHandler).ConsumeClaim(0xc00010f220, {0xc29f90, 0xc0003a2120}, {0xc26880, 0xc00039a3c0})"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "t/home/taco/go/pkg/mod/github.com/colataco/[email protected]/plugins/inputs/ose_kafka_consumer/ose_kafka_consumer.go:561 +0x119"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "github.com/IBM/sarama.(*consumerGroupSession).consume(0xc0003a2120, {0xc0002fa000, 0x12}, 0x0)"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "t/home/taco/go/pkg/mod/github.com/!i!b!m/[email protected]/consumer_group.go:952 +0x21b"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "github.com/IBM/sarama.newConsumerGroupSession.func2({0xc0002fa000?, 0x0?}, 0x0?)"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "t/home/taco/go/pkg/mod/github.com/!i!b!m/[email protected]/consumer_group.go:877 +0x70"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "created by github.com/IBM/sarama.newConsumerGroupSession in goroutine 23"
2024-05-08T08:43:15Z E! [inputs.execd] stderr: "t/home/taco/go/pkg/mod/github.com/!i!b!m/[email protected]/consumer_group.go:869 +0x456"
2024-05-08T08:43:15Z E! [inputs.execd] Process /home/user/telegraf/shared-plugins/ose-kafka-consumer exited: exit status 2
2024-05-08T08:43:15Z I! [inputs.execd] Restarting in 10s...

the source code for line 493

func (h *ConsumerGroupHandler) Handle(session sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) error {
    if h.MaxMessageLen != 0 && len(msg.Value) > h.MaxMessageLen {
        session.MarkMessage(msg, "")
        h.release()
        return fmt.Errorf("message exceeds max_message_len (actual %d, max %d)",
            len(msg.Value), h.MaxMessageLen)
    }
   ===> metrics, err := h.parser.Parse(msg.Value)
    if err != nil {
        session.MarkMessage(msg, "")
        h.release()
        return err
    }
...

I’ve tried adding data_format and data_type options in the configuration, but I’m still encountering a nil error message when running the plugin.

Can anyone provide guidance on how to resolve this issue?