feat(migrations): Add migration for inputs.kafka_consumer connection_strategy option

The connection_strategy option was deprecated in v1.33.0 for removal in
v1.40.0, replaced by startup_error_behavior. Remove it and add a migration
mapping "defer" to startup_error_behavior = "retry", dropping the "startup"
default and erroring on contradicting values.

Author: skartikey

migrations/all/inputs_kafka_consumer.go

@@ -0,0 +1,5 @@
+//go:build !custom || (migrations && (inputs || inputs.kafka_consumer))
+
+package all
+
+import _ "github.com/influxdata/telegraf/migrations/inputs_kafka_consumer" // register migration

migrations/inputs_kafka_consumer/migration.go

@@ -0,0 +1,77 @@
+package inputs_kafka_consumer
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/influxdata/toml"
+	"github.com/influxdata/toml/ast"
+
+	"github.com/influxdata/telegraf/migrations"
+)
+
+// Migration function to migrate the deprecated 'connection_strategy' option to
+// its replacement 'startup_error_behavior'. The "defer" strategy, which allowed
+// the plugin to start even if the broker was unavailable, maps to the "retry"
+// behavior. The "startup" strategy (and the empty default) corresponds to the
+// default "error" behavior, so the option is simply removed in that case.
+func migrate(tbl *ast.Table) ([]byte, string, error) {
+	// Decode the old data structure
+	var plugin map[string]interface{}
+	if err := toml.UnmarshalTable(tbl, &plugin); err != nil {
+		return nil, "", err
+	}
+
+	// Check for the deprecated option and migrate it
+	var applied bool
+	if rawOldValue, found := plugin["connection_strategy"]; found {
+		applied = true
+
+		// Convert to the actual type
+		oldValue, ok := rawOldValue.(string)
+		if !ok {
+			return nil, "", fmt.Errorf("unexpected type %T for 'connection_strategy'", rawOldValue)
+		}
+
+		switch strings.ToLower(oldValue) {
+		case "", "startup":
+			// Default behavior corresponding to 'startup_error_behavior = "error"'
+		case "defer":
+			// Check if the new option already exists and if it has a
+			// contradicting value. If the new option is not present, migrate
+			// the old value to the equivalent behavior.
+			if rawNewValue, found := plugin["startup_error_behavior"]; found {
+				if newValue, ok := rawNewValue.(string); !ok {
+					return nil, "", fmt.Errorf("unexpected type %T for 'startup_error_behavior'", rawNewValue)
+				} else if newValue != "retry" {
+					return nil, "", errors.New("contradicting setting for 'startup_error_behavior' and 'connection_strategy'")
+				}
+			} else {
+				plugin["startup_error_behavior"] = "retry"
+			}
+		default:
+			return nil, "", fmt.Errorf("invalid connection strategy %q", oldValue)
+		}
+
+		// Remove the deprecated setting
+		delete(plugin, "connection_strategy")
+	}
+
+	// No options migrated so we can exit early
+	if !applied {
+		return nil, "", migrations.ErrNotApplicable
+	}
+
+	// Create the corresponding plugin configuration
+	cfg := migrations.CreateTOMLStruct("inputs", "kafka_consumer")
+	cfg.Add("inputs", "kafka_consumer", plugin)
+
+	output, err := toml.Marshal(cfg)
+	return output, "", err
+}
+
+// Register the migration function for the plugin type
+func init() {
+	migrations.AddPluginOptionMigration("inputs.kafka_consumer", migrate)
+}

migrations/inputs_kafka_consumer/migration_test.go

