diff --git a/plugins/inputs/directory_monitor/README.md b/plugins/inputs/directory_monitor/README.md index a35c6550c0fda..db4843e36455a 100644 --- a/plugins/inputs/directory_monitor/README.md +++ b/plugins/inputs/directory_monitor/README.md @@ -75,6 +75,10 @@ plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details. ## https://docs.influxdata.com/influxdb/cloud/reference/glossary/#series-cardinality # file_tag = "" # + ## Preserve the original access and modification timestamps when moving + ## processed files to the finished or error directory. + # preserve_timestamps = false + # ## Specify if the file can be read completely at once or if it needs to be read line by line (default). ## Possible values: "line-by-line", "at-once" # parse_method = "line-by-line" diff --git a/plugins/inputs/directory_monitor/directory_monitor.go b/plugins/inputs/directory_monitor/directory_monitor.go index 2ef24a9fdafc9..f813620073d69 100644 --- a/plugins/inputs/directory_monitor/directory_monitor.go +++ b/plugins/inputs/directory_monitor/directory_monitor.go @@ -41,11 +41,12 @@ const ( ) type DirectoryMonitor struct { - Directory string `toml:"directory"` - FinishedDirectory string `toml:"finished_directory"` - Recursive bool `toml:"recursive"` - ErrorDirectory string `toml:"error_directory"` - FileTag string `toml:"file_tag"` + Directory string `toml:"directory"` + FinishedDirectory string `toml:"finished_directory"` + Recursive bool `toml:"recursive"` + ErrorDirectory string `toml:"error_directory"` + FileTag string `toml:"file_tag"` + PreserveTimestamps bool `toml:"preserve_timestamps"` FilesToMonitor []string `toml:"files_to_monitor"` FilesToIgnore []string `toml:"files_to_ignore"` @@ -436,6 +437,25 @@ func (monitor *DirectoryMonitor) moveFile(srcPath, dstBaseDir string) { monitor.Log.Errorf("Could not close input file: %s", err) } + // Close the destination file + if err := outputFile.Close(); err != nil { + monitor.Log.Errorf("Could not close output file: %s", err) + } + + // Restore the timestamps on the moved file to be able to keep track of the original file + if monitor.PreserveTimestamps { + srcTimes, err := times.Stat(srcPath) + if err != nil { + monitor.Log.Errorf("Could not read timestamps of %q: %v", srcPath, err) + } + + if srcTimes != nil { + if err := os.Chtimes(dstPath, srcTimes.AccessTime(), srcTimes.ModTime()); err != nil { + monitor.Log.Errorf("Could not preserve timestamps on %q: %v", dstPath, err) + } + } + } + if err := os.Remove(srcPath); err != nil { monitor.Log.Errorf("Failed removing original file: %s", err) } diff --git a/plugins/inputs/directory_monitor/directory_monitor_test.go b/plugins/inputs/directory_monitor/directory_monitor_test.go index 35b8f39ccae25..230d4c89eb4af 100644 --- a/plugins/inputs/directory_monitor/directory_monitor_test.go +++ b/plugins/inputs/directory_monitor/directory_monitor_test.go @@ -7,6 +7,7 @@ import ( "path/filepath" "runtime" "testing" + "time" "github.com/stretchr/testify/require" @@ -680,3 +681,54 @@ func TestParseSubdirectoriesFilesIgnore(t *testing.T) { _, err = os.Stat(filepath.Join(finishedDirectory, testJSONFile)) require.NoError(t, err) } + +func TestPreserveTimestamps(t *testing.T) { + testJSONFile := "test.json" + originalTime := time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC) + + runMonitor := func(t *testing.T, preserve bool) os.FileInfo { + t.Helper() + acc := testutil.Accumulator{} + finishedDirectory := t.TempDir() + processDirectory := t.TempDir() + + r := DirectoryMonitor{ + Directory: processDirectory, + FinishedDirectory: finishedDirectory, + PreserveTimestamps: preserve, + MaxBufferedMetrics: defaultMaxBufferedMetrics, + FileQueueSize: defaultFileQueueSize, + ParseMethod: defaultParseMethod, + } + require.NoError(t, r.Init()) + r.SetParserFunc(func() (telegraf.Parser, error) { + p := &json.Parser{NameKey: "Name"} + err := p.Init() + return p, err + }) + + srcPath := filepath.Join(processDirectory, testJSONFile) + require.NoError(t, os.WriteFile(srcPath, []byte(`{"Name": "event1", "Speed": 100.1}`), 0640)) + require.NoError(t, os.Chtimes(srcPath, originalTime, originalTime)) + + r.Log = testutil.Logger{} + require.NoError(t, r.Start(&acc)) + require.NoError(t, r.Gather(&acc)) + acc.Wait(1) + r.Stop() + + info, err := os.Stat(filepath.Join(finishedDirectory, testJSONFile)) + require.NoError(t, err) + return info + } + + t.Run("enabled keeps the original modification time", func(t *testing.T) { + info := runMonitor(t, true) + require.True(t, info.ModTime().Equal(originalTime), "expected %s, got %s", originalTime, info.ModTime()) + }) + + t.Run("disabled uses the move time", func(t *testing.T) { + info := runMonitor(t, false) + require.False(t, info.ModTime().Equal(originalTime), "expected the move time, got the original timestamp") + }) +} diff --git a/plugins/inputs/directory_monitor/sample.conf b/plugins/inputs/directory_monitor/sample.conf index 489a6839a6030..2dfe433d59c58 100644 --- a/plugins/inputs/directory_monitor/sample.conf +++ b/plugins/inputs/directory_monitor/sample.conf @@ -39,6 +39,10 @@ ## https://docs.influxdata.com/influxdb/cloud/reference/glossary/#series-cardinality # file_tag = "" # + ## Preserve the original access and modification timestamps when moving + ## processed files to the finished or error directory. + # preserve_timestamps = false + # ## Specify if the file can be read completely at once or if it needs to be read line by line (default). ## Possible values: "line-by-line", "at-once" # parse_method = "line-by-line"