Skip to content

wip: refactor test registry helpers #48315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
AndersonQ wants to merge 1 commit into from
Closed

wip: refactor test registry helpers #48315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions filebeat/testing/integration/helper.go → filebeat/testhelpers/helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.

package integration
// This file was contributed to by generative AI

package testhelpers

import (
"bufio"
Expand All @@ -32,7 +34,7 @@ import (
// false is returned. It will fail the test on any error reading/parsing
// the registry file.
func AssertLastOffset(t *testing.T, path string, offset int) bool {
entries, _ := readFilestreamRegistryLog(t, path)
entries, _ := ReadFilestreamRegistryLog(t, path)
lastEntry := entries[len(entries)-1]
if lastEntry.Offset != offset {
t.Errorf("expecting offset %d got %d instead", offset, lastEntry.Offset)
Expand All @@ -53,7 +55,7 @@ func AssertLastOffset(t *testing.T, path string, offset int) bool {
return true
}

type registryEntry struct {
type RegistryEntry struct {
Key string
Offset int
EOF bool
Expand All @@ -63,13 +65,14 @@ type registryEntry struct {
Removed bool
}

func readFilestreamRegistryLog(t *testing.T, path string) ([]registryEntry, map[string]string) {
func ReadFilestreamRegistryLog(t *testing.T, path string) ([]RegistryEntry, map[string]string) {
file, err := os.Open(path)
if err != nil {
t.Fatalf("could not open file '%s': %s", path, err)
}
defer file.Close()

var entries []registryEntry
var entries []RegistryEntry
fileNameToNative := map[string]string{}
s := bufio.NewScanner(file)

Expand All @@ -89,7 +92,7 @@ func readFilestreamRegistryLog(t *testing.T, path string) ([]registryEntry, map[
continue
}
// Filestream entry
et := registryEntry{
et := RegistryEntry{
Key: e.Key,
Offset: e.Value.Cursor.Offset,
EOF: e.Value.Cursor.EOF,
Expand Down
5 changes: 4 additions & 1 deletion filebeat/testing/integration/container_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.

// This file was contributed to by generative AI

//go:build integration

package integration
Expand All @@ -29,6 +31,7 @@ import (

"github.com/stretchr/testify/require"

"github.com/elastic/beats/v7/filebeat/testhelpers"
"github.com/elastic/beats/v7/libbeat/testing/integration"
"github.com/elastic/elastic-agent-libs/mapstr"
)
Expand Down Expand Up @@ -133,7 +136,7 @@ queue.mem.flush.timeout: 0s
require.Eventually(
t,
func() bool {
return AssertLastOffset(t, registryLogFile, 2243)
return testhelpers.AssertLastOffset(t, registryLogFile, 2243)
},
20*time.Second,
250*time.Millisecond, "did not find the expected registry offset of 2243")
Expand Down
13 changes: 8 additions & 5 deletions filebeat/tests/integration/filestream_gzip_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.

// This file was contributed to by generative AI

//go:build integration

package integration
Expand All @@ -34,6 +36,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

"github.com/elastic/beats/v7/filebeat/testhelpers"
"github.com/elastic/beats/v7/filebeat/testing/gziptest"
"github.com/elastic/beats/v7/libbeat/tests/integration"
"github.com/elastic/elastic-agent-libs/iobuf"
Expand Down Expand Up @@ -460,9 +463,9 @@ logging.level: debug

registryLogFile := filepath.Join(workDir,
"data", "registry", "filebeat", "log.json")
entries, _ := readFilestreamRegistryLog(t, registryLogFile)
entries, _ := testhelpers.ReadFilestreamRegistryLog(t, registryLogFile)

var lastEntry *registryEntry
var lastEntry *testhelpers.RegistryEntry
for i := range entries {
entry := &entries[i]
if entry.Filename == logFilepath {
Expand All @@ -486,7 +489,7 @@ logging.level: debug
)

// =============== Verify file read to EOF isn't read again ================
gotEntries, _ := readFilestreamRegistryLog(t, registryLogFile)
gotEntries, _ := testhelpers.ReadFilestreamRegistryLog(t, registryLogFile)
// when the harvester starts, before attempting to open the log file, it
// updates the registry, thus reading it again will bring one more entry
assert.Equal(t, entries, gotEntries[:len(gotEntries)-1],
Expand Down Expand Up @@ -1336,8 +1339,8 @@ logging.level: debug
// assert EOF is set on registry
registryLogFile := filepath.Join(workDir,
"data", "registry", "filebeat", "log.json")
entries, _ := readFilestreamRegistryLog(t, registryLogFile)
var lastEntry *registryEntry
entries, _ := testhelpers.ReadFilestreamRegistryLog(t, registryLogFile)
var lastEntry *testhelpers.RegistryEntry
for i := range entries {
entry := &entries[i]
if entry.Filename == logPath {
Expand Down
15 changes: 9 additions & 6 deletions filebeat/tests/integration/filestream_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.

// This file was contributed to by generative AI

//go:build integration

package integration
Expand All @@ -35,6 +37,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

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

Expand Down Expand Up @@ -701,8 +704,8 @@ func requireRegistryEntryRemoved(t *testing.T, workDir, identity string) {
t.Helper()

registryFile := filepath.Join(workDir, "data", "registry", "filebeat", "log.json")
entries, _ := readFilestreamRegistryLog(t, registryFile)
inputEntries := []registryEntry{}
entries, _ := testhelpers.ReadFilestreamRegistryLog(t, registryFile)
inputEntries := []testhelpers.RegistryEntry{}
for _, currentEntry := range entries {
if strings.Contains(currentEntry.Key, identity) {
inputEntries = append(inputEntries, currentEntry)
Expand Down Expand Up @@ -757,8 +760,8 @@ func createFileAndWaitIngestion(
requirePublishedEvents(t, fb, outputTotal, outputFilepath)
}

func parseRegistry(entries []registryEntry) map[string]registryEntry {
registry := map[string]registryEntry{}
func parseRegistry(entries []testhelpers.RegistryEntry) map[string]testhelpers.RegistryEntry {
registry := map[string]testhelpers.RegistryEntry{}

for _, e := range entries {
switch e.Op {
Expand Down Expand Up @@ -791,13 +794,13 @@ func assertRegistry(t *testing.T, workDir, testdataDir, registry, msg string) {
data = bytes.ReplaceAll(data, []byte(`/`), []byte(`\`))
data = bytes.ReplaceAll(data, []byte(`\`), []byte(`\\`))
}
expectedRegistry := map[string]registryEntry{}
expectedRegistry := map[string]testhelpers.RegistryEntry{}
if err := json.Unmarshal(data, &expectedRegistry); err != nil {
t.Fatalf("cannot unmarshal expected registry file: %s", err)
}

registryFile := filepath.Join(workDir, "data", "registry", "filebeat", "log.json")
entries, nameToInode := readFilestreamRegistryLog(t, registryFile)
entries, nameToInode := testhelpers.ReadFilestreamRegistryLog(t, registryFile)
reg := parseRegistry(entries)

// More Windows workarounds.
Expand Down
128 changes: 7 additions & 121 deletions filebeat/tests/integration/filestream_truncation_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,21 @@
// specific language governing permissions and limitations
// under the License.

// This file was contributed to by generative AI

//go:build integration

package integration

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

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

Expand Down Expand Up @@ -99,7 +99,7 @@ func TestFilestreamLiveFileTruncation(t *testing.T) {
filebeat.Stop()

// Assert we offset in the registry
assertLastOffset(t, registryLogFile, 10_000)
testhelpers.AssertLastOffset(t, registryLogFile, 10_000)

// Open for appending because the file has already been truncated
integration.WriteLogFile(t, logFile, 10, true)
Expand All @@ -109,7 +109,7 @@ func TestFilestreamLiveFileTruncation(t *testing.T) {
filebeat.WaitLogsContains("End of file reached", 30*time.Second, "Filebeat did not finish reading the log file")
filebeat.WaitLogsContains("End of file reached", 30*time.Second, "Filebeat did not finish reading the log file")

assertLastOffset(t, registryLogFile, 500)
testhelpers.AssertLastOffset(t, registryLogFile, 500)
}

func TestFilestreamOfflineFileTruncation(t *testing.T) {
Expand All @@ -134,7 +134,7 @@ func TestFilestreamOfflineFileTruncation(t *testing.T) {
filebeat.Stop()

// 3. Assert the offset is correctly set in the registry
assertLastOffset(t, registryLogFile, 500)
testhelpers.AssertLastOffset(t, registryLogFile, 500)

// 4. Truncate the file and write some data (less than before)
if err := os.Truncate(logFile, 0); err != nil {
Expand All @@ -149,119 +149,5 @@ func TestFilestreamOfflineFileTruncation(t *testing.T) {
filebeat.Stop()

// 6. Assert the registry offset is new, smaller file size.
assertLastOffset(t, registryLogFile, 250)
}

func assertLastOffset(t *testing.T, path string, offset int) {
t.Helper()
entries, _ := readFilestreamRegistryLog(t, path)
lastEntry := entries[len(entries)-1]
if lastEntry.Offset != offset {
t.Errorf("expecting offset %d got %d instead", offset, lastEntry.Offset)
t.Log("last registry entries:")

max := len(entries)
if max > 10 {
max = 10
}
for _, e := range entries[:max] {
t.Logf("%+v\n", e)
}

t.FailNow()
}
}

type registryEntry struct {
Key string
Offset int
EOF bool
Filename string
TTL time.Duration
Op string
Removed bool
}

func readFilestreamRegistryLog(t *testing.T, path string) ([]registryEntry, map[string]string) {
file, err := os.Open(path)
if err != nil {
t.Fatalf("could not open file '%s': %s", path, err)
}

var entries []registryEntry
fileNameToNative := map[string]string{}
s := bufio.NewScanner(file)

var lastOperation string
for s.Scan() {
line := s.Bytes()

e := entry{}
if err := json.Unmarshal(line, &e); err != nil {
t.Fatalf("could not read line '%s': %s", string(line), err)
}

// Skips registry log entries containing the operation ID like:
// '{"op":"set","id":46}'
if e.Key == "" {
lastOperation = e.Op
continue
}
// Filestream entry
et := registryEntry{
Key: e.Key,
Offset: e.Value.Cursor.Offset,
EOF: e.Value.Cursor.EOF,
TTL: e.Value.TTL,
Filename: e.Value.Meta.Source,
Removed: lastOperation == "remove",
Op: lastOperation,
}

// Handle the log input entries, they have a different format.
if strings.HasPrefix(e.Key, "filebeat::logs") {
et.Offset = e.Value.Offset
et.Filename = e.Value.Source

if lastOperation != "set" {
continue
}

// Extract the native file identity so we can update the
// expected registry accordingly
name := filepath.Base(et.Filename)
id := strings.Join(strings.Split(et.Key, "::")[2:], "::")
fileNameToNative[name] = id
}

entries = append(entries, et)
}

return entries, fileNameToNative
}

type entry struct {
Key string `json:"k"`
Value struct {
// Filestream fields
Cursor struct {
Offset int `json:"offset"`
EOF bool `json:"eof"`
} `json:"cursor"`
Meta struct {
Source string `json:"source"`
} `json:"meta"`

// Log input fields
Source string `json:"source"`
Offset int `json:"offset"`

// Common to both inputs
TTL time.Duration `json:"ttl"`
} `json:"v"`

// Keys to read the "operation"
// e.g: {"op":"set","id":46}
Op string `json:"op"`
ID int `json:"id"`
testhelpers.AssertLastOffset(t, registryLogFile, 250)
}
Loading