filestream with copytruncate strategy loses data after rotation and restart

//go:build integration

package integration

import (
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/elastic/beats/v7/libbeat/tests/integration"
)

// TestFilestreamGZIPLogRotation_2_rotations test filebeat experiencing 2 file
// rotations. It simulates 3 log files, 2 of them will be rotated:
//   - 1st: 'a' content
//   - 2nd: 'b' content
//   - 3rd: 'c' content
//
// Filebeat will "see the files" in 3 different moments, by "see" it means,
// filebeat will start with the files written to disk, read then until their end
// and then stop. While filebeat is stopped, the logs are rotated, then filebeat
// is started again.
// This test simulates the following moments:
//   - 1st: only one active log file, 1/2 of the content
//   - active file: 'a' content, only 1/2 of the logs
//   - 2nd: 1st log rotation
//   - active file: 'b' content
//   - *.1.gz file: 'a' content. Full content
//   - 3rd: 2nd log rotation
//   - active file: 'c' content
//   - *.1.gz file: 'b' content
//   - *.2.gz file: 'a' content
func TestFilestreamGZIPLogRotation_2_rotations(t *testing.T) {
	want1stRunLines := make([]string, 0, 50)
	want2ndRunLines := make([]string, 0, 150)
	want3rdRunLines := make([]string, 0, 100)

	var dataPlainA1stHalf []byte
	for i := range 50 {
		l := fmt.Sprintf("%d: 1st 1/2 aaaaaaaaaaaaaaaaaaaaaaaaa", i)
		want1stRunLines = append(want1stRunLines, l)
		dataPlainA1stHalf = append(dataPlainA1stHalf, []byte(l+"\n")...)
	}

	var dataPlainA2ndHalf []byte
	for i := range 50 {
		l := fmt.Sprintf("%d: 2nd 1/2 aaaaaaaaaaaaaaaaaaaaaaaaa", i)
		want2ndRunLines = append(want2ndRunLines, l)
		dataPlainA2ndHalf = append(dataPlainA2ndHalf, []byte(l+"\n")...)
	}

	var dataPlainB []byte
	for i := range 100 {
		l := fmt.Sprintf("%d: bbbbbbbbbbbbbbbbbbbbbbbb", i)
		want2ndRunLines = append(want2ndRunLines, l)
		dataPlainB = append(dataPlainB, []byte(l+"\n")...)
	}

	var dataPlainC []byte
	for i := range 100 {
		l := fmt.Sprintf("%d: cccccccccccccccccccccccccccc", i)
		want3rdRunLines = append(want3rdRunLines, l)
		dataPlainC = append(dataPlainC, []byte(l+"\n")...)
	}

	dataGZA := append(dataPlainA1stHalf, dataPlainA2ndHalf...)
	dataGZB := dataPlainB

	filebeat := integration.NewBeat(
		t,
		"filebeat",
		"../../filebeat.test",
	)
	tempDir := filebeat.TempDir()
	t.Log("temp dir:", tempDir)

	logFileBaseName := "plain.log"
	logPathActive := filepath.Join(tempDir, logFileBaseName)
	logPath1stRotation := filepath.Join(tempDir, logFileBaseName+".1")
	logPath2ndRotation := filepath.Join(tempDir, logFileBaseName+".2")

	outputFilePattern := "output-file"
	cfg := fmt.Sprintf(`
filebeat.inputs:
  - type: filestream
    id: "test-filestream"
    paths:
      - %s
    file_identity.native: ~
    rotation.external.strategy.copytruncate.suffix_regex: \.\d$
output.file:
  enabled: true
  path: %s
  filename: "%s"
logging.level: debug
`, logPathActive+"*", filebeat.TempDir(), outputFilePattern)
	filebeat.WriteConfigFile(cfg)

	// 1st: only one active log file, 1/2 of the content
	err := os.WriteFile(logPathActive, dataPlainA1stHalf, 0644)
	require.NoError(t, err, "could not write 'a' file to disk")

	filebeat.Start()
	eofLine := fmt.Sprintf("End of file reached: %s; Backoff now.", logPathActive)
	filebeat.WaitForLogs(
		eofLine,
		30*time.Second,
		"Filebeat did not reach EOF. Did not find log [%s]",
		eofLine,
	)
	filebeat.Stop()

	// 2nd: 1st log rotation
	// finish writing the 'a' file
	f, err := os.OpenFile(logPathActive, os.O_APPEND|os.O_WRONLY, 0644)
	require.NoError(t, err, "could not open 'a' file to append")
	_, err = f.Write(dataPlainA2ndHalf)
	require.NoError(t, err, "could not append to 'a' file")
	require.NoError(t, f.Close(), "could not close 'a' file after appending")

	// "copy" and gzip the 'a' file
	err = os.WriteFile(logPath1stRotation, dataGZA, 0644)
	require.NoError(t, err, "could not write gzipped 'a' file")

	// truncate active and write 'b' file
	err = os.WriteFile(logPathActive, dataPlainB, 0644)
	require.NoError(t, err, "could not write 'b' file")

	// at this point there is:
	// - an active file with 'b' content
	// - a '.1.gz' file with the full 'a' content

	filebeat.Start()

	waitForLatestOutput(t, outputFilePattern, tempDir, len(want2ndRunLines))
	// check the output
	files := getOutputFilesSorted(t, outputFilePattern, tempDir)
	require.Len(t, files, 2, "expected 2 output files")
	got, err := os.ReadFile(files[1])
	require.NoError(t, err, "could not open output file")
	matchPublishedLines(t, got, want2ndRunLines)
	filebeat.Stop()

	// 3rd: 2nd log rotation
	// move '.1.gz' to '.2.gz'
	err = os.Rename(logPath1stRotation, logPath2ndRotation)
	require.NoError(t, err, "could not move 'a' gzipped file")

	// "copy" and gzip the 'b' file
	err = os.WriteFile(logPath1stRotation, dataGZB, 0644)
	require.NoError(t, err, "could not write gzipped 'b' file")

	// truncate active and write 'c' file
	err = os.WriteFile(logPathActive, dataPlainC, 0644)
	require.NoError(t, err, "could not write 'c' file")

	// at this point there is:
	// - an active file with 'c' content
	// - a '.1.gz' file with 'b' content
	// - a '.2.gz' file with 'a' content

	filebeat.Start()
	waitForLatestOutput(t, outputFilePattern, tempDir, len(want3rdRunLines))
	filebeat.Stop()

	// So far so good. Now check all the output files
	files = getOutputFilesSorted(t, outputFilePattern, tempDir)

	got, err = os.ReadFile(files[0])
	require.NoError(t, err, "could not open output file")
	matchPublishedLines(t, got, want1stRunLines)

	got, err = os.ReadFile(files[1])
	require.NoError(t, err, "could not open output file")
	matchPublishedLines(t, got, want2ndRunLines)

	got, err = os.ReadFile(files[2])
	require.NoError(t, err, "could not open output file")
	matchPublishedLines(t, got, want3rdRunLines)
}

