Reset file offset when re-reading from the beginning (#24414) (#24416)

(cherry picked from commit 1fd8c4a)

CHANGELOG.next.asciidoc

@@ -157,6 +157,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - system/package: Fix an error that can occur while trying to persist package metadata. {issue}18536[18536] {pull}18887[18887]
 - system/socket: Fix dataset using 100% CPU and becoming unresponsive in some scenarios. {pull}19033[19033] {pull}19764[19764]
 - system/socket: Fixed tracking of long-running connections. {pull}19033[19033]
+- system/login: Fixed offset reset on inode reuse. {pull}24414[24414]
 
 *Filebeat*
 

x-pack/auditbeat/module/system/login/login_test.go

@@ -101,6 +101,13 @@ func TestWtmp(t *testing.T) {
 		t.Fatalf("error opening %v: %v", wtmpFilepath, err)
 	}
 
+	wtmpFileInfo, err := os.Stat(wtmpFilepath)
+	if err != nil {
+		t.Fatalf("error performing stat on %v: %v", wtmpFilepath, err)
+	}
+
+	size := wtmpFileInfo.Size()
+
 	loginUtmp := utmpC{
 		Type: DEAD_PROCESS,
 	}
@@ -132,6 +139,35 @@ func TestWtmp(t *testing.T) {
 	checkFieldValue(t, events[0].RootFields, "user.name", "vagrant")
 	checkFieldValue(t, events[0].RootFields, "related.user", []string{"vagrant"})
 	checkFieldValue(t, events[0].RootFields, "user.terminal", "pts/2")
+
+	// We truncate to the previous size to force a full re-read, simulating an inode reuse.
+	if err := wtmpFile.Truncate(size); err != nil {
+		t.Fatalf("error truncating %v: %v", wtmpFilepath, err)
+	}
+
+	events, errs = mbtest.ReportingFetchV2(f)
+	if len(errs) > 0 {
+		t.Fatalf("received error: %+v", errs[0])
+	}
+
+	if len(events) == 0 {
+		t.Fatal("no events were generated")
+	} else if len(events) != 1 {
+		t.Fatalf("only one event expected, got %d", len(events))
+	}
+
+	// utmpdump: [7] [14962] [ts/2] [vagrant ] [pts/2       ] [10.0.2.2            ] [10.0.2.2       ] [2019-01-24T09:51:51,367964+00:00]
+	checkFieldValue(t, events[0].RootFields, "event.kind", "event")
+	checkFieldValue(t, events[0].RootFields, "event.category", []string{"authentication"})
+	checkFieldValue(t, events[0].RootFields, "event.type", []string{"start", "authentication_success"})
+	checkFieldValue(t, events[0].RootFields, "event.action", "user_login")
+	checkFieldValue(t, events[0].RootFields, "event.outcome", "success")
+	checkFieldValue(t, events[0].RootFields, "process.pid", 14962)
+	checkFieldValue(t, events[0].RootFields, "source.ip", "10.0.2.2")
+	checkFieldValue(t, events[0].RootFields, "user.name", "vagrant")
+	checkFieldValue(t, events[0].RootFields, "user.terminal", "pts/2")
+	assert.True(t, events[0].Timestamp.Equal(time.Date(2019, 1, 24, 9, 51, 51, 367964000, time.UTC)),
+		"Timestamp is not equal: %+v", events[0].Timestamp)
 }
 
 func TestBtmp(t *testing.T) {

x-pack/auditbeat/module/system/login/utmp.go

@@ -219,11 +219,18 @@ func (r *UtmpFileReader) readNewInFile(loginRecordC chan<- LoginRecord, errorC c
 			f.Close()
 		}()
 
-		_, err = f.Seek(utmpFile.Offset, 0)
-		if err != nil {
-			errorC <- errors.Wrapf(err, "error setting offset for file %v", utmpFile.Path)
+		// This will be the usual case, but we do not want to seek with the stored offset
+		// if the saved size is smaller than the current one.
+		if size >= oldSize {
+			_, err = f.Seek(utmpFile.Offset, 0)
+			if err != nil {
+				errorC <- errors.Wrapf(err, "error setting offset %d for file %v", utmpFile.Offset, utmpFile.Path)
+			}
+		}
 
-			// Try one more time, this time resetting to the beginning of the file.
+		// If the saved size is smaller than the current one, or the previous Seek failed,
+		// we retry one more time, this time resetting to the beginning of the file.
+		if size < oldSize || err != nil {
 			_, err = f.Seek(0, 0)
 			if err != nil {
 				errorC <- errors.Wrapf(err, "error setting offset 0 for file %v", utmpFile.Path)