fix after review

Author: DmitryRomanov

plugin/input/file/offset.go

@@ -220,18 +220,7 @@ func (o *offsetDB) parseOptionalLine(content string, prefix string) (string, str
 		return o.parseLine(content, prefix)
 	}
 
-	if strings.HasPrefix(content, "  streams:") || content[0] == '-' {
-		return "", content, nil
-	}
-
-	linePos := strings.IndexByte(content, '\n')
-	var nextLine string
-	if linePos >= 0 {
-		nextLine = content[:linePos]
-	} else {
-		nextLine = content
-	}
-	return "", "", fmt.Errorf("unexpected line format, expected %q or streams section, got %q", prefix, safeSubstring(nextLine, len(prefix)))
+	return "", content, nil
 }
 
 func safeSubstring(s string, length int) string {

plugin/input/file/offset_test.go

@@ -387,32 +387,6 @@ func TestParseOptionalLineLastReadTimestamp(t *testing.T) {
 			expectedRemain: "",
 			desc:           "empty content",
 		},
-
-		// Error cases
-		{
-			name:        "unexpected line before streams",
-			content:     "  unknown_field: value\n  streams:",
-			expectedErr: true,
-			desc:        "unexpected line causes error",
-		},
-		{
-			name:        "wrong field name",
-			content:     "  lastreadtimestamp: 1763651665000000000\n  streams:",
-			expectedErr: true,
-			desc:        "wrong field name (no underscores)",
-		},
-		{
-			name:        "case mismatch",
-			content:     "  Last_Read_Timestamp: 1763651665000000000\n  streams:",
-			expectedErr: true,
-			desc:        "wrong case in field name",
-		},
-		{
-			name:        "missing colon",
-			content:     "  last_read_timestamp 1763651665000000000\n  streams:",
-			expectedErr: true,
-			desc:        "missing colon after field name",
-		},
 	}
 
 	for _, tt := range tests {

plugin/input/file/provider.go

@@ -9,8 +9,6 @@ import (
 	"syscall"
 	"time"
 
-	sync_atomic "sync/atomic"
-
 	"github.com/ozontech/file.d/logger"
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/xtime"
@@ -104,37 +102,38 @@ func (j *Job) seek(offset int64, whence int, hint string) (n int64) {
 	return n
 }
 
+// eofInfo tracks end-of-file state for read operations.
+// Used to determine when files can be safely removed after processing.
 type eofInfo struct {
-	timestamp int64
-	offset    int64
+	timestamp atomic.Int64
+	offset    atomic.Int64
 }
 
 func (e *eofInfo) setTimestamp(t time.Time) {
-	sync_atomic.StoreInt64(&e.timestamp, t.UnixNano())
+	e.timestamp.Store(t.UnixNano())
 }
 
 func (e *eofInfo) getTimestamp() time.Time {
-	nanos := sync_atomic.LoadInt64(&e.timestamp)
+	nanos := e.timestamp.Load()
 	return time.Unix(0, nanos)
 }
 
 // setUnixNanoTimestamp sets timestamp in seconds
 func (e *eofInfo) setUnixNanoTimestamp(nanos int64) {
-	sync_atomic.StoreInt64(&e.timestamp, nanos)
+	e.timestamp.Store(nanos)
 }
 
 // getUnixNanoTimestamp returns timestamp in seconds
 func (e *eofInfo) getUnixNanoTimestamp() int64 {
-	nanos := sync_atomic.LoadInt64(&e.timestamp)
-	return nanos
+	return e.timestamp.Load()
 }
 
 func (e *eofInfo) setOffset(offset int64) {
-	sync_atomic.StoreInt64(&e.offset, offset)
+	e.offset.Store(offset)
 }
 
 func (e *eofInfo) getOffset() int64 {
-	return sync_atomic.LoadInt64(&e.offset)
+	return e.offset.Load()
 }
 
 type inodeID uint64

plugin/input/file/provider_test.go

@@ -286,5 +286,4 @@ func TestEOFInfo(t *testing.T) {
 		gotTime := e.getTimestamp()
 		assert.Equal(t, newUnixTs, gotTime.UnixNano())
 	})
-
 }

plugin/input/file/worker.go

@@ -100,7 +100,7 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 		if job.mimeType == "application/x-lz4" {
 			if isNotFileBeingWritten(file.Name()) {
 				// lz4 does not support appending, so we check that no one is writing to the file
-				logger.Error("cannot lock file", zap.String("filename", file.Name()))
+				logger.Error("cannot process incomplete file. write in progress", zap.String("filename", file.Name()))
 				break
 			}
 			lz4Reader := lz4.NewReader(file)
@@ -132,14 +132,14 @@ func (w *worker) work(controller inputer, jobProvider *jobProvider, readBufferSi
 		for {
 			n, err := reader.Read(readBuf)
 			controller.IncReadOps()
-			// if we read to end of file it's time to check truncation etc and process next job
-			if (!job.isCompressed && err == io.EOF) || n == 0 {
-				// cause lz4reader can return EOF and n > 0
-				isEOFReached = true
-				break
-			}
 			if err != nil {
-				if !job.isCompressed && err != io.EOF {
+				if (!job.isCompressed && err == io.EOF) || n == 0 {
+					// cause lz4reader can return EOF and n > 0
+					isEOFReached = true
+					break
+				}
+				if !(job.isCompressed && (err == io.EOF || err == io.ErrUnexpectedEOF)) {
+					// except EOF-errors on compressed file
 					logger.Fatalf("file %d:%s read error, %s read=%d", sourceID, sourceName, err.Error(), n)
 				}
 			}