func waitForLatestOutput(t *testing.T, outputFilePattern string, tempDir string, want int) {
	// wait for all lines in the output
	msg := &strings.Builder{}
	var files []string
	condition := func() bool {
		// writeMsg is used to avoid the message being reset, the function being
		// executed, and before the function completes and writes the new
		// message, the timeout elapses and the error message is written with an
		// empty msg. Whereas it does not completely prevent this scenario, it
		// minimises it as much as possible.
		writeMsg := func(format string, a ...any) {
			msg.Reset()
			msg.WriteString(fmt.Sprintf(format, a...))
		}
		msg.Reset()
		files = getOutputFilesSorted(t, outputFilePattern, tempDir)

		got, _ := os.ReadFile(files[len(files)-1])
		lines := strings.Split(strings.TrimSuffix(string(got), "\n"), "\n")

		if len(lines) != want {
			writeMsg("want %d lines, got %d",
				want, len(lines))
			return false
		}

		return true
	}

	if !assert.Eventuallyf(t, condition, 60*time.Second, 500*time.Millisecond,
		"output file isn't what we expect: %s", msg) {
		// call the condition one last time to ensure the msg isn't reset and not
		// yet written when it's called as part of the eventually message
		condition()
		require.Failf(t, "condition never satisfied",
			"output file isn't what we expect: %s", msg)
	}
}

func getOutputFilesSorted(t *testing.T, outputFilePattern string, tempDir string) []string {
	globPattern := outputFilePattern + "-*.ndjson"
	files, err := filepath.Glob(filepath.Join(tempDir, globPattern))
	require.NoError(t, err, "could not glob output file pattern")

	slices.SortFunc(files, func(a, b string) int {
		if len(a) < len(b) {
			return -1
		}
		if len(a) > len(b) {
			return 1
		}
		if len(a) == len(b) {
			return strings.Compare(a, b)
		}

		panic("unreachable")
	})
	return files
}

func matchPublishedLines(t *testing.T, got []byte, want []string) {
	gotLinesJSON := strings.Split(strings.TrimSpace(string(got)), "\n")
	assert.Equal(t, len(want), len(gotLinesJSON), "unexpected number of events")

	gotLines := make([]string, len(gotLinesJSON))

	logLine := struct {
		Message string `json:"message"`
	}{}
	for i, line := range gotLinesJSON {
		err := json.Unmarshal([]byte(line), &logLine)
		require.NoError(t, err, "could not Unmarshal log line")
		gotLines[i] = logLine.Message
	}

	slices.Sort(gotLines)
	slices.Sort(want)

	assert.Equal(t, want, gotLines, "not all lines match")
}