make creation of meta.json file atomic (#51791)

* make creation of meta.json file atomic

* improve changelog summary

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: leehinman

changelog/fragments/1783441656-make-creation-of-meta.json-atomic.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Prevent Filebeat startup failure when meta.json is left empty after migration
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/beats/pull/51791
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

filebeat/registrar/migrate.go

@@ -241,7 +241,7 @@ func (m *Migrator) updateToVersion1(regHome string) error {
 		return fmt.Errorf("migration complete but failed to remove original data file: %v: %w", origDataFile, err)
 	}
 
-	if err := os.WriteFile(filepath.Join(regHome, "meta.json"), []byte(`{"version": "1"}`), m.permissions); err != nil {
+	if err := safeWriteFile(filepath.Join(regHome, "meta.json"), []byte(`{"version": "1"}`), m.permissions); err != nil {
 		return fmt.Errorf("failed to update the meta.json file: %w", err)
 	}
 
@@ -290,29 +290,38 @@ func isFile(path string, log *logp.Logger) bool {
 }
 
 func safeWriteFile(path string, data []byte, perm os.FileMode) error {
-	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
+	f, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".tmp-*")
 	if err != nil {
-		return err
+		return fmt.Errorf("error creating tmp file: %w", err)
+	}
+	tmpName := f.Name()
+	// Always close and remove the tmp file on error path, errors will be ignored
+	// in the success case
+	defer func() {
+		_ = f.Close()
+		_ = os.Remove(tmpName)
+	}()
+
+	if err = f.Chmod(perm); err != nil {
+		return fmt.Errorf("error setting tmp file permissions: %w", err)
 	}
 
-	for len(data) > 0 {
-		var n int
-		n, err = f.Write(data)
-		if err != nil {
-			break
-		}
+	if _, err = f.Write(data); err != nil {
+		return fmt.Errorf("error writing to tmp file: %w", err)
+	}
 
-		data = data[n:]
+	if err = f.Sync(); err != nil {
+		return fmt.Errorf("error syncing data to tmp file: %w", err)
 	}
 
-	if err == nil {
-		err = f.Sync()
+	if err = f.Close(); err != nil {
+		return fmt.Errorf("error closing tmp file: %w", err)
 	}
 
-	if err1 := f.Close(); err == nil {
-		err = err1
+	if err = helper.SafeFileRotate(path, tmpName); err != nil {
+		return fmt.Errorf("error moving tmp file to target: %w", err)
 	}
-	return err
+	return nil
 }
 
 // fixStates cleans up the registry states when updating from an older version

filebeat/registrar/migrate_test.go

@@ -0,0 +1,82 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux || darwin
+
+package registrar
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSafeWriteFile(t *testing.T) {
+	data := []byte(`{"version": "1"}`)
+	perm := os.FileMode(0o600)
+
+	t.Run("success", func(t *testing.T) {
+		dir := tempDir(t)
+		path := filepath.Join(dir, "meta.json")
+
+		require.NoError(t, safeWriteFile(path, data, perm))
+
+		got, err := os.ReadFile(path)
+		require.NoError(t, err)
+		assert.Equal(t, data, got)
+
+		fi, err := os.Stat(path)
+		require.NoError(t, err)
+		assert.Equal(t, perm, fi.Mode().Perm(), "file permissions")
+
+		assertNoTempFiles(t, dir)
+	})
+
+	t.Run("parent_dir_missing", func(t *testing.T) {
+		// os.CreateTemp fails because the directory does not exist.
+		path := filepath.Join(tempDir(t), "nonexistent", "meta.json")
+
+		err := safeWriteFile(path, data, perm)
+		require.Error(t, err)
+	})
+
+	t.Run("rotate_fails_cleans_up_temp_file", func(t *testing.T) {
+		dir := tempDir(t)
+		// Place a directory at the target path so os.Rename returns EISDIR,
+		// forcing SafeFileRotate to fail after the temp file has been written.
+		path := filepath.Join(dir, "meta.json")
+		mkDir(t, path)
+
+		err := safeWriteFile(path, data, perm)
+		require.Error(t, err)
+
+		// The temp file must not be left behind despite the rotate failure.
+		assertNoTempFiles(t, dir)
+	})
+}
+
+// assertNoTempFiles fails the test if any temp files from safeWriteFile remain
+// in dir. The naming pattern is <base>.tmp-<random>.
+func assertNoTempFiles(t *testing.T, dir string) {
+	t.Helper()
+	matches, err := filepath.Glob(filepath.Join(dir, "*.tmp-*"))
+	require.NoError(t, err)
+	assert.Empty(t, matches, "unexpected temp files in %s", dir)
+}