monitoring: update apm-server metrics collection to avoid conflicts (#17512) (#17527)

* monitoring: update apm-server metrics collection to avoid conflicts

* refactor beats monitoring since there are no more global registries

* removed redundant temp slice from apm-server monitoring func

(cherry picked from commit f1c279b)

Co-authored-by: Isaac Flores <34590010+isaacaflores2@users.noreply.github.com>

Author: mergify[bot]

internal/beatcmd/beat.go

@@ -547,14 +547,22 @@ func (b *Beat) registerStatsMetrics() {
 		}
 		v.OnRegistryStart()
 		defer v.OnRegistryFinished()
+
+		// first collect all apm-server metrics
+		beatsMetrics := make(map[string]any)
 		for _, sm := range rm.ScopeMetrics {
 			switch {
 			case strings.HasPrefix(sm.Scope.Name, "github.com/elastic/apm-server"):
 				// All simple scalar metrics that begin with the name "apm-server."
 				// in github.com/elastic/apm-server/... scopes are mapped directly.
-				addAPMServerMetrics(v, sm)
+				addAPMServerMetricsToMap(beatsMetrics, sm.Metrics)
 			}
 		}
+
+		// register all metrics once
+		// this prevents metrics with the same prefix in the name
+		// from different scoped meters from overwriting each other
+		reportOnKey(v, beatsMetrics)
 	})
 }
 
@@ -577,9 +585,10 @@ func getScalarInt64(data metricdata.Aggregation) (int64, bool) {
 	return 0, false
 }
 
-func addAPMServerMetrics(v monitoring.Visitor, sm metricdata.ScopeMetrics) {
-	beatsMetrics := make(map[string]any)
-	for _, m := range sm.Metrics {
+// addAPMServerMetricsToMap adds simple scalar metrics with the "apm-server." prefix
+// to the map.
+func addAPMServerMetricsToMap(beatsMetrics map[string]any, metrics []metricdata.Metrics) {
+	for _, m := range metrics {
 		if suffix, ok := strings.CutPrefix(m.Name, "apm-server."); ok {
 			if value, ok := getScalarInt64(m.Data); ok {
 				current := beatsMetrics
@@ -597,8 +606,6 @@ func addAPMServerMetrics(v monitoring.Visitor, sm metricdata.ScopeMetrics) {
 			}
 		}
 	}
-
-	reportOnKey(v, beatsMetrics)
 }
 
 func reportOnKey(v monitoring.Visitor, m map[string]any) {

internal/beatcmd/beat_test.go

@@ -268,6 +268,7 @@ func TestLibbeatMetrics(t *testing.T) {
 	}, snapshot)
 }
 
+// TestAddAPMServerMetrics tests basic functionality of the metrics collection and reporting
 func TestAddAPMServerMetrics(t *testing.T) {
 	r := monitoring.NewRegistry()
 	sm := metricdata.ScopeMetrics{
@@ -296,7 +297,12 @@ func TestAddAPMServerMetrics(t *testing.T) {
 	}
 
 	monitoring.NewFunc(r, "apm-server", func(m monitoring.Mode, v monitoring.Visitor) {
-		addAPMServerMetrics(v, sm)
+		v.OnRegistryStart()
+		defer v.OnRegistryFinished()
+
+		beatsMetrics := make(map[string]any)
+		addAPMServerMetricsToMap(beatsMetrics, sm.Metrics)
+		reportOnKey(v, beatsMetrics)
 	})
 
 	snapshot := monitoring.CollectStructSnapshot(r, monitoring.Full, false)
@@ -305,6 +311,47 @@ func TestAddAPMServerMetrics(t *testing.T) {
 			"request":  int64(1),
 			"response": int64(1),
 		},
+	}, snapshot["apm-server"])
+}
+
+func TestMonitoringApmServer(t *testing.T) {
+	b := newNopBeat(t, "")
+	b.registerStatsMetrics()
+
+	// add metrics similar to lsm_size in storage_manager.go and events.processed in processor.go
+	meter := b.meterProvider.Meter("github.com/elastic/apm-server/x-pack/apm-server/sampling/eventstorage")
+	lsmSizeGauge, _ := meter.Int64Gauge("apm-server.sampling.tail.storage.lsm_size")
+	lsmSizeGauge.Record(context.Background(), 123)
+
+	meter2 := b.meterProvider.Meter("github.com/elastic/apm-server/x-pack/apm-server/sampling")
+	processedCounter, _ := meter2.Int64Counter("apm-server.sampling.tail.events.processed")
+	processedCounter.Add(context.Background(), 456)
+
+	meter3 := b.meterProvider.Meter("github.com/elastic/apm-server/x-pack/apm-server/foo")
+	otherCounter, _ := meter3.Int64Counter("apm-server.sampling.foo.request")
+	otherCounter.Add(context.Background(), 1)
+
+	// collect metrics
+	snapshot := monitoring.CollectStructSnapshot(b.Monitoring.StatsRegistry(), monitoring.Full, false)
+
+	// assert that the snapshot contains data for all scoped metrics
+	// with the same metric name prefix 'apm-server.sampling'
+	assert.Equal(t, map[string]any{
+		"apm-server": map[string]any{
+			"sampling": map[string]any{
+				"foo": map[string]any{
+					"request": int64(1),
+				},
+				"tail": map[string]any{
+					"storage": map[string]any{
+						"lsm_size": int64(123),
+					},
+					"events": map[string]any{
+						"processed": int64(456),
+					},
+				},
+			},
+		},
 	}, snapshot)
 }