From 9e29a857f0bfae1ea154f4b4bdaa6fc89b02d71d Mon Sep 17 00:00:00 2001 From: Lee E Hinman <57081003+leehinman@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:50:51 -0500 Subject: [PATCH] make statestore creation of meta.json atomic (#51897) * make statestore creation of meta.json atomic * add changelog * fix linter issues * address scenario that Anderson caught * fix linter issue (cherry picked from commit 052272d405973e4dec92404dadf95626d967d5d7) --- ...43-statestore-atomic-create-meta.json.yaml | 7 +++ .../statestore/backend/memlog/diskstore.go | 41 ++++++------ libbeat/statestore/backend/memlog/store.go | 13 +++- .../statestore/backend/memlog/store_test.go | 63 +++++++++++++++++++ 4 files changed, 104 insertions(+), 20 deletions(-) create mode 100644 changelog/fragments/1783957443-statestore-atomic-create-meta.json.yaml diff --git a/changelog/fragments/1783957443-statestore-atomic-create-meta.json.yaml b/changelog/fragments/1783957443-statestore-atomic-create-meta.json.yaml new file mode 100644 index 000000000000..327c96d3cd3e --- /dev/null +++ b/changelog/fragments/1783957443-statestore-atomic-create-meta.json.yaml @@ -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 diff --git a/libbeat/statestore/backend/memlog/diskstore.go b/libbeat/statestore/backend/memlog/diskstore.go index 39ac216b0d40..aa0890e5039b 100644 --- a/libbeat/statestore/backend/memlog/diskstore.go +++ b/libbeat/statestore/backend/memlog/diskstore.go @@ -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" ) @@ -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 @@ -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) @@ -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 } diff --git a/libbeat/statestore/backend/memlog/store.go b/libbeat/statestore/backend/memlog/store.go index 164c1024f456..ec99733541e6 100644 --- a/libbeat/statestore/backend/memlog/store.go +++ b/libbeat/statestore/backend/memlog/store.go @@ -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 } @@ -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) } } diff --git a/libbeat/statestore/backend/memlog/store_test.go b/libbeat/statestore/backend/memlog/store_test.go index 9c7fe86a667f..3feb0e0db988 100644 --- a/libbeat/statestore/backend/memlog/store_test.go +++ b/libbeat/statestore/backend/memlog/store_test.go @@ -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()