fix: add_cloud_metadata: Do not block on String() (#47058) (#47172)

In `addCloudMetadata`, do not block by waiting for `init()` when calling `String()`, but return `add_cloud_metadata=<uninitialized>`. After the timeout, 3 seconds by default, `add_cloud_metadata` will be logged as previously.

I've decided to keep the old log format because this is also dumped during agent diagnostics and I figured the complexity is not significant. We can still go for a simpler approach of not logging any metadata instead.

(cherry picked from commit 8879dda)

Co-authored-by: Orestis Floros <[email protected]>

Author: mergify[bot]

changelog/fragments/1760359519-fix-44203-add_cloud_metadata-delay.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 3s startup delay when add_cloud_metadata is used with debug logs
+
+# 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: all
+
+# 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/owner/repo/1234
+
+# 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

libbeat/processors/add_cloud_metadata/add_cloud_metadata.go

@@ -46,6 +46,7 @@ func init() {
 type addCloudMetadata struct {
 	initOnce sync.Once
 	initData *initData
+	initDone chan struct{}
 	metadata mapstr.M
 	logger   *logp.Logger
 }
@@ -81,7 +82,8 @@ func New(c *cfg.C, log *logp.Logger) (beat.Processor, error) {
 			tlsConfig: tlsConfig,
 			overwrite: config.Overwrite,
 		},
-		logger: log.Named("add_cloud_metadata"),
+		initDone: make(chan struct{}),
+		logger:   log.Named("add_cloud_metadata"),
 	}
 
 	go p.init()
@@ -94,7 +96,8 @@ func (r result) String() string {
 }
 
 func (p *addCloudMetadata) init() {
-	p.initOnce.Do(func() {
+	p.initOnce.Do(func() { // fetch metadata only once
+		defer close(p.initDone) // signal that init() completed
 		result := p.fetchMetadata()
 		if result == nil {
 			p.logger.Info("add_cloud_metadata: hosting provider type not detected.")
@@ -125,7 +128,14 @@ func (p *addCloudMetadata) Run(event *beat.Event) (*beat.Event, error) {
 }
 
 func (p *addCloudMetadata) String() string {
-	return "add_cloud_metadata=" + p.getMeta().String()
+	metadataStr := "<uninitialized>"
+	select {
+	case <-p.initDone:
+		// init() completed
+		metadataStr = p.getMeta().String()
+	default:
+	}
+	return "add_cloud_metadata=" + metadataStr
 }
 
 func (p *addCloudMetadata) addMeta(event *beat.Event, meta mapstr.M) error {

libbeat/processors/add_cloud_metadata/add_cloud_metadata_test.go

@@ -0,0 +1,42 @@
+// 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.
+
+package add_cloud_metadata
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+)
+
+func Test_addCloudMetadata_String(t *testing.T) {
+	const timeout = 100 * time.Millisecond
+	cfg := conf.MustNewConfigFrom(map[string]any{
+		"providers": []string{"openstack"},
+		"host":      "fake:1234",
+		"timeout":   timeout.String(),
+	})
+	p, err := New(cfg, logptest.NewTestingLogger(t, ""))
+	require.NoError(t, err)
+	assert.Eventually(t, func() bool { return p.String() == "add_cloud_metadata=<uninitialized>" }, timeout, 10*time.Millisecond)
+	assert.Eventually(t, func() bool { return p.String() == "add_cloud_metadata={}" }, 2*timeout, 10*time.Millisecond)
+}

x-pack/filebeat/fbreceiver/receiver_test.go

@@ -106,9 +106,8 @@ func TestNewReceiver(t *testing.T) {
 					FilterMessageSnippet("add_host_metadata").
 					FilterMessageSnippet("add_cloud_metadata").
 					FilterMessageSnippet("add_docker_metadata").
-					FilterMessageSnippet("add_kubernetes_metadata").
-					Len() == 1
-				assert.True(c, processorsLoaded, "processors not loaded")
+					FilterMessageSnippet("add_kubernetes_metadata")
+				assert.Len(t, processorsLoaded.All(), 1, "processors not loaded")
 				// Check that add_host_metadata works, other processors are not guaranteed to add fields in all environments
 				return assert.Contains(c, logs["r1"][0].Flatten(), "host.architecture")
 			}, "failed to check processors loaded")
@@ -117,6 +116,14 @@ func TestNewReceiver(t *testing.T) {
 }
 
 func BenchmarkFactory(b *testing.B) {
+	for _, level := range []zapcore.Level{zapcore.InfoLevel, zapcore.DebugLevel} {
+		b.Run(level.String(), func(b *testing.B) {
+			benchmarkFactoryWithLogLevel(b, level)
+		})
+	}
+}
+
+func benchmarkFactoryWithLogLevel(b *testing.B, level zapcore.Level) {
 	tmpDir := b.TempDir()
 
 	cfg := &Config{
@@ -135,7 +142,7 @@ func BenchmarkFactory(b *testing.B) {
 				"otelconsumer": map[string]any{},
 			},
 			"logging": map[string]any{
-				"level": "info",
+				"level": level.String(),
 				"selectors": []string{
 					"*",
 				},
@@ -148,7 +155,7 @@ func BenchmarkFactory(b *testing.B) {
 	core := zapcore.NewCore(
 		zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()),
 		zapcore.Lock(zapcore.AddSync(&zapLogs)),
-		zapcore.InfoLevel)
+		level)
 
 	factory := NewFactory()
 
@@ -480,7 +487,7 @@ func TestConsumeContract(t *testing.T) {
 
 	gen := newLogGenerator(t, tmpDir)
 
-	os.Setenv("OTELCONSUMER_RECEIVERTEST", "1")
+	t.Setenv("OTELCONSUMER_RECEIVERTEST", "1")
 
 	cfg := &Config{
 		Beatconfig: map[string]any{