@@ -0,0 +1,103 @@
+package inputs_kafka_consumer_test
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf/config"
+	_ "github.com/influxdata/telegraf/migrations/inputs_kafka_consumer" // register migration
+	"github.com/influxdata/telegraf/plugins/inputs/kafka_consumer"      // register plugin
+	_ "github.com/influxdata/telegraf/plugins/parsers/influx"           // register parser
+)
+
+func TestNoMigration(t *testing.T) {
+	plugin := &kafka_consumer.KafkaConsumer{}
+	defaultCfg := plugin.SampleConfig()
+
+	// Migrate and check that nothing changed
+	output, n, err := config.ApplyMigrations([]byte(defaultCfg))
+	require.NoError(t, err)
+	require.NotEmpty(t, output)
+	require.Zero(t, n)
+	require.Equal(t, defaultCfg, string(output))
+}
+
+func TestStartupErrorBehaviorConflict(t *testing.T) {
+	cfg := []byte(`
+[[inputs.kafka_consumer]]
+  brokers = ["localhost:9092"]
+  topics = ["telegraf"]
+  connection_strategy = "defer"
+  startup_error_behavior = "ignore"
+	`)
+	// Migrate and check that the contradicting settings are caught
+	output, n, err := config.ApplyMigrations(cfg)
+	require.ErrorContains(t, err, "contradicting setting for 'startup_error_behavior' and 'connection_strategy'")
+	require.Empty(t, output)
+	require.Zero(t, n)
+}
+
+func TestInvalidConnectionStrategy(t *testing.T) {
+	cfg := []byte(`
+[[inputs.kafka_consumer]]
+  brokers = ["localhost:9092"]
+  topics = ["telegraf"]
+  connection_strategy = "invalid"
+	`)
+	// Migrate and check that the invalid value is caught
+	output, n, err := config.ApplyMigrations(cfg)
+	require.ErrorContains(t, err, `invalid connection strategy "invalid"`)
+	require.Empty(t, output)
+	require.Zero(t, n)
+}
+
+func TestCases(t *testing.T) {
+	// Get all directories in testcases
+	folders, err := os.ReadDir("testcases")
+	require.NoError(t, err)
+
+	for _, f := range folders {
+		// Only handle folders
+		if !f.IsDir() {
+			continue
+		}
+
+		t.Run(f.Name(), func(t *testing.T) {
+			testcasePath := filepath.Join("testcases", f.Name())
+			inputFile := filepath.Join(testcasePath, "telegraf.conf")
+			expectedFile := filepath.Join(testcasePath, "expected.conf")
+
+			// Read the expected output
+			expected := config.NewConfig()
+			require.NoError(t, expected.LoadConfig(expectedFile))
+			require.NotEmpty(t, expected.Inputs)
+
+			// Read the input data
+			input, remote, err := config.LoadConfigFile(inputFile)
+			require.NoError(t, err)
+			require.False(t, remote)
+			require.NotEmpty(t, input)
+
+			// Migrate
+			output, n, err := config.ApplyMigrations(input)
+			require.NoError(t, err)
+			require.NotEmpty(t, output)
+			require.GreaterOrEqual(t, n, uint64(1))
+			actual := config.NewConfig()
+			require.NoError(t, actual.LoadConfigData(output, config.EmptySourcePath))
+
+			// Test the output
+			require.Len(t, actual.Inputs, len(expected.Inputs))
+			actualIDs := make([]string, 0, len(expected.Inputs))
+			expectedIDs := make([]string, 0, len(expected.Inputs))
+			for i := range actual.Inputs {
+				actualIDs = append(actualIDs, actual.Inputs[i].ID())
+				expectedIDs = append(expectedIDs, expected.Inputs[i].ID())
+			}
+			require.ElementsMatch(t, expectedIDs, actualIDs, string(output))
+		})
+	}
+}

migrations/inputs_kafka_consumer/testcases/defer/expected.conf

@@ -0,0 +1,10 @@
+# Read metrics from Kafka topics
+[[inputs.kafka_consumer]]
+  ## Kafka brokers.
+  brokers = ["localhost:9092"]
+
+  ## Topics to consume.
+  topics = ["telegraf"]
+
+  ## Allow the plugin to start before the broker is available
+  startup_error_behavior = "retry"

migrations/inputs_kafka_consumer/testcases/defer/telegraf.conf

@@ -0,0 +1,10 @@
+# Read metrics from Kafka topics
+[[inputs.kafka_consumer]]
+  ## Kafka brokers.
+  brokers = ["localhost:9092"]
+
+  ## Topics to consume.
+  topics = ["telegraf"]
+
+  ## Allow the plugin to start before the broker is available
+  connection_strategy = "defer"

migrations/inputs_kafka_consumer/testcases/startup/expected.conf

