feat(inputs.directory_monitor): Allow to preserve timestamps when moving file

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"

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)
 	}

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")
+	})
+}

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"