chore: log kinesis consumer events

plugins/inputs/kinesis_consumer/kinesis_consumer.go

@@ -15,8 +15,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis"
+	"github.com/aws/smithy-go/logging"
 	consumer "github.com/harlow/kinesis-consumer"
 	"github.com/harlow/kinesis-consumer/store/ddb"
 
@@ -89,6 +91,25 @@ func (k *KinesisConsumer) SetParser(parser telegraf.Parser) {
 	k.parser = parser
 }
 
+type TelegrafLoggerWrapper struct {
+	telegraf.Logger
+}
+
+func (t *TelegrafLoggerWrapper) Log(args ...interface{}) {
+	t.Debug(args)
+}
+
+func (t *TelegrafLoggerWrapper) Logf(classification logging.Classification, format string, v ...interface{}) {
+	switch classification {
+	case logging.Debug:
+		t.Logger.Debugf(format, v...)
+	case logging.Warn:
+		t.Logger.Warnf(format, v...)
+	default:
+		t.Logger.Infof(format, v...)
+	}
+}
+
 func (k *KinesisConsumer) connect(ac telegraf.Accumulator) error {
 	cfg, err := k.CredentialConfig.Credentials()
 	if err != nil {
@@ -98,6 +119,9 @@ func (k *KinesisConsumer) connect(ac telegraf.Accumulator) error {
 	if k.EndpointURL != "" {
 		cfg.BaseEndpoint = &k.EndpointURL
 	}
+
+	cfg.Logger = &TelegrafLoggerWrapper{k.Log}
+	cfg.ClientLogMode = aws.LogRetries
 	client := kinesis.NewFromConfig(cfg)
 
 	k.checkpoint = &noopStore{}
@@ -119,6 +143,7 @@ func (k *KinesisConsumer) connect(ac telegraf.Accumulator) error {
 		consumer.WithClient(client),
 		consumer.WithShardIteratorType(k.ShardIteratorType),
 		consumer.WithStore(k),
+		consumer.WithLogger(&TelegrafLoggerWrapper{k.Log}),
 	)
 	if err != nil {
 		return err