@@ -0,0 +1,9 @@
+# Read metrics from Kafka topics
+[[inputs.kafka_consumer]]
+  ## Kafka brokers.
+  brokers = ["localhost:9092"]
+
+  ## Topics to consume.
+  topics = ["telegraf"]
+
+  ## Connect to the broker at startup

migrations/inputs_kafka_consumer/testcases/startup/telegraf.conf

@@ -0,0 +1,10 @@
+# Read metrics from Kafka topics
+[[inputs.kafka_consumer]]
+  ## Kafka brokers.
+  brokers = ["localhost:9092"]
+
+  ## Topics to consume.
+  topics = ["telegraf"]
+
+  ## Connect to the broker at startup
+  connection_strategy = "startup"

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -51,7 +51,6 @@ type KafkaConsumer struct {
 	ConsumerFetchDefault                 config.Size     `toml:"consumer_fetch_default"`
 	ConsumerFetchMin                     config.Size     `toml:"consumer_fetch_min"`
 	ConsumerFetchMaxWait                 config.Duration `toml:"consumer_fetch_max_wait"`
-	ConnectionStrategy                   string          `toml:"connection_strategy" deprecated:"1.33.0;1.40.0;use 'startup_error_behavior' instead"`
 	ResolveCanonicalBootstrapServersOnly bool            `toml:"resolve_canonical_bootstrap_servers_only"`
 	Log                                  telegraf.Logger `toml:"-"`
 	kafka.ReadConfig
@@ -200,12 +199,6 @@ func (k *KafkaConsumer) Init() error {
 		cfg.Consumer.MaxWaitTime = time.Duration(k.ConsumerFetchMaxWait)
 	}
 
-	switch strings.ToLower(k.ConnectionStrategy) {
-	default:
-		return fmt.Errorf("invalid connection strategy %q", k.ConnectionStrategy)
-	case "defer", "startup", "":
-	}
-
 	k.config = cfg
 
 	if len(k.TopicRegexps) == 0 {
@@ -231,8 +224,6 @@ func (k *KafkaConsumer) SetParser(parser telegraf.Parser) {
 }
 
 func (k *KafkaConsumer) Start(acc telegraf.Accumulator) error {
-	var err error
-
 	// If TopicRegexps is set, add matches to Topics
 	if len(k.TopicRegexps) > 0 {
 		if err := k.refreshTopics(); err != nil {
@@ -243,33 +234,19 @@ func (k *KafkaConsumer) Start(acc telegraf.Accumulator) error {
 	ctx, cancel := context.WithCancel(context.Background())
 	k.cancel = cancel
 
-	if k.ConnectionStrategy != "defer" {
-		err = k.create()
-		if err != nil {
-			return &internal.StartupError{
-				Err:   fmt.Errorf("create consumer: %w", err),
-				Retry: errors.Is(err, sarama.ErrOutOfBrokers),
-			}
+	if err := k.create(); err != nil {
+		return &internal.StartupError{
+			Err:   fmt.Errorf("create consumer: %w", err),
+			Retry: errors.Is(err, sarama.ErrOutOfBrokers),
 		}
-		k.startErrorAdder(acc)
 	}
+	k.startErrorAdder(acc)
 
 	// Start consumer goroutine
 	k.wg.Add(1)
 	go func() {
-		var err error
 		defer k.wg.Done()
 
-		if k.consumer == nil {
-			err = k.create()
-			if err != nil {
-				acc.AddError(fmt.Errorf("create consumer async: %w", err))
-				return
-			}
-		}
-
-		k.startErrorAdder(acc)
-
 		for ctx.Err() == nil {
 			handler := newConsumerGroupHandler(acc, k.MaxUndeliveredMessages, k.parser, k.Log)
 			handler.maxMessageLen = k.MaxMessageLen
@@ -300,8 +277,7 @@ func (k *KafkaConsumer) Start(acc telegraf.Accumulator) error {
 				internal.SleepContext(ctx, reconnectDelay) //nolint:errcheck // ignore returned error as we cannot do anything about it anyway
 			}
 		}
-		err = k.consumer.Close()
-		if err != nil {
+		if err := k.consumer.Close(); err != nil {
 			acc.AddError(fmt.Errorf("close: %w", err))
 		}
 	}()