feat: replace apmelasticsearch with otel tracing

go.mod

@@ -29,7 +29,6 @@ require (
 	github.com/spf13/cobra v1.10.1
 	github.com/spf13/pflag v1.0.10
 	github.com/stretchr/testify v1.11.1
-	go.elastic.co/apm/module/apmelasticsearch/v2 v2.7.1
 	go.elastic.co/apm/module/apmhttp/v2 v2.7.1
 	go.elastic.co/apm/module/apmotel/v2 v2.7.1
 	go.elastic.co/apm/v2 v2.7.1
@@ -224,6 +223,7 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	github.com/zclconf/go-cty v1.15.0 // indirect
 	gitlab.com/digitalxero/go-conventional-commit v1.0.7 // indirect
+	go.elastic.co/apm/module/apmelasticsearch/v2 v2.7.1 // indirect
 	go.elastic.co/ecszap v1.0.3 // indirect
 	go.elastic.co/go-licence-detector v0.7.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.0 // indirect

internal/beater/api/mux_intake_rum_test.go

@@ -45,7 +45,7 @@ func TestOPTIONS(t *testing.T) {
 
 	cfg := cfgEnabledRUM()
 	cfg.RumConfig.AllowOrigins = []string{"*"}
-	authenticator, _ := auth.NewAuthenticator(cfg.AgentAuth, logptest.NewTestingLogger(t, ""))
+	authenticator, _ := auth.NewAuthenticator(cfg.AgentAuth, tracenoop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 
 	lastMiddleware := func(h request.Handler) (request.Handler, error) {
 		return func(c *request.Context) {

internal/beater/api/mux_test.go

@@ -173,7 +173,7 @@ func (m muxBuilder) build(cfg *config.Config) (sdkmetric.Reader, http.Handler, e
 
 	nopBatchProcessor := modelpb.ProcessBatchFunc(func(context.Context, *modelpb.Batch) error { return nil })
 	ratelimitStore, _ := ratelimit.NewStore(1000, 1000, 1000)
-	authenticator, _ := auth.NewAuthenticator(cfg.AgentAuth, m.Logger)
+	authenticator, _ := auth.NewAuthenticator(cfg.AgentAuth, noop.NewTracerProvider(), m.Logger)
 	r, err := NewMux(
 		cfg,
 		nopBatchProcessor,

internal/beater/auth/anonymous_test.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 
 	"github.com/elastic/apm-server/internal/beater/auth"
 	"github.com/elastic/apm-server/internal/beater/config"
@@ -133,7 +134,7 @@ func getAnonymousAuthorizer(t testing.TB, cfg config.AnonymousAgentAuth) auth.Au
 	authenticator, err := auth.NewAuthenticator(config.AgentAuth{
 		SecretToken: "whatever", // required to enable anonymous auth
 		Anonymous:   cfg,
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	_, authorizer, err := authenticator.Authenticate(context.Background(), "", "")
 	require.NoError(t, err)

internal/beater/auth/apikey_test.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 
 	"github.com/elastic/apm-server/internal/beater/config"
 	"github.com/elastic/apm-server/internal/beater/headers"
@@ -52,7 +53,7 @@ func TestAPIKeyAuthorizer(t *testing.T) {
 	esConfig := elasticsearch.DefaultConfig()
 	esConfig.Hosts = elasticsearch.Hosts{srv.URL}
 	apikeyAuthConfig := config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 1, ESConfig: esConfig}
-	authenticator, err := NewAuthenticator(config.AgentAuth{APIKey: apikeyAuthConfig}, logptest.NewTestingLogger(t, ""))
+	authenticator, err := NewAuthenticator(config.AgentAuth{APIKey: apikeyAuthConfig}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
 	credentials := base64.StdEncoding.EncodeToString([]byte("valid_id:key_value"))

internal/beater/auth/authenticator.go

@@ -24,6 +24,8 @@ import (
 	"fmt"
 	"time"
 
+	"go.opentelemetry.io/otel/trace"
+
 	"github.com/elastic/apm-server/internal/beater/config"
 	"github.com/elastic/apm-server/internal/beater/headers"
 	"github.com/elastic/apm-server/internal/elasticsearch"
@@ -141,7 +143,7 @@ type APIKeyAuthenticationDetails struct {
 
 // NewAuthenticator creates an Authenticator with config, authenticating
 // clients with one of the allowed methods.
-func NewAuthenticator(cfg config.AgentAuth, logger *logp.Logger) (*Authenticator, error) {
+func NewAuthenticator(cfg config.AgentAuth, tp trace.TracerProvider, logger *logp.Logger) (*Authenticator, error) {
 	b := Authenticator{secretToken: cfg.SecretToken}
 	if cfg.APIKey.Enabled {
 		// Do not use apm-server's credentials for API Key requests;
@@ -150,7 +152,11 @@ func NewAuthenticator(cfg config.AgentAuth, logger *logp.Logger) (*Authenticator
 		cfg.APIKey.ESConfig.Username = ""
 		cfg.APIKey.ESConfig.Password = ""
 		cfg.APIKey.ESConfig.APIKey = ""
-		client, err := elasticsearch.NewClient(cfg.APIKey.ESConfig, logger)
+		client, err := elasticsearch.NewClientParams(elasticsearch.ClientParams{
+			Config:         cfg.APIKey.ESConfig,
+			Logger:         logger,
+			TracerProvider: tp,
+		})
 		if err != nil {
 			return nil, err
 		}

internal/beater/auth/authenticator_test.go

@@ -25,12 +25,14 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.elastic.co/apm/v2/apmtest"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/trace/noop"
 
 	"github.com/elastic/apm-server/internal/beater/config"
 	"github.com/elastic/apm-server/internal/beater/headers"
@@ -39,7 +41,7 @@ import (
 )
 
 func TestAuthenticatorNone(t *testing.T) {
-	authenticator, err := NewAuthenticator(config.AgentAuth{}, logptest.NewTestingLogger(t, ""))
+	authenticator, err := NewAuthenticator(config.AgentAuth{}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
 	// If the server has no configured auth methods, all requests are allowed.
@@ -57,7 +59,7 @@ func TestAuthenticatorAuthRequired(t *testing.T) {
 		APIKey: config.APIKeyAgentAuth{Enabled: true, ESConfig: elasticsearch.DefaultConfig()},
 	}
 	for _, cfg := range []config.AgentAuth{withSecretToken, withAPIKey} {
-		authenticator, err := NewAuthenticator(cfg, logptest.NewTestingLogger(t, ""))
+		authenticator, err := NewAuthenticator(cfg, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 		require.NoError(t, err)
 
 		details, authz, err := authenticator.Authenticate(context.Background(), "", "")
@@ -77,7 +79,7 @@ func TestAuthenticatorAuthRequired(t *testing.T) {
 }
 
 func TestAuthenticatorSecretToken(t *testing.T) {
-	authenticator, err := NewAuthenticator(config.AgentAuth{SecretToken: "valid"}, logptest.NewTestingLogger(t, ""))
+	authenticator, err := NewAuthenticator(config.AgentAuth{SecretToken: "valid"}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
 	details, authz, err := authenticator.Authenticate(context.Background(), headers.Bearer, "invalid")
@@ -115,7 +117,7 @@ func TestAuthenticatorAPIKey(t *testing.T) {
 	esConfig.Hosts = elasticsearch.Hosts{srv.URL}
 	authenticator, err := NewAuthenticator(config.AgentAuth{
 		APIKey: config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 100, ESConfig: esConfig},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
 	credentials := base64.StdEncoding.EncodeToString([]byte("id_value:key_value"))
@@ -146,7 +148,7 @@ func TestAuthenticatorAPIKeyErrors(t *testing.T) {
 	esConfig.Backoff.Max = time.Nanosecond
 	authenticator, err := NewAuthenticator(config.AgentAuth{
 		APIKey: config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 100, ESConfig: esConfig},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
 	// Make sure that we can't auth with an empty secret token if secret token auth is not configured, but API Key auth is.
@@ -186,7 +188,7 @@ func TestAuthenticatorAPIKeyErrors(t *testing.T) {
 	esConfig.Hosts = elasticsearch.Hosts{srv.URL}
 	authenticator, err = NewAuthenticator(config.AgentAuth{
 		APIKey: config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 2, ESConfig: esConfig},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	details, authz, err = authenticator.Authenticate(context.Background(), headers.APIKey, credentials)
 	assert.Equal(t, ErrAuthFailed, err)
@@ -206,7 +208,7 @@ func TestAuthenticatorAPIKeyErrors(t *testing.T) {
 	esConfig.Hosts = elasticsearch.Hosts{srv.URL}
 	authenticator, err = NewAuthenticator(config.AgentAuth{
 		APIKey: config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 100, ESConfig: esConfig},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	details, authz, err = authenticator.Authenticate(context.Background(), headers.APIKey, credentials)
 	assert.Equal(t, ErrAuthFailed, err)
@@ -241,35 +243,35 @@ func TestAuthenticatorAPIKeyCache(t *testing.T) {
 	}))
 	defer srv.Close()
 
+	exporter := &manualExporter{}
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(exporter)))
+
 	esConfig := elasticsearch.DefaultConfig()
 	esConfig.Hosts = elasticsearch.Hosts{srv.URL}
 	apikeyAuthConfig := config.APIKeyAgentAuth{Enabled: true, LimitPerMin: 2, ESConfig: esConfig}
-	authenticator, err := NewAuthenticator(config.AgentAuth{APIKey: apikeyAuthConfig}, logptest.NewTestingLogger(t, ""))
+	authenticator, err := NewAuthenticator(config.AgentAuth{APIKey: apikeyAuthConfig}, tp, logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 
-	_, spans, _ := apmtest.WithTransaction(func(ctx context.Context) {
-		for i := 0; i < apikeyAuthConfig.LimitPerMin+1; i++ {
-			_, _, err := authenticator.Authenticate(ctx, headers.APIKey, validCredentials)
-			assert.NoError(t, err)
-		}
-	})
+	for i := 0; i < apikeyAuthConfig.LimitPerMin+1; i++ {
+		_, _, err := authenticator.Authenticate(context.Background(), headers.APIKey, validCredentials)
+		assert.NoError(t, err)
+	}
+	spans := exporter.payloads()
 	assert.Len(t, spans, 1)
-	assert.Equal(t, "elasticsearch", spans[0].Subtype)
+	assert.Equal(t, "Elasticsearch: GET _security/user/_has_privileges", spans[0].Name())
+	exporter.clear()
 
-	_, spans, _ = apmtest.WithTransaction(func(ctx context.Context) {
-		// API Key checks are cached based on the API Key ID, not the full credential.
-		_, _, err := authenticator.Authenticate(ctx, headers.APIKey, validCredentials2)
-		assert.NoError(t, err)
-	})
-	assert.Len(t, spans, 0)
+	// API Key checks are cached based on the API Key ID, not the full credential.
+	_, _, err = authenticator.Authenticate(context.Background(), headers.APIKey, validCredentials2)
+	assert.NoError(t, err)
+	assert.Len(t, exporter.payloads(), 0)
+	exporter.clear()
 
-	_, spans, _ = apmtest.WithTransaction(func(ctx context.Context) {
-		for i := 0; i < apikeyAuthConfig.LimitPerMin+1; i++ {
-			_, _, err = authenticator.Authenticate(ctx, headers.APIKey, invalidCredentials)
-			assert.Equal(t, ErrAuthFailed, err)
-		}
-	})
-	assert.Len(t, spans, 1)
+	for i := 0; i < apikeyAuthConfig.LimitPerMin+1; i++ {
+		_, _, err = authenticator.Authenticate(context.Background(), headers.APIKey, invalidCredentials)
+		assert.Equal(t, ErrAuthFailed, err)
+	}
+	assert.Len(t, exporter.payloads(), 1)
 
 	credentials := base64.StdEncoding.EncodeToString([]byte("id_value3:key_value"))
 	_, _, err = authenticator.Authenticate(context.Background(), headers.APIKey, credentials)
@@ -280,7 +282,7 @@ func TestAuthenticatorAnonymous(t *testing.T) {
 	// Anonymous access is only effective when some other auth method is enabled.
 	authenticator, err := NewAuthenticator(config.AgentAuth{
 		Anonymous: config.AnonymousAgentAuth{Enabled: true},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	details, authz, err := authenticator.Authenticate(context.Background(), "", "")
 	assert.NoError(t, err)
@@ -290,10 +292,38 @@ func TestAuthenticatorAnonymous(t *testing.T) {
 	authenticator, err = NewAuthenticator(config.AgentAuth{
 		SecretToken: "secret_token",
 		Anonymous:   config.AnonymousAgentAuth{Enabled: true},
-	}, logptest.NewTestingLogger(t, ""))
+	}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	details, authz, err = authenticator.Authenticate(context.Background(), "", "")
 	assert.NoError(t, err)
 	assert.Equal(t, AuthenticationDetails{Method: MethodAnonymous}, details)
 	assert.Equal(t, newAnonymousAuth(nil, nil), authz)
 }
+
+type manualExporter struct {
+	mu    sync.Mutex
+	spans []sdktrace.ReadOnlySpan
+}
+
+func (e *manualExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.spans = append(e.spans, spans...)
+	return nil
+}
+
+func (e *manualExporter) Shutdown(ctx context.Context) error {
+	return nil
+}
+
+func (e *manualExporter) payloads() []sdktrace.ReadOnlySpan {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	return e.spans
+}
+
+func (e *manualExporter) clear() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.spans = nil
+}

internal/beater/beater.go

@@ -368,7 +368,7 @@ func (s *Runner) Run(ctx context.Context) error {
 	// Create the runServer function. We start with newBaseRunServer, and then
 	// wrap depending on the configuration in order to inject behaviour.
 	runServer := newBaseRunServer(s.listener)
-	authenticator, err := auth.NewAuthenticator(s.config.AgentAuth, s.logger)
+	authenticator, err := auth.NewAuthenticator(s.config.AgentAuth, s.tracerProvider, s.logger)
 	if err != nil {
 		return err
 	}
@@ -663,7 +663,11 @@ func (s *Runner) waitReady(
 		if err != nil {
 			return err
 		}
-		esOutputClient, err = elasticsearch.NewClient(esConfig, s.logger)
+		esOutputClient, err = elasticsearch.NewClientParams(elasticsearch.ClientParams{
+			Config:         esConfig,
+			Logger:         s.logger,
+			TracerProvider: s.tracerProvider,
+		})
 		if err != nil {
 			return err
 		}

internal/beater/interceptors/auth_test.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/trace/noop"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
@@ -117,7 +118,7 @@ func TestUnaryAuthenticator(t *testing.T) {
 }
 
 func TestAuthorizationMetadataAuthenticator(t *testing.T) {
-	authenticator, err := auth.NewAuthenticator(config.AgentAuth{SecretToken: "abc123"}, logptest.NewTestingLogger(t, ""))
+	authenticator, err := auth.NewAuthenticator(config.AgentAuth{SecretToken: "abc123"}, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
 	interceptor := interceptors.Auth(authenticator)
 

internal/beater/otlp/http_test.go

@@ -170,7 +170,7 @@ func newHTTPServer(t *testing.T, batchProcessor modelpb.BatchProcessor) (string,
 	))
 	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
 	cfg := &config.Config{}
-	auth, _ := auth.NewAuthenticator(cfg.AgentAuth, logptest.NewTestingLogger(t, ""))
+	auth, _ := auth.NewAuthenticator(cfg.AgentAuth, noop.NewTracerProvider(), logptest.NewTestingLogger(t, ""))
 	ratelimitStore, _ := ratelimit.NewStore(1000, 1000, 1000)
 	router, err := api.NewMux(
 		cfg,

internal/beater/tracing.go

@@ -42,7 +42,7 @@ func newTracerServer(cfg *config.Config, listener net.Listener, logger *logp.Log
 	if err != nil {
 		return nil, err
 	}
-	authenticator, err := auth.NewAuthenticator(config.AgentAuth{}, logger)
+	authenticator, err := auth.NewAuthenticator(config.AgentAuth{}, nooptrace.NewTracerProvider(), logger)
 	if err != nil {
 		return nil, err
 	}

internal/elasticsearch/client.go

@@ -24,9 +24,7 @@ import (
 	"io"
 	"net/http"
 
-	"go.elastic.co/apm/module/apmelasticsearch/v2"
 	"go.opentelemetry.io/otel/trace"
-	"go.opentelemetry.io/otel/trace/noop"
 
 	"github.com/elastic/apm-server/internal/version"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -114,12 +112,7 @@ func NewClientParams(args ClientParams) (*Client, error) {
 		apikey = base64.StdEncoding.EncodeToString([]byte(args.Config.APIKey))
 	}
 
-	if _, ok := args.TracerProvider.(noop.TracerProvider); !ok {
-		// only enable tracing with apm agent if a non-noop tracerprovider
-		// has been passed.
-		// TODO replace apmelasticsearch with otel (https://github.com/elastic/apm-server/issues/18949)
-		transport = apmelasticsearch.WrapRoundTripper(transport)
-	}
+	transport = WrapRoundTripper(transport, args.TracerProvider)
 
 	return elastictransport.New(elastictransport.Config{
 		APIKey:        apikey,