Translate otel metrics to libbeat monitoring

internal/agentcfg/elasticsearch.go

@@ -29,11 +29,11 @@ import (
 
 	"github.com/pkg/errors"
 	"go.elastic.co/apm/v2"
+	"go.opentelemetry.io/otel/metric"
 
 	"github.com/elastic/apm-server/internal/elasticsearch"
 	"github.com/elastic/apm-server/internal/logs"
 	"github.com/elastic/elastic-agent-libs/logp"
-	"github.com/elastic/elastic-agent-libs/monitoring"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
 )
 
@@ -71,22 +71,34 @@ type ElasticsearchFetcher struct {
 
 	logger, rateLimitedLogger *logp.Logger
 
-	tracer  *apm.Tracer
-	metrics fetcherMetrics
-}
+	tracer *apm.Tracer
 
-type fetcherMetrics struct {
-	fetchES, fetchFallback, fetchFallbackUnavailable, fetchInvalid,
-	cacheRefreshSuccesses, cacheRefreshFailures,
-	cacheEntriesCount atomic.Int64
+	esCacheEntriesCount     metric.Int64Gauge
+	esFetchCount            metric.Int64Counter
+	esFetchFallbackCount    metric.Int64Counter
+	esFetchUnavailableCount metric.Int64Counter
+	esFetchInvalidCount     metric.Int64Counter
+	esCacheRefreshSuccesses metric.Int64Counter
+	esCacheRefreshFailures  metric.Int64Counter
 }
 
 func NewElasticsearchFetcher(
 	client *elasticsearch.Client,
 	cacheDuration time.Duration,
 	fetcher Fetcher,
 	tracer *apm.Tracer,
+	mp metric.MeterProvider,
 ) *ElasticsearchFetcher {
+	meter := mp.Meter("github.com/elastic/apm-server/internal/agentcfg")
+
+	esCacheEntriesCount, _ := meter.Int64Gauge("apm-server.agentcfg.elasticsearch.cache.entries.count")
+	esFetchCount, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.fetch.es")
+	esFetchFallbackCount, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.fetch.fallback")
+	esFetchUnavailableCount, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.fetch.unavailable")
+	esFetchInvalidCount, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.fetch.invalid")
+	esCacheRefreshSuccesses, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.cache.refresh.successes")
+	esCacheRefreshFailures, _ := meter.Int64Counter("apm-server.agentcfg.elasticsearch.cache.refresh.failures")
+
 	logger := logp.NewLogger("agentcfg")
 	return &ElasticsearchFetcher{
 		client:            client,
@@ -96,6 +108,14 @@ func NewElasticsearchFetcher(
 		logger:            logger,
 		rateLimitedLogger: logger.WithOptions(logs.WithRateLimit(loggerRateLimit)),
 		tracer:            tracer,
+
+		esCacheEntriesCount:     esCacheEntriesCount,
+		esFetchCount:            esFetchCount,
+		esFetchFallbackCount:    esFetchFallbackCount,
+		esFetchUnavailableCount: esFetchUnavailableCount,
+		esFetchInvalidCount:     esFetchInvalidCount,
+		esCacheRefreshSuccesses: esCacheRefreshSuccesses,
+		esCacheRefreshFailures:  esCacheRefreshFailures,
 	}
 }
 
@@ -105,22 +125,22 @@ func (f *ElasticsearchFetcher) Fetch(ctx context.Context, query Query) (Result,
 		// Happy path: serve fetch requests using an initialized cache.
 		f.mu.RLock()
 		defer f.mu.RUnlock()
-		f.metrics.fetchES.Add(1)
+		f.esFetchCount.Add(ctx, 1)
 		return matchAgentConfig(query, f.cache), nil
 	}
 
 	if f.fallbackFetcher != nil {
-		f.metrics.fetchFallback.Add(1)
+		f.esFetchFallbackCount.Add(ctx, 1)
 		return f.fallbackFetcher.Fetch(ctx, query)
 	}
 
 	if f.invalidESCfg.Load() {
-		f.metrics.fetchInvalid.Add(1)
+		f.esFetchInvalidCount.Add(ctx, 1)
 		f.rateLimitedLogger.Errorf("rejecting fetch request: no valid elasticsearch config")
 		return Result{}, errors.New(ErrNoValidElasticsearchConfig)
 	}
 
-	f.metrics.fetchFallbackUnavailable.Add(1)
+	f.esFetchUnavailableCount.Add(ctx, 1)
 	f.rateLimitedLogger.Warnf("rejecting fetch request: infrastructure is not ready")
 	return Result{}, errors.New(ErrInfrastructureNotReady)
 }
