Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
kind: bug-fix

summary: Prevent statestore startup failure when meta.json is left empty after an unclean shutdown

component: libbeat

pr: https://github.com/elastic/beats/pull/51897
41 changes: 23 additions & 18 deletions libbeat/statestore/backend/memlog/diskstore.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (
"strconv"

"github.com/elastic/beats/v7/libbeat/common/cleanup"
agentfile "github.com/elastic/elastic-agent-libs/file"
"github.com/elastic/elastic-agent-libs/logp"
"github.com/elastic/elastic-agent-libs/mapstr"
)
Expand Down Expand Up @@ -195,7 +196,7 @@ func (s *diskstore) tryOpenLog() error {
return err
}

s.logFileSize = uint64(info.Size())
s.logFileSize = uint64(info.Size()) //nolint:gosec // G115: info.Size() is non-negative for a regular file
}

ok = true
Expand Down Expand Up @@ -640,31 +641,35 @@ func checkMeta(meta storeMeta) error {

func writeMetaFile(home string, mode os.FileMode) error {
path := filepath.Join(home, metaFileName)
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_TRUNC, mode)
f, err := os.CreateTemp(home, metaFileName+".tmp-*")
if err != nil {
return err
return fmt.Errorf("error creating tmp meta file: %w", err)
}
tmpName := f.Name()
defer func() {
_ = f.Close()
_ = os.Remove(tmpName)
}()

ok := false
defer cleanup.IfNot(&ok, func() {
f.Close()
})
if err = f.Chmod(mode); err != nil {
return fmt.Errorf("error setting tmp meta file permissions: %w", err)
}

enc := newJSONEncoder(f)
err = enc.Encode(storeMeta{
Version: storeVersion,
})
if err != nil {
return err
if err = enc.Encode(storeMeta{Version: storeVersion}); err != nil {
return fmt.Errorf("error encoding meta file: %w", err)
}

if err := f.Sync(); err != nil {
return err
if err = f.Sync(); err != nil {
return fmt.Errorf("error syncing meta file: %w", err)
}

ok = true
if err := f.Close(); err != nil {
return err
if err = f.Close(); err != nil {
return fmt.Errorf("error closing tmp meta file: %w", err)
}

if err = agentfile.SafeFileRotate(path, tmpName); err != nil {
return fmt.Errorf("error rotating meta file into place: %w", err)
}

trySyncPath(home)
Expand Down Expand Up @@ -695,5 +700,5 @@ func readMetaFile(home string) (storeMeta, error) {
// - a < b (mod 2^63)
// - b > a after an integer rollover that is still within the distance of <2^63-1
func isTxIDLessEqual(a, b uint64) bool {
return int64(a-b) <= 0
return a == b || a-b >= 1<<63
}
13 changes: 11 additions & 2 deletions libbeat/statestore/backend/memlog/store.go
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ type entry struct {
func openStore(log *logp.Logger, home string, mode os.FileMode, bufSz uint, ignoreVersionCheck bool, checkpoint CheckpointPredicate) (*store, error) {
fi, err := os.Stat(home)
if os.IsNotExist(err) {
err = os.MkdirAll(home, os.ModeDir|0770)
err = os.MkdirAll(home, os.ModeDir|0o770)
if err != nil {
return nil, err
}
Expand All @@ -75,7 +75,16 @@ func openStore(log *logp.Logger, home string, mode os.FileMode, bufSz uint, igno
} else if !fi.Mode().IsDir() {
return nil, fmt.Errorf("'%v' is not a directory", home)
} else {
if err := pathEnsurePermissions(filepath.Join(home, metaFileName), mode); err != nil {
metaPath := filepath.Join(home, metaFileName)
if _, err := os.Stat(metaPath); os.IsNotExist(err) {
// Directory exists but meta.json is absent — the process likely crashed
// after MkdirAll but before writeMetaFile completed on a prior run.
if err := writeMetaFile(home, mode); err != nil {
return nil, err
}
} else if err != nil {
return nil, fmt.Errorf("failed to stat meta file: %w", err)
} else if err := pathEnsurePermissions(metaPath, mode); err != nil {
return nil, fmt.Errorf("failed to update meta file permissions: %w", err)
}
}
Expand Down
63 changes: 63 additions & 0 deletions libbeat/statestore/backend/memlog/store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,75 @@ import (
"path/filepath"
"testing"

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

"github.com/elastic/elastic-agent-libs/logp/logptest"
"github.com/elastic/elastic-agent-libs/mapstr"
)

func TestWriteMetaFileAtomic(t *testing.T) {
t.Run("writes_valid_meta_and_cleans_up_temp", func(t *testing.T) {
dir := t.TempDir()

require.NoError(t, writeMetaFile(dir, 0o600))

meta, err := readMetaFile(dir)
require.NoError(t, err)
require.NoError(t, checkMeta(meta))

matches, err := filepath.Glob(filepath.Join(dir, "*.tmp-*"))
require.NoError(t, err)
assert.Empty(t, matches, "no temp files should remain after successful write")
})

t.Run("no_partial_file_on_rotate_failure", func(t *testing.T) {
dir := t.TempDir()
// Place a directory at the meta.json path so SafeFileRotate (rename) fails.
// This simulates a crash-safe scenario: the destination is never touched
// until the temp file is fully written and synced.
metaPath := filepath.Join(dir, metaFileName)
require.NoError(t, os.Mkdir(metaPath, 0o700))

err := writeMetaFile(dir, 0o600)
require.Error(t, err)

// The directory at the destination must be intact (not replaced or emptied).
fi, statErr := os.Stat(metaPath)
require.NoError(t, statErr)
assert.True(t, fi.IsDir(), "destination should still be a directory after failed rotate")

// No temp files should remain.
matches, globErr := filepath.Glob(filepath.Join(dir, "*.tmp-*"))
require.NoError(t, globErr)
assert.Empty(t, matches, "no temp files should remain after failed rotate")
})
}

// TestOpenStoreRecoversMissingMeta verifies that openStore creates meta.json when
// the store directory already exists but meta.json is absent — the scenario that
// arises when the process crashes after MkdirAll but before writeMetaFile on a
// prior run.
func TestOpenStoreRecoversMissingMeta(t *testing.T) {
dir := t.TempDir()
// Simulate a crash after MkdirAll: the directory exists but is empty.
storePath := filepath.Join(dir, "store")
require.NoError(t, os.Mkdir(storePath, 0o770))

logger := logptest.NewTestingLogger(t, "")
store, err := openStore(logger.Named("test"), storePath, 0o660, 4096, false, func(_ uint64) bool {
return false
})
require.NoError(t, err, "openStore must succeed even when meta.json was missing")
store.Close()

require.FileExists(t, filepath.Join(storePath, metaFileName), "meta.json must be created on recovery")

meta, err := readMetaFile(storePath)
require.NoError(t, err)
assert.NoError(t, checkMeta(meta))
}

func TestRecoverFromCorruption(t *testing.T) {
path := t.TempDir()

Expand Down
Loading