feat: start hiding beats monitoring behind otel abstraction

Author: kruskall

cmd/apm-server/main.go

@@ -30,6 +30,9 @@ func main() {
 			return beater.NewRunner(beater.RunnerParams{
 				Config: args.Config,
 				Logger: args.Logger,
+
+				MeterProvider:   args.MeterProvider,
+				MetricsGatherer: args.MetricsGatherer,
 			})
 		},
 	})

internal/beatcmd/beat.go

@@ -33,6 +33,11 @@ import (
 	"time"
 
 	"github.com/gofrs/uuid/v5"
+	"go.elastic.co/apm/module/apmotel/v2"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.uber.org/zap"
 	"go.uber.org/zap/exp/zapslog"
 	"golang.org/x/sync/errgroup"
@@ -76,6 +81,10 @@ type Beat struct {
 
 	rawConfig *config.C
 	newRunner NewRunnerFunc
+
+	metricReader   *sdkmetric.ManualReader
+	meterProvider  *sdkmetric.MeterProvider
+	metricGatherer *apmotel.Gatherer
 }
 
 // BeatParams holds parameters for NewBeat.
@@ -109,6 +118,18 @@ func NewBeat(args BeatParams) (*Beat, error) {
 		beatName = hostname
 	}
 
+	exporter, err := apmotel.NewGatherer()
+	if err != nil {
+		return nil, err
+	}
+
+	metricReader := sdkmetric.NewManualReader()
+	meterProvider := sdkmetric.NewMeterProvider(
+		sdkmetric.WithReader(exporter),
+		sdkmetric.WithReader(metricReader),
+	)
+	otel.SetMeterProvider(meterProvider)
+
 	eid := uuid.FromStringOrNil(metricreport.EphemeralID().String())
 	b := &Beat{
 		Beat: beat.Beat{
@@ -127,9 +148,12 @@ func NewBeat(args BeatParams) (*Beat, error) {
 			BeatConfig: cfg.APMServer,
 			Registry:   reload.NewRegistry(),
 		},
-		Config:    cfg,
-		newRunner: args.NewRunner,
-		rawConfig: rawConfig,
+		Config:         cfg,
+		newRunner:      args.NewRunner,
+		rawConfig:      rawConfig,
+		metricReader:   metricReader,
+		meterProvider:  meterProvider,
+		metricGatherer: &exporter,
 	}
 
 	if err := b.init(); err != nil {
@@ -374,7 +398,7 @@ func (b *Beat) Run(ctx context.Context) error {
 	}
 
 	if b.Manager.Enabled() {
-		reloader, err := NewReloader(b.Info, b.Registry, b.newRunner)
+		reloader, err := NewReloader(b.Info, b.Registry, b.newRunner, b.meterProvider, b.metricReader, b.metricGatherer)
 		if err != nil {
 			return err
 		}
@@ -390,9 +414,12 @@ func (b *Beat) Run(ctx context.Context) error {
 			return errors.New("no output defined, please define one under the output section")
 		}
 		runner, err := b.newRunner(RunnerParams{
-			Config: b.rawConfig,
-			Info:   b.Info,
-			Logger: logp.NewLogger(""),
+			Config:          b.rawConfig,
+			Info:            b.Info,
+			Logger:          logp.NewLogger(""),
+			MeterProvider:   b.meterProvider,
+			MetricReader:    b.metricReader,
+			MetricsGatherer: b.metricGatherer,
 		})
 		if err != nil {
 			return err
@@ -410,7 +437,12 @@ func (b *Beat) Run(ctx context.Context) error {
 // is then exposed through the HTTP monitoring endpoint (e.g. /info and /state)
 // and/or pushed to Elasticsearch through the x-pack monitoring feature.
 func (b *Beat) registerMetrics() {
-	// info
+	b.registerInfoMetrics()
+	b.registerStateMetrics()
+	b.registerStatsMetrics()
+}
+
+func (b *Beat) registerInfoMetrics() {
 	infoRegistry := monitoring.GetNamespace("info").GetRegistry()
 	monitoring.NewString(infoRegistry, "version").Set(b.Info.Version)
 	monitoring.NewString(infoRegistry, "beat").Set(b.Info.Beat)
@@ -436,7 +468,9 @@ func (b *Beat) registerMetrics() {
 			monitoring.NewString(infoRegistry, "gid").Set(u.Gid)
 		}
 	}()
+}
 
+func (b *Beat) registerStateMetrics() {
 	stateRegistry := monitoring.GetNamespace("state").GetRegistry()
 
 	// state.service
@@ -457,6 +491,166 @@ func (b *Beat) registerMetrics() {
 	monitoring.NewBool(managementRegistry, "enabled").Set(b.Manager.Enabled())
 }
 
+func (b *Beat) registerStatsMetrics() {
+	// TODO: we should ensure all metrics are produced in the expected JSON
+	// hierarchy for _source compatibility.
+	libbeatRegistry := monitoring.Default.GetRegistry("libbeat")
+	monitoring.NewFunc(libbeatRegistry, "output", 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()
+		monitoring.ReportString(v, "type", "elasticsearch")
+		for _, sm := range rm.ScopeMetrics {
+			switch {
+			case sm.Scope.Name == "github.com/elastic/go-docappender":
+				addDocappenderLibbeatOutputMetrics(context.Background(), v, sm)
+			}
+		}
+	})
+	monitoring.NewFunc(libbeatRegistry, "pipeline", 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 sm.Scope.Name == "github.com/elastic/go-docappender":
+				addDocappenderLibbeatPipelineMetrics(context.Background(), v, sm)
+			}
+		}
+	})
+	monitoring.NewFunc(monitoring.Default, "output.elasticsearch", 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 sm.Scope.Name == "github.com/elastic/go-docappender":
+				addDocappenderOutputElasticsearchMetrics(context.Background(), v, sm)
+			}
+		}
+	})
+}
+
+// getScalarInt64 returns a single-value, dimensionless
+// gauge or counter integer value, or (0, false) if the
+// data does not match these constraints.
+func getScalarInt64(data metricdata.Aggregation) (int64, bool) {
+	switch data := data.(type) {
+	case metricdata.Sum[int64]:
+		if len(data.DataPoints) != 1 || data.DataPoints[0].Attributes.Len() != 0 {
+			break
+		}
+		return data.DataPoints[0].Value, true
+	case metricdata.Gauge[int64]:
+		if len(data.DataPoints) != 1 || data.DataPoints[0].Attributes.Len() != 0 {
+			break
+		}
+		return data.DataPoints[0].Value, true
+	}
+	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 {
+				monitoring.ReportInt(v, suffix, value)
+			}
+		}
+	}
+}
+
+// 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) {
+	for _, m := range sm.Metrics {
+		switch m.Name {
+		case "elasticsearch.events.processed":
+			var acked, toomany, failed int64
+			data, _ := m.Data.(metricdata.Sum[int64])
+			for _, dp := range data.DataPoints {
+				status, ok := dp.Attributes.Value(attribute.Key("status"))
+				if !ok {
+					continue
+				}
+				switch status.AsString() {
+				case "Success":
+					acked++
+				case "TooMany":
+					toomany++
+					fallthrough
+				default:
+					failed++
+				}
+			}
+			monitoring.ReportInt(v, "events.acked", acked)
+			monitoring.ReportInt(v, "events.failed", failed)
+			monitoring.ReportInt(v, "events.toomany", toomany)
+		case "elasticsearch.events.count":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "events.total", value)
+			}
+		case "elasticsearch.events.queued":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "events.active", value)
+			}
+		case "elasticsearch.flushed.bytes":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "write.bytes", value)
+			}
+		case "elasticsearch.bulk_requests.count":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "events.batches", value)
+			}
+		}
+	}
+}
+
+func addDocappenderLibbeatPipelineMetrics(ctx context.Context, v monitoring.Visitor, sm metricdata.ScopeMetrics) {
+	for _, m := range sm.Metrics {
+		switch m.Name {
+		case "elasticsearch.events.count":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "events.total", value)
+			}
+		}
+	}
+}
+
+// Add non-libbeat Elasticsearch output metrics under "output.elasticsearch".
+func addDocappenderOutputElasticsearchMetrics(ctx context.Context, v monitoring.Visitor, sm metricdata.ScopeMetrics) {
+	for _, m := range sm.Metrics {
+		switch m.Name {
+		case "elasticsearch.bulk_requests.count":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "bulk_requests.completed", value)
+			}
+		case "elasticsearch.bulk_requests.available":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "bulk_requests.available", value)
+			}
+		case "elasticsearch.indexers.created":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "indexers.created", value)
+			}
+		case "elasticsearch.indexers.destroyed":
+			if value, ok := getScalarInt64(m.Data); ok {
+				monitoring.ReportInt(v, "indexers.destroyed", value)
+			}
+			// TODO output.elasticsearch.indexers.active (created - destroyed?)
+		}
+	}
+}
+
 // registerElasticsearchVerfication registers a global callback to make sure
 // the Elasticsearch instance we are connecting to has a valid license, and is
 // at least on the same version as APM Server.

