fix: collect expvar metrics from local registry (#17594) (#17606)

monitoring.Do uses the default registry which is not used anymore

propagate the local registries to correctly populate the
exp var endpoint response


(cherry picked from commit 0b9020f)

Co-authored-by: kruskall <99559985+kruskall@users.noreply.github.com>
Co-authored-by: Eric <22215921+ericywl@users.noreply.github.com>

Author: 3 people

internal/beater/api/expvar.go

@@ -30,26 +30,28 @@ import (
 // TODO(axw) this is copied from libbeat/service. We should move the
 // handler to libbeat/monitoring, and export it for libbeat/service and
 // apm-server to use.
-func debugVarsHandler(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+func debugVarsHandler(statsRegistry *monitoring.Registry) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
 
-	first := true
-	report := func(key string, value interface{}) {
-		if !first {
-			fmt.Fprintf(w, ",\n")
+		first := true
+		report := func(key string, value interface{}) {
+			if !first {
+				fmt.Fprintf(w, ",\n")
+			}
+			first = false
+			if str, ok := value.(string); ok {
+				fmt.Fprintf(w, "%q: %q", key, str)
+			} else {
+				fmt.Fprintf(w, "%q: %v", key, value)
+			}
 		}
-		first = false
-		if str, ok := value.(string); ok {
-			fmt.Fprintf(w, "%q: %q", key, str)
-		} else {
-			fmt.Fprintf(w, "%q: %v", key, value)
-		}
-	}
 
-	fmt.Fprintf(w, "{\n")
-	monitoring.Do(monitoring.Full, report)
-	expvar.Do(func(kv expvar.KeyValue) {
-		report(kv.Key, kv.Value)
-	})
-	fmt.Fprintf(w, "\n}\n")
+		fmt.Fprintf(w, "{\n")
+		statsRegistry.Do(monitoring.Full, report)
+		expvar.Do(func(kv expvar.KeyValue) {
+			report(kv.Key, kv.Value)
+		})
+		fmt.Fprintf(w, "\n}\n")
+	}
 }

internal/beater/api/mux.go

@@ -30,6 +30,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	"github.com/elastic/apm-data/input"
 	"github.com/elastic/apm-data/input/elasticapm"
@@ -93,6 +94,7 @@ func NewMux(
 	meterProvider metric.MeterProvider,
 	traceProvider trace.TracerProvider,
 	logger *logp.Logger,
+	statsRegistry *monitoring.Registry,
 ) (*mux.Router, error) {
 	pool := request.NewContextPool()
 	logger = logger.Named(logs.Handler)
@@ -147,7 +149,7 @@ func NewMux(
 	if beaterConfig.Expvar.Enabled {
 		path := beaterConfig.Expvar.URL
 		logger.Infof("Path %s added to request handler", path)
-		router.Handle(path, http.HandlerFunc(debugVarsHandler))
+		router.Handle(path, debugVarsHandler(statsRegistry))
 	}
 	if beaterConfig.Pprof.Enabled {
 		const path = "/debug/pprof"

internal/beater/api/mux_test.go

@@ -44,6 +44,7 @@ import (
 	"github.com/elastic/apm-server/internal/sourcemap"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
+	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
 func TestBackendRequestMetadata(t *testing.T) {
@@ -185,6 +186,7 @@ func (m muxBuilder) build(cfg *config.Config) (sdkmetric.Reader, http.Handler, e
 		mp,
 		noop.NewTracerProvider(),
 		m.Logger,
+		monitoring.NewRegistry(),
 	)
 	return reader, r, err
 }

internal/beater/beater.go

@@ -460,6 +460,7 @@ func (s *Runner) Run(ctx context.Context) error {
 		NewElasticsearchClient: newElasticsearchClient,
 		GRPCServer:             grpcServer,
 		Semaphore:              semaphore.NewWeighted(int64(s.config.MaxConcurrentDecoders)),
+		BeatMonitoring:         s.beatMonitoring,
 	}
 	if s.wrapServer != nil {
 		// Wrap the serverParams and runServer function, enabling

internal/beater/otlp/http_test.go

@@ -46,6 +46,7 @@ import (
 	"github.com/elastic/apm-server/internal/beater/monitoringtest"
 	"github.com/elastic/apm-server/internal/beater/ratelimit"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
+	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
 func TestConsumeTracesHTTP(t *testing.T) {
@@ -183,6 +184,7 @@ func newHTTPServer(t *testing.T, batchProcessor modelpb.BatchProcessor) (string,
 		mp,
 		noop.NewTracerProvider(),
 		logptest.NewTestingLogger(t, ""),
+		monitoring.NewRegistry(),
 	)
 	require.NoError(t, err)
 	srv := http.Server{Handler: router}

internal/beater/server.go

@@ -30,6 +30,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
+	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/version"
 	"github.com/elastic/elastic-agent-libs/logp"
 
@@ -147,6 +148,9 @@ type ServerParams struct {
 	// Semaphore holds a shared semaphore used to limit the number of
 	// concurrently running requests
 	Semaphore input.Semaphore
+
+	// BeatMonitoring holds the beat monitoring registries
+	BeatMonitoring beat.Monitoring
 }
 
 // newBaseRunServer returns the base RunServerFunc.
@@ -191,6 +195,7 @@ func newServer(args ServerParams, listener net.Listener) (server, error) {
 		args.MeterProvider,
 		args.TracerProvider,
 		args.Logger,
+		args.BeatMonitoring.StatsRegistry(),
 	)
 	if err != nil {
 		return server{}, err

internal/beater/tracing.go

@@ -26,6 +26,7 @@ import (
 	nooptrace "go.opentelemetry.io/otel/trace/noop"
 
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	"github.com/elastic/apm-data/input"
 	"github.com/elastic/apm-data/model/modelpb"
@@ -58,6 +59,7 @@ func newTracerServer(cfg *config.Config, listener net.Listener, logger *logp.Log
 		noopmetric.NewMeterProvider(),
 		nooptrace.NewTracerProvider(),
 		logger,
+		monitoring.NewRegistry(), // unused
 	)
 	if err != nil {
 		return nil, err