@@ -213,9 +233,9 @@ func (f *ElasticsearchFetcher) refreshCache(ctx context.Context) (err error) {
 
 	defer func() {
 		if err != nil {
-			f.metrics.cacheRefreshFailures.Add(1)
+			f.esCacheRefreshFailures.Add(ctx, 1)
 		} else {
-			f.metrics.cacheRefreshSuccesses.Add(1)
+			f.esCacheRefreshSuccesses.Add(ctx, 1)
 		}
 	}()
 
@@ -247,7 +267,7 @@ func (f *ElasticsearchFetcher) refreshCache(ctx context.Context) (err error) {
 	f.cache = buffer
 	f.mu.Unlock()
 	f.cacheInitialized.Store(true)
-	f.metrics.cacheEntriesCount.Store(int64(len(f.cache)))
+	f.esCacheEntriesCount.Record(ctx, int64(len(f.cache)))
 	f.last = time.Now()
 	return nil
 }
@@ -304,20 +324,3 @@ func (f *ElasticsearchFetcher) singlePageRefresh(ctx context.Context, scrollID s
 	}
 	return result, json.NewDecoder(resp.Body).Decode(&result)
 }
-
-// CollectMonitoring may be called to collect monitoring metrics from the
-// fetcher. It is intended to be used with libbeat/monitoring.NewFunc.
-//
-// The metrics should be added to the "apm-server.agentcfg.elasticsearch" registry.
-func (f *ElasticsearchFetcher) CollectMonitoring(_ monitoring.Mode, V monitoring.Visitor) {
-	V.OnRegistryStart()
-	defer V.OnRegistryFinished()
-
-	monitoring.ReportInt(V, "cache.entries.count", f.metrics.cacheEntriesCount.Load())
-	monitoring.ReportInt(V, "fetch.es", f.metrics.fetchES.Load())
-	monitoring.ReportInt(V, "fetch.fallback", f.metrics.fetchFallback.Load())
-	monitoring.ReportInt(V, "fetch.unavailable", f.metrics.fetchFallbackUnavailable.Load())
-	monitoring.ReportInt(V, "fetch.invalid", f.metrics.fetchInvalid.Load())
-	monitoring.ReportInt(V, "cache.refresh.successes", f.metrics.cacheRefreshSuccesses.Load())
-	monitoring.ReportInt(V, "cache.refresh.failures", f.metrics.cacheRefreshFailures.Load())
-}

internal/agentcfg/elasticsearch_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.elastic.co/apm/v2"
 	"go.elastic.co/apm/v2/apmtest"
+	"go.opentelemetry.io/otel/metric/noop"
 
 	"github.com/elastic/apm-server/internal/elasticsearch"
 )
@@ -104,7 +105,7 @@ func newElasticsearchFetcher(
 		w.WriteHeader(200)
 		w.Write(b)
 		i += searchSize
-	}), time.Second, nil, rt)
+	}), time.Second, nil, rt, noop.NewMeterProvider())
 	fetcher.searchSize = searchSize
 	return fetcher
 }
@@ -193,6 +194,7 @@ func TestFetchUseFallback(t *testing.T) {
 		time.Second,
 		fallbackFetcher,
 		apmtest.NewRecordingTracer().Tracer,
+		noop.NewMeterProvider(),
 	)
 
 	fetcher.refreshCache(context.Background())
@@ -208,6 +210,7 @@ func TestFetchNoFallbackInvalidESCfg(t *testing.T) {
 		time.Second,
 		nil,
 		apmtest.NewRecordingTracer().Tracer,
+		noop.NewMeterProvider(),
 	)
 
 	err := fetcher.refreshCache(context.Background())
@@ -224,6 +227,7 @@ func TestFetchNoFallback(t *testing.T) {
 		time.Second,
 		nil,
 		apmtest.NewRecordingTracer().Tracer,
+		noop.NewMeterProvider(),
 	)
 
 	err := fetcher.refreshCache(context.Background())

internal/beatcmd/beat.go

@@ -539,6 +539,22 @@ func (b *Beat) registerStatsMetrics() {
 			}
 		}
 	})
+	monitoring.NewFunc(monitoring.Default, "apm-server", func(_ monitoring.Mode, v monitoring.Visitor) {
+		var rm metricdata.ResourceMetrics
+		if err := b.metricReader.Collect(context.Background(), &rm); err != nil {
+			return
+		}
+		v.OnRegistryStart()
+		defer v.OnRegistryFinished()
+		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)
+			}
+		}
+	})
 }
 
 // getScalarInt64 returns a single-value, dimensionless