internal/beatcmd/beat_test.go

@@ -220,7 +220,7 @@ func TestRunManager_Reloader(t *testing.T) {
 			stopCount.Add(1)
 			return nil
 		}), nil
-	})
+	}, nil, nil, nil)
 	require.NoError(t, err)
 
 	agentInfo := &proto.AgentInfo{
@@ -346,7 +346,7 @@ func TestRunManager_Reloader_newRunnerError(t *testing.T) {
 
 	_, err := NewReloader(beat.Info{}, registry, func(_ RunnerParams) (Runner, error) {
 		return nil, errors.New("newRunner error")
-	})
+	}, nil, nil, nil)
 	require.NoError(t, err)
 
 	onObserved := func(observed *proto.CheckinObserved, currentIdx int) {

internal/beatcmd/reloader.go

@@ -24,6 +24,9 @@ import (
 	"sync"
 
 	"github.com/joeshaw/multierror"
+	"go.elastic.co/apm/module/apmotel/v2"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -50,6 +53,21 @@ type RunnerParams struct {
 
 	// Logger holds a logger to use for logging throughout the APM Server.
 	Logger *logp.Logger
+
+	// MeterProvider holds a metric.MeterProvider that can be used for
+	// creating metrics. The same MeterProvider is expected to be used
+	// for each instance of the Runner, to ensure counter metrics are
+	// not reset.
+	//
+	// NOTE(axw) metrics registered through this provider are used for
+	// feeding into both Elastic APM (if enabled) and the libbeat
+	// monitoring framework. For the latter, only gauge and counter
+	// metrics are supported, and attributes (dimensions) are ignored.
+	MeterProvider metric.MeterProvider
+
+	MetricsGatherer *apmotel.Gatherer
+
+	MetricReader *sdkmetric.ManualReader
 }
 
 // Runner is an interface returned by NewRunnerFunc.
@@ -60,12 +78,16 @@ type Runner interface {
 
 // NewReloader returns a new Reloader which creates Runners using the provided
 // beat.Info and NewRunnerFunc.
-func NewReloader(info beat.Info, registry *reload.Registry, newRunner NewRunnerFunc) (*Reloader, error) {
+func NewReloader(info beat.Info, registry *reload.Registry, newRunner NewRunnerFunc, meterProvider metric.MeterProvider, metricReader *sdkmetric.ManualReader, metricGatherer *apmotel.Gatherer) (*Reloader, error) {
 	r := &Reloader{
 		info:      info,
 		logger:    logp.NewLogger(""),
 		newRunner: newRunner,
 		stopped:   make(chan struct{}),
+
+		meterProvider:  meterProvider,
+		metricReader:   metricReader,
+		metricGatherer: metricGatherer,
 	}
 	if err := registry.RegisterList(reload.InputRegName, reloadableListFunc(r.reloadInputs)); err != nil {
 		return nil, fmt.Errorf("failed to register inputs reloader: %w", err)
@@ -86,6 +108,10 @@ type Reloader struct {
 	logger    *logp.Logger
 	newRunner NewRunnerFunc
 
+	meterProvider  metric.MeterProvider
+	metricReader   *sdkmetric.ManualReader
+	metricGatherer *apmotel.Gatherer
+
 	runner     Runner
 	stopRunner func() error
 

internal/beatcmd/reloader_test.go

@@ -72,7 +72,7 @@ func TestReloader(t *testing.T) {
 			<-ctx.Done()
 			return nil
 		}), nil
-	})
+	}, nil, nil, nil)
 	require.NoError(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -156,7 +156,7 @@ func TestReloaderNewRunnerParams(t *testing.T) {
 			<-ctx.Done()
 			return nil
 		}), nil
-	})
+	}, nil, nil, nil)
 	require.NoError(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())