feat(migrations): Add migration for inputs.tail from_beginning option

The 'from_beginning' option in inputs.tail was deprecated in v1.34.0 for removal in v1.40.0, replaced by 'initial_read_offset'.

Author: skartikey

migrations/all/inputs_tail.go

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

migrations/inputs_tail/migration.go

@@ -0,0 +1,76 @@
+package inputs_tail
+
+import (
+	"fmt"
+
+	"github.com/influxdata/toml"
+	"github.com/influxdata/toml/ast"
+
+	"github.com/influxdata/telegraf/migrations"
+)
+
+// Migration function to migrate the deprecated 'from_beginning' option to its
+// replacement 'initial_read_offset'.
+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
+	var message string
+	if rawOldValue, found := plugin["from_beginning"]; found {
+		applied = true
+
+		// Convert to the actual type
+		oldValue, ok := rawOldValue.(bool)
+		if !ok {
+			return nil, "", fmt.Errorf("unexpected type %T for 'from_beginning'", rawOldValue)
+		}
+
+		// A 'from_beginning' value of 'true' corresponds to reading from the
+		// 'beginning' while 'false' corresponds to the default 'saved-or-end'
+		// behavior.
+		expected := "saved-or-end"
+		if oldValue {
+			expected = "beginning"
+		}
+
+		// The 'initial_read_offset' option supersedes 'from_beginning' and takes
+		// precedence at runtime. If it is already set, keep it and just drop the
+		// deprecated option, but warn the user if the two settings disagree.
+		if rawNewValue, found := plugin["initial_read_offset"]; found {
+			newValue, ok := rawNewValue.(string)
+			if !ok {
+				return nil, "", fmt.Errorf("unexpected type %T for 'initial_read_offset'", rawNewValue)
+			}
+			if newValue != expected {
+				message = fmt.Sprintf("ignoring 'from_beginning = %v' as it conflicts with 'initial_read_offset = %q'", oldValue, newValue)
+			}
+		} else {
+			plugin["initial_read_offset"] = expected
+		}
+
+		// Remove the deprecated setting
+		delete(plugin, "from_beginning")
+	}
+
+	// No options migrated so we can exit early
+	if !applied {
+		return nil, "", migrations.ErrNotApplicable
+	}
+
+	// Create the corresponding plugin configuration
+	cfg := migrations.CreateTOMLStruct("inputs", "tail")
+	cfg.Add("inputs", "tail", plugin)
+
+	output, err := toml.Marshal(cfg)
+	return output, message, err
+}
+
+// Register the migration function for the plugin type
+func init() {
+	migrations.AddPluginOptionMigration("inputs.tail", migrate)
+}

migrations/inputs_tail/migration_test.go

@@ -0,0 +1,126 @@
+package inputs_tail_test
+
+import (
+	"bytes"
+	"log"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf/config"
+	_ "github.com/influxdata/telegraf/migrations/inputs_tail" // register migration
+	"github.com/influxdata/telegraf/plugins/inputs/tail"      // register plugin
+	_ "github.com/influxdata/telegraf/plugins/parsers/influx" // register parser
+)
+
+func TestNoMigration(t *testing.T) {
+	plugin := &tail.Tail{}
+	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 TestConflictWarning(t *testing.T) {
+	tests := []struct {
+		name string
+		conf string
+		warn bool
+	}{
+		{
+			name: "conflicting values warn",
+			conf: `
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+  from_beginning = true
+  initial_read_offset = "end"
+  data_format = "influx"
+`,
+			warn: true,
+		},
+		{
+			name: "agreeing values do not warn",
+			conf: `
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+  from_beginning = true
+  initial_read_offset = "beginning"
+  data_format = "influx"
+`,
+			warn: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var buf bytes.Buffer
+			log.SetOutput(&buf)
+			t.Cleanup(func() { log.SetOutput(os.Stderr) })
+
+			output, n, err := config.ApplyMigrations([]byte(tt.conf))
+			require.NoError(t, err)
+			require.NotEmpty(t, output)
+			require.Equal(t, uint64(1), n)
+
+			if tt.warn {
+				require.Contains(t, buf.String(), `conflicts with 'initial_read_offset = "end"'`)
+			} else {
+				require.NotContains(t, buf.String(), "conflicts with")
+			}
+		})
+	}
+}
+
+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_tail/testcases/existing/expected.conf

@@ -0,0 +1,8 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## The new option takes precedence over the deprecated one
+  initial_read_offset = "end"
+
+  data_format = "influx"

migrations/inputs_tail/testcases/existing/telegraf.conf

@@ -0,0 +1,9 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## The new option takes precedence over the deprecated one
+  from_beginning = true
+  initial_read_offset = "end"
+
+  data_format = "influx"

migrations/inputs_tail/testcases/from_beginning_false/expected.conf

@@ -0,0 +1,8 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## Keep the default behavior using the deprecated option
+  initial_read_offset = "saved-or-end"
+
+  data_format = "influx"

migrations/inputs_tail/testcases/from_beginning_false/telegraf.conf

@@ -0,0 +1,8 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## Keep the default behavior using the deprecated option
+  from_beginning = false
+
+  data_format = "influx"

migrations/inputs_tail/testcases/from_beginning_true/expected.conf

@@ -0,0 +1,8 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## Read from the beginning using the deprecated option
+  initial_read_offset = "beginning"
+
+  data_format = "influx"

migrations/inputs_tail/testcases/from_beginning_true/telegraf.conf

@@ -0,0 +1,8 @@
+# Parse the new lines appended to a file
+[[inputs.tail]]
+  files = ["/var/mymetrics.out"]
+
+  ## Read from the beginning using the deprecated option
+  from_beginning = true
+
+  data_format = "influx"