Revert "demo: send metrics directly and add another reader"

This reverts commit 166a717.

Author: kruskall

cmd/apm-server/main.go

@@ -28,9 +28,8 @@ func main() {
 	rootCmd := beatcmd.NewRootCommand(beatcmd.BeatParams{
 		NewRunner: func(args beatcmd.RunnerParams) (beatcmd.Runner, error) {
 			return beater.NewRunner(beater.RunnerParams{
-				Config:         args.Config,
-				Logger:         args.Logger,
-				MetricExporter: args.MetricExporter,
+				Config: args.Config,
+				Logger: args.Logger,
 			})
 		},
 	})

internal/beatcmd/beat.go

@@ -65,7 +65,6 @@ import (
 	sysinfo "github.com/elastic/go-sysinfo"
 	"github.com/elastic/go-sysinfo/types"
 
-	"github.com/elastic/apm-server/internal/telemetry"
 	"github.com/elastic/apm-server/internal/version"
 )
 
@@ -79,11 +78,10 @@ type Beat struct {
 
 	Config *Config
 
-	rawConfig      *config.C
-	newRunner      NewRunnerFunc
-	metricReader   *sdkmetric.ManualReader
-	meterProvider  *sdkmetric.MeterProvider
-	metricExporter *telemetry.MetricExporter
+	rawConfig     *config.C
+	newRunner     NewRunnerFunc
+	metricReader  *sdkmetric.ManualReader
+	meterProvider *sdkmetric.MeterProvider
 	//TODO metricGatherer apmotel.Gatherer
 }
 
@@ -118,14 +116,12 @@ func NewBeat(args BeatParams) (*Beat, error) {
 		beatName = hostname
 	}
 
-	localExporter := telemetry.NewMetricExporter()
 	metricReader := sdkmetric.NewManualReader()
 	meterProvider := sdkmetric.NewMeterProvider(
 		// TODO(axw) register an apmotel gatherer,
 		// pass it into the runner so it can be
 		// registered with tracers.
 		sdkmetric.WithReader(metricReader),
-		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(localExporter)),
 	)
 	otel.SetMeterProvider(meterProvider)
 
@@ -147,12 +143,11 @@ func NewBeat(args BeatParams) (*Beat, error) {
 			BeatConfig: cfg.APMServer,
 			Registry:   reload.NewRegistry(),
 		},
-		Config:         cfg,
-		newRunner:      args.NewRunner,
-		rawConfig:      rawConfig,
-		metricReader:   metricReader,
-		meterProvider:  meterProvider,
-		metricExporter: localExporter,
+		Config:        cfg,
+		newRunner:     args.NewRunner,
+		rawConfig:     rawConfig,
+		metricReader:  metricReader,
+		meterProvider: meterProvider,
 	}
 
 	if err := b.init(); err != nil {
@@ -414,12 +409,11 @@ 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(""),
-			MeterProvider:  b.meterProvider,
-			MetricReader:   b.metricReader,
-			MetricExporter: b.metricExporter,
+			Config:        b.rawConfig,
+			Info:          b.Info,
+			Logger:        logp.NewLogger(""),
+			MeterProvider: b.meterProvider,
+			MetricReader:  b.metricReader,
 		})
 		if err != nil {
 			return err

internal/beatcmd/reloader.go

@@ -28,7 +28,6 @@ import (
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/elastic/apm-server/internal/telemetry"
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/beats/v7/libbeat/common/reload"
@@ -66,8 +65,6 @@ type RunnerParams struct {
 	MeterProvider metric.MeterProvider
 
 	MetricReader *sdkmetric.ManualReader
-
-	MetricExporter *telemetry.MetricExporter
 }
 
 // Runner is an interface returned by NewRunnerFunc.

internal/beater/beater.go

@@ -69,7 +69,6 @@ import (
 	srvmodelprocessor "github.com/elastic/apm-server/internal/model/modelprocessor"
 	"github.com/elastic/apm-server/internal/publish"
 	"github.com/elastic/apm-server/internal/sourcemap"
-	"github.com/elastic/apm-server/internal/telemetry"
 	"github.com/elastic/apm-server/internal/version"
 )
 
@@ -84,9 +83,8 @@ type Runner struct {
 	outputConfig              agentconfig.Namespace
 	elasticsearchOutputConfig *agentconfig.C
 
-	metricReader   *sdkmetric.ManualReader
-	metricExporter *telemetry.MetricExporter
-	listener       net.Listener
+	metricReader *sdkmetric.ManualReader
+	listener     net.Listener
 }
 
 // RunnerParams holds parameters for NewRunner.
@@ -98,8 +96,7 @@ type RunnerParams struct {
 	// Logger holds a logger to use for logging throughout the APM Server.
 	Logger *logp.Logger
 
-	MetricReader   *sdkmetric.ManualReader
-	MetricExporter *telemetry.MetricExporter
+	MetricReader *sdkmetric.ManualReader
 
 	// WrapServer holds an optional WrapServerFunc, for wrapping the
 	// ServerParams and RunServerFunc used to run the APM Server.
@@ -149,9 +146,8 @@ func NewRunner(args RunnerParams) (*Runner, error) {
 		outputConfig:              unpackedConfig.Output,
 		elasticsearchOutputConfig: elasticsearchOutputConfig,
 
-		metricReader:   args.MetricReader,
-		metricExporter: args.MetricExporter,
-		listener:       listener,
+		metricReader: args.MetricReader,
+		listener:     listener,
 	}, nil
 }
 
@@ -403,7 +399,6 @@ func (s *Runner) Run(ctx context.Context) error {
 		}),
 		finalBatchProcessor,
 	})
-	s.metricExporter.SetBatchProcessor(batchProcessor)
 
 	agentConfigFetcher, fetcherRunFunc, err := newAgentConfigFetcher(
 		ctx,

x-pack/apm-server/main.go

@@ -288,10 +288,9 @@ func Main() error {
 	rootCmd := newXPackRootCommand(
 		func(args beatcmd.RunnerParams) (beatcmd.Runner, error) {
 			return beater.NewRunner(beater.RunnerParams{
-				Config:         args.Config,
-				Logger:         args.Logger,
-				WrapServer:     wrapServer,
-				MetricExporter: args.MetricExporter,
+				Config:     args.Config,
+				Logger:     args.Logger,
+				WrapServer: wrapServer,
 			})
 		},
 	)