[8.0](backport #30445) Set log.offset to the start of the reported line in filestream

CHANGELOG.next.asciidoc

@@ -47,10 +47,13 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 - Fix handling of execve call events which have no argument. {issue}30585[30585] {pull}30586[30586]
 
 *Filebeat*
-
+- Fix broken Kafka input {issue}29746[29746] {pull}30277[30277]
+- cisco module: Fix change the broke ASA and FTD configs that used `var.input: syslog`. {pull}30072[30072]
+- aws-s3: fix race condition in states used by s3-poller. {issue}30123[30123] {pull}30131[30131]
 - auditd: Prevent mapping explosion when truncated EXECVE records are ingested. {pull}30382[30382]
 - elasticsearch: fix duplicate ingest when using a common appender configuration {issue}30428[30428] {pull}30440[30440]
 - Set `url` as a pointer in the `httpjson` template context to ensure access to all methods. {pull}28695[28695]
+- Report the starting offset of the line in `log.offset` when using `filestream` instead of the end to be ECS compliant. {pull}30445[30445]
 
 *Heartbeat*
 

libbeat/reader/readfile/metafields.go

@@ -41,10 +41,9 @@ func NewFilemeta(r reader.Reader, path string, offset int64) reader.Reader {
 func (r *FileMetaReader) Next() (reader.Message, error) {
 	message, err := r.reader.Next()
 
-	r.offset += int64(message.Bytes)
-
 	// if the message is empty, there is no need to enrich it with file metadata
 	if message.IsEmpty() {
+		r.offset += int64(message.Bytes)
 		return message, err
 	}
 
@@ -56,6 +55,9 @@ func (r *FileMetaReader) Next() (reader.Message, error) {
 			},
 		},
 	})
+
+	r.offset += int64(message.Bytes)
+
 	return message, err
 }
 

libbeat/reader/readfile/metafields_test.go

@@ -54,7 +54,6 @@ func TestMetaFieldsOffset(t *testing.T) {
 		if err == io.EOF {
 			break
 		}
-		offset += int64(msg.Bytes)
 
 		expectedFields := common.MapStr{}
 		if len(msg.Content) != 0 {
@@ -67,6 +66,8 @@ func TestMetaFieldsOffset(t *testing.T) {
 				},
 			}
 		}
+		offset += int64(msg.Bytes)
+
 		require.Equal(t, expectedFields, msg.Fields)
 		require.Equal(t, offset, in.offset)
 	}