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

(cherry picked from commit 629a223)

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"