[8.19](backport #48296) libbeat: memoize asset decoding per beat (#48376)

* libbeat: memoize asset decoding per beat (#48296)

Asset decoding is an expensive operation that happens during Beat
initialization. For standalone beats it runs only once at startup, so the cost
is amortized over the execution time.

For Beats receivers, multiple beat instances can be started by the
collector in the same process, delaying startup.
Since asset data is static per beat, decoded assets can be reused if the same
Beat was already initialized.

This change adds a global asset cache protected by a mutex,
ensuring each Beat is decoded only once, with subsequent instances
reading from the cache. This works because we only set fields in the registry
during program initialization, so calls to GetFields are immutable per beat.

While at it, cache the fqdn resolution that takes a significant portion
of the remaining cpu time.

Real collector (oteltestcol with 4 filebeat receivers) benchmarks
Measured time to first log ingested:

sec/op
46.01m ± 2% 10.74m ± 1% -76.67%

B/op
47.694Mi ± 0% 5.186Mi ± 4% -89.13%

allocs/op
49.82k ± 0% 39.22k ± 0% -21.28%

(cherry picked from commit aa318d4)

# Conflicts:
#	libbeat/asset/registry.go

* fix conflicts

---------

Co-authored-by: Mauri de Souza Meneguzzo <mauri870@gmail.com>

Author: mergify[bot]

libbeat/asset/registry.go

@@ -22,10 +22,14 @@ import (
 	"compress/zlib"
 	"encoding/base64"
 	"sort"
+	"sync"
 
 	"github.com/elastic/elastic-agent-libs/iobuf"
 )
 
+var beatFieldsCacheMu sync.Mutex
+var beatFieldsCache = map[string][]byte{}
+
 // FieldsRegistry contains a list of fields.yml files
 // As each entry is an array of bytes multiple fields.yml can be added under one path.
 // This can become useful as we don't have to generate anymore the fields.yml but can
@@ -53,6 +57,10 @@ func SetFields(beat, name string, p Priority, asset func() string) error {
 
 // GetFields returns a byte array contains all fields for the given beat
 func GetFields(beat string) ([]byte, error) {
+	if cached, ok := getBeatFieldsCache(beat); ok {
+		return cached, nil
+	}
+
 	var fields []byte
 
 	// Get all priorities and sort them
@@ -86,6 +94,8 @@ func GetFields(beat string) ([]byte, error) {
 			}
 		}
 	}
+
+	setBeatFieldsCache(beat, fields)
 	return fields, nil
 }
 
@@ -105,6 +115,19 @@ func EncodeData(data string) (string, error) {
 	return base64.StdEncoding.EncodeToString(zlibBuf.Bytes()), nil
 }
 
+func getBeatFieldsCache(beat string) ([]byte, bool) {
+	beatFieldsCacheMu.Lock()
+	defer beatFieldsCacheMu.Unlock()
+	cached, ok := beatFieldsCache[beat]
+	return cached, ok
+}
+
+func setBeatFieldsCache(beat string, fields []byte) {
+	beatFieldsCacheMu.Lock()
+	defer beatFieldsCacheMu.Unlock()
+	beatFieldsCache[beat] = fields
+}
+
 // DecodeData base64 decodes the data and uncompresses it
 func DecodeData(data string) ([]byte, error) {
 	decoded, err := base64.StdEncoding.DecodeString(data)

x-pack/filebeat/tests/integration/otel_test.go

@@ -1333,3 +1333,73 @@ func setupRoleMapping(t *testing.T, client *elasticsearch.Client) {
 
 	require.Equal(t, resp.StatusCode, http.StatusOK, "incorrect response code")
 }
+
+func BenchmarkFilebeatOTelCollector(b *testing.B) {
+	numReceivers := 4
+
+	for b.Loop() {
+		b.StopTimer()
+		tmpDir := b.TempDir()
+
+		type receiverConfig struct {
+			Index    int
+			PathHome string
+		}
+
+		configData := struct {
+			Receivers []receiverConfig
+		}{
+			Receivers: make([]receiverConfig, numReceivers),
+		}
+
+		for i := range numReceivers {
+			configData.Receivers[i] = receiverConfig{
+				Index:    i + 1,
+				PathHome: filepath.Join(tmpDir, strconv.Itoa(i+1)),
+			}
+		}
+
+		cfgTemplate := `receivers:
+{{range .Receivers}}
+  filebeatreceiver/{{.Index}}:
+    filebeat:
+      inputs:
+        - type: benchmark
+          enabled: true
+          count: 1
+    path.home: {{.PathHome}}
+    queue.mem.flush.timeout: 0s
+{{end}}
+exporters:
+  debug:
+    verbosity: detailed
+service:
+  pipelines:
+    logs:
+      receivers:
+{{range .Receivers}}
+        - filebeatreceiver/{{.Index}}
+{{end}}
+      exporters:
+        - debug
+  telemetry:
+    logs:
+      level: DEBUG
+    metrics:
+      level: none
+`
+
+		var configBuffer bytes.Buffer
+		require.NoError(b, template.Must(template.New("config").Parse(cfgTemplate)).Execute(&configBuffer, configData))
+
+		b.StartTimer()
+
+		col := oteltestcol.New(b, configBuffer.String())
+		require.NotNil(b, col)
+		require.Eventually(b, func() bool {
+			return col.ObservedLogs().
+				FilterMessageSnippet("Publish event").Len() == numReceivers
+		}, 30*time.Second, 1*time.Millisecond, "expected all receivers to publish events")
+		col.Shutdown()
+	}
+}

x-pack/libbeat/cmd/instance/beat.go

@@ -7,6 +7,7 @@ package instance
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"go.opentelemetry.io/collector/consumer"
@@ -38,6 +39,18 @@ import (
 // input type.
 const receiverPublisherCloseTimeout = 5 * time.Second
 
+var fqdnOnce = sync.OnceValues(func() (string, error) {
+	h, err := sysinfo.Host()
+	if err != nil {
+		return "", fmt.Errorf("failed to get host information: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
+	defer cancel()
+
+	return h.FQDNWithContext(ctx)
+})
+
 // NewBeatForReceiver creates a Beat that will be used in the context of an otel receiver
 func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]any, consumer consumer.Logs, componentID string, core zapcore.Core) (*instance.Beat, error) {
 	b, err := instance.NewBeat(settings.Name,
@@ -180,15 +193,7 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 	logger.Infof("Beat ID: %v", b.Info.ID)
 
 	// Try to get the host's FQDN and set it.
-	h, err := sysinfo.Host()
-	if err != nil {
-		return nil, fmt.Errorf("failed to get host information: %w", err)
-	}
-
-	fqdnLookupCtx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
-	defer cancel()
-
-	fqdn, err := h.FQDNWithContext(fqdnLookupCtx)
+	fqdn, err := fqdnOnce()
 	if err != nil {
 		// FQDN lookup is "best effort".  We log the error, fallback to
 		// the OS-reported hostname, and move on.

x-pack/libbeat/common/otelbeat/oteltestcol/collector.go

@@ -94,6 +94,10 @@ func (c *Collector) ObservedLogs() *observer.ObservedLogs {
 	return c.observer
 }
 
+func (c *Collector) Shutdown() {
+	c.collector.Shutdown()
+}
+
 func getComponent() (otelcol.Factories, error) {
 	receivers, err := otelcol.MakeFactoryMap(
 		fbreceiver.NewFactory(),