@@ -560,6 +576,24 @@ func getScalarInt64(data metricdata.Aggregation) (int64, bool) {
 	return 0, false
 }
 
+func addAPMServerMetrics(v monitoring.Visitor, sm metricdata.ScopeMetrics) {
+	for _, m := range sm.Metrics {
+		if suffix, ok := strings.CutPrefix(m.Name, "apm-server."); ok {
+			if value, ok := getScalarInt64(m.Data); ok {
+				keys := strings.Split(suffix, ".")
+				for i := 0; i < len(keys)-1; i++ {
+					v.OnRegistryStart()
+					v.OnKey(keys[i])
+				}
+				monitoring.ReportInt(v, keys[len(keys)-1], value)
+				for i := 0; i < len(keys)-1; i++ {
+					v.OnRegistryFinished()
+				}
+			}
+		}
+	}
+}
+
 // Adapt go-docappender's OTel metrics to beats stack monitoring metrics,
 // with a mixture of libbeat-specific and apm-server specific metric names.
 func addDocappenderLibbeatOutputMetrics(ctx context.Context, v monitoring.Visitor, sm metricdata.ScopeMetrics) {

internal/beatcmd/reloader.go

@@ -254,9 +254,11 @@ func (r *Reloader) reload(inputConfig, outputConfig, apmTracingConfig *config.C)
 	// allow the runner to perform initialisations that must run
 	// synchronously.
 	newRunner, err := r.newRunner(RunnerParams{
-		Config: mergedConfig,
-		Info:   r.info,
-		Logger: r.logger,
+		Config:          mergedConfig,
+		Info:            r.info,
+		Logger:          r.logger,
+		MeterProvider:   r.meterProvider,
+		MetricsGatherer: r.metricGatherer,
 	})
 	if err != nil {
 		return err

internal/beater/api/config/agent/handler.go

@@ -26,8 +26,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/elastic/elastic-agent-libs/monitoring"
-
 	"github.com/elastic/apm-server/internal/agentcfg"
 	"github.com/elastic/apm-server/internal/beater/auth"
 	"github.com/elastic/apm-server/internal/beater/headers"
@@ -43,10 +41,6 @@ const (
 )
 
 var (
-	// MonitoringMap holds a mapping for request.IDs to monitoring counters
-	MonitoringMap = request.DefaultMonitoringMapForRegistry(registry)
-	registry      = monitoring.Default.NewRegistry("apm-server.acm")
-
 	errCacheControl = fmt.Sprintf("max-age=%v, must-revalidate", errMaxAgeDuration.Seconds())
 )
 

internal/beater/api/intake/handler.go

@@ -18,12 +18,13 @@
 package intake
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/http"
 	"strings"
 
-	"github.com/elastic/elastic-agent-libs/monitoring"
+	"go.opentelemetry.io/otel/metric"
 
 	"github.com/elastic/apm-data/input/elasticapm"
 	"github.com/elastic/apm-data/model/modelpb"
@@ -39,15 +40,6 @@ const (
 )
 
 var (
-	// MonitoringMap holds a mapping for request.IDs to monitoring counters
-	MonitoringMap = request.DefaultMonitoringMapForRegistry(registry)
-	registry      = monitoring.Default.NewRegistry("apm-server.server")
-
-	streamRegistry = monitoring.Default.NewRegistry("apm-server.processor.stream")
-	eventsAccepted = monitoring.NewInt(streamRegistry, "accepted")
-	eventsInvalid  = monitoring.NewInt(streamRegistry, "errors.invalid")
-	eventsTooLarge = monitoring.NewInt(streamRegistry, "errors.toolarge")
-
 	errMethodNotAllowed   = errors.New("only POST requests are supported")
 	errServerShuttingDown = errors.New("server is shutting down")
 	errInvalidContentType = errors.New("invalid content type")
@@ -59,7 +51,12 @@ var (
 type RequestMetadataFunc func(*request.Context) *modelpb.APMEvent
 
 // Handler returns a request.Handler for managing intake requests for backend and rum events.
-func Handler(handler elasticapm.StreamHandler, requestMetadataFunc RequestMetadataFunc, batchProcessor modelpb.BatchProcessor) request.Handler {
+func Handler(mp metric.MeterProvider, handler elasticapm.StreamHandler, requestMetadataFunc RequestMetadataFunc, batchProcessor modelpb.BatchProcessor) request.Handler {
+	meter := mp.Meter("github.com/elastic/apm-server/internal/beater/api/intake")
+	eventsAccepted, _ := meter.Int64Counter("apm-server.processor.stream.accepted")
+	eventsInvalid, _ := meter.Int64Counter("apm-server.processor.stream.errors.invalid")
+	eventsTooLarge, _ := meter.Int64Counter("apm-server.processor.stream.errors.toolarge")
+
 	return func(c *request.Context) {
 		if err := validateRequest(c); err != nil {
 			writeError(c, err)
@@ -82,9 +79,9 @@ func Handler(handler elasticapm.StreamHandler, requestMetadataFunc RequestMetada
 			batchProcessor,
 			&result,
 		)
-		eventsAccepted.Add(int64(result.Accepted))
-		eventsInvalid.Add(int64(result.Invalid))
-		eventsTooLarge.Add(int64(result.TooLarge))
+		eventsAccepted.Add(context.Background(), int64(result.Accepted))
+		eventsInvalid.Add(context.Background(), int64(result.Invalid))
+		eventsTooLarge.Add(context.Background(), int64(result.TooLarge))
 		writeStreamResult(c, result, err)
 	}
 }

internal/beater/api/intake/handler_test.go

@@ -32,12 +32,16 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"golang.org/x/sync/semaphore"
 
 	"github.com/elastic/apm-data/input/elasticapm"
 	"github.com/elastic/apm-data/model/modelpb"
 	"github.com/elastic/apm-server/internal/beater/config"
 	"github.com/elastic/apm-server/internal/beater/headers"
+	"github.com/elastic/apm-server/internal/beater/monitoringtest"
 	"github.com/elastic/apm-server/internal/beater/request"
 	"github.com/elastic/apm-server/internal/model/modelprocessor"
 	"github.com/elastic/apm-server/internal/publish"
@@ -141,7 +145,7 @@ func TestIntakeHandler(t *testing.T) {
 			tc.setup(t)
 
 			// call handler
-			h := Handler(tc.processor, emptyRequestMetadata, tc.batchProcessor)
+			h := Handler(noop.NewMeterProvider(), tc.processor, emptyRequestMetadata, tc.batchProcessor)
 			h(tc.c)
 
 			require.Equal(t, string(tc.id), string(tc.c.Result.ID))
@@ -163,10 +167,6 @@ func TestIntakeHandler(t *testing.T) {
 }
 
 func TestIntakeHandlerMonitoring(t *testing.T) {
-	eventsAccepted.Set(1)
-	eventsInvalid.Set(2)
-	eventsTooLarge.Set(3)
-
 	streamHandler := streamHandlerFunc(func(
 		ctx context.Context,
 		base *modelpb.APMEvent,
@@ -181,15 +181,24 @@ func TestIntakeHandlerMonitoring(t *testing.T) {
 		return errors.New("something bad happened at the end")
 	})
 
-	h := Handler(streamHandler, emptyRequestMetadata, modelprocessor.Nop{})
+	reader := sdkmetric.NewManualReader(sdkmetric.WithTemporalitySelector(
+		func(ik sdkmetric.InstrumentKind) metricdata.Temporality {
+			return metricdata.DeltaTemporality
+		},
+	))
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+
+	h := Handler(mp, streamHandler, emptyRequestMetadata, modelprocessor.Nop{})
 	req := httptest.NewRequest("POST", "/", nil)
 	c := request.NewContext()
 	c.Reset(httptest.NewRecorder(), req)
 	h(c)
 
-	assert.Equal(t, int64(11), eventsAccepted.Get())
-	assert.Equal(t, int64(102), eventsInvalid.Get())
-	assert.Equal(t, int64(1003), eventsTooLarge.Get())
+	monitoringtest.ExpectOtelMetrics(t, reader, map[string]any{
+		"apm-server.processor.stream.accepted":        10,
+		"apm-server.processor.stream.errors.invalid":  100,
+		"apm-server.processor.stream.errors.toolarge": 1000,
+	})
 }
 
 func TestIntakeHandlerContentType(t *testing.T) {
@@ -206,7 +215,7 @@ func TestIntakeHandlerContentType(t *testing.T) {
 		}
 
 		tc.setup(t)
-		h := Handler(tc.processor, emptyRequestMetadata, tc.batchProcessor)
+		h := Handler(noop.NewMeterProvider(), tc.processor, emptyRequestMetadata, tc.batchProcessor)
 		h(tc.c)
 		assert.Equal(t, tc.code, tc.w.Code, tc.c.Result.Err)
 	}