feat(sftp): added mod_time metadata to sftp input

Author: anthonyvitale

internal/impl/sftp/input.go

@@ -53,6 +53,7 @@ func sftpInputSpec() *service.ConfigSpec {
 This input adds the following metadata fields to each message:
 
 - sftp_path
+- sftp_mod_time
 
 You can access these metadata fields using xref:configuration:interpolation.adoc#bloblang-queries[function interpolation].`).
 		Fields(
@@ -122,14 +123,19 @@ type sftpReader struct {
 	watcherMinAge       time.Duration
 
 	// State
-	stateLock   sync.Mutex
-	scanner     codec.DeprecatedFallbackStream
-	currentPath string
+	stateLock       sync.Mutex
+	scanner         codec.DeprecatedFallbackStream
+	currentFileInfo currentFileInfo
 
 	client       *clientPool
 	pathProvider pathProvider
 }
 
+type currentFileInfo struct {
+	path    string
+	modTime time.Time
+}
+
 func newSFTPReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (s *sftpReader, err error) {
 	s = &sftpReader{
 		log: mgr.Logger(),
@@ -224,15 +230,16 @@ func (s *sftpReader) tryReadBatch(ctx context.Context) (service.MessageBatch, se
 
 		_ = s.scanner.Close(ctx)
 		s.scanner = nil
-		s.currentPath = ""
+		s.currentFileInfo.path = ""
 		if errors.Is(err, io.EOF) {
 			err = service.ErrNotConnected
 		}
 		return nil, nil, err
 	}
 
 	for _, part := range parts {
-		part.MetaSetMut("sftp_path", s.currentPath)
+		part.MetaSetMut("sftp_path", s.currentFileInfo.path)
+		part.MetaSetMut("sftp_mod_time", s.currentFileInfo.modTime)
 	}
 
 	return parts, codecAckFn, nil
@@ -259,6 +266,11 @@ func (s *sftpReader) initScanner(ctx context.Context) (codec.DeprecatedFallbackS
 			return nil, service.ErrEndOfInput
 		}
 
+		fileInfo, err := s.client.Stat(path)
+		if err != nil {
+			return nil, fmt.Errorf("stat path: %w", err)
+		}
+
 		file, err = s.client.Open(path)
 		if err != nil {
 			s.log.With("path", path, "err", err.Error()).Warn("Unable to open previously identified file")
@@ -284,7 +296,10 @@ func (s *sftpReader) initScanner(ctx context.Context) (codec.DeprecatedFallbackS
 
 		s.stateLock.Lock()
 		s.scanner = scanner
-		s.currentPath = path
+		s.currentFileInfo = currentFileInfo{
+			path:    path,
+			modTime: fileInfo.ModTime(),
+		}
 		s.stateLock.Unlock()
 		return scanner, nil
 	}
@@ -335,7 +350,7 @@ func (s *sftpReader) closeScanner(ctx context.Context) {
 			s.log.With("error", err).Error("Failed to close scanner")
 		}
 		s.scanner = nil
-		s.currentPath = ""
+		s.currentFileInfo.path = ""
 	}
 }
 

internal/impl/sftp/integration_test.go

@@ -161,6 +161,10 @@ cache_resources:
 		if !ok {
 			return errors.New("sftp_path metadata not found")
 		}
+		_, ok = msg.MetaGet("sftp_mod_time")
+		if !ok {
+			return errors.New("sftp_mod_time metadata not found")
+		}
 		receivedPaths = append(receivedPaths, path)
 		return nil
 	}))