feat(inputs.directory_monitor): Improve internal stats (#13089)

plugins/inputs/directory_monitor/README.md

@@ -82,7 +82,23 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
 The format of metrics produced by this plugin depends on the content and data
 format of the file.
 
+When the [internal][] input is enabled:
+
+- internal_directory_monitor
+  - fields:
+    - files_processed - How many files have been processed (counter)
+    - files_dropped - How many files have been dropped (counter)
+- internal_directory_monitor
+  - tags:
+    - directory - The monitored directory
+  - fields:
+    - files_processed_per_dir - How many files have been processed (counter)
+    - files_dropped_per_dir - How many files have been dropped (counter)
+    - files_queue_per_dir - How many files to be processed (gauge)
+
 ## Example Output
 
 The metrics produced by this plugin depends on the content and data
 format of the file.
+
+[internal]: /plugins/inputs/internal

plugins/inputs/directory_monitor/directory_monitor.go

@@ -59,7 +59,10 @@ type DirectoryMonitor struct {
 	context             context.Context
 	parserFunc          parsers.ParserFunc
 	filesProcessed      selfstat.Stat
+	filesProcessedDir   selfstat.Stat
 	filesDropped        selfstat.Stat
+	filesDroppedDir     selfstat.Stat
+	filesQueuedDir      selfstat.Stat
 	waitGroup           *sync.WaitGroup
 	acc                 telegraf.TrackingAccumulator
 	sem                 *semaphore.Weighted
@@ -174,6 +177,9 @@ func (monitor *DirectoryMonitor) Monitor() {
 
 		// We've finished reading the file and moved it away, delete it from files in use.
 		monitor.filesInUse.Delete(filePath)
+
+		// Keep track of how many files still to process
+		monitor.filesQueuedDir.Set(int64(len(monitor.filesToProcess)))
 	}
 }
 
@@ -208,6 +214,7 @@ func (monitor *DirectoryMonitor) read(filePath string) {
 	if err != nil {
 		monitor.Log.Errorf("Error while reading file: '" + filePath + "'. " + err.Error())
 		monitor.filesDropped.Incr(1)
+		monitor.filesDroppedDir.Incr(1)
 		if monitor.ErrorDirectory != "" {
 			monitor.moveFile(filePath, monitor.ErrorDirectory)
 		}
@@ -217,6 +224,7 @@ func (monitor *DirectoryMonitor) read(filePath string) {
 	// File is finished, move it to the 'finished' directory.
 	monitor.moveFile(filePath, monitor.FinishedDirectory)
 	monitor.filesProcessed.Incr(1)
+	monitor.filesProcessedDir.Incr(1)
 }
 
 func (monitor *DirectoryMonitor) ingestFile(filePath string) error {
@@ -404,8 +412,14 @@ func (monitor *DirectoryMonitor) Init() error {
 		}
 	}
 
+	tags := map[string]string{
+		"directory": monitor.Directory,
+	}
 	monitor.filesDropped = selfstat.Register("directory_monitor", "files_dropped", map[string]string{})
+	monitor.filesDroppedDir = selfstat.Register("directory_monitor", "files_dropped_per_dir", tags)
 	monitor.filesProcessed = selfstat.Register("directory_monitor", "files_processed", map[string]string{})
+	monitor.filesProcessedDir = selfstat.Register("directory_monitor", "files_processed_per_dir", tags)
+	monitor.filesQueuedDir = selfstat.Register("directory_monitor", "files_queue_per_dir", tags)
 
 	// If an error directory should be used but has not been configured yet, create one ourselves.
 	if monitor.ErrorDirectory != "" {