Add support for receiving traces over OTLP/gRPC

beater/grpcauth.go

@@ -0,0 +1,72 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package beater
+
+import (
+	"context"
+	"strings"
+
+	"github.com/elastic/apm-server/beater/authorization"
+	"github.com/elastic/apm-server/beater/headers"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+// newAuthUnaryServerInterceptor returns a grpc.UnaryServerInterceptor which
+// performs per-RPC auth using "Authorization" metadata for OpenTelemetry methods.
+//
+// TODO(axw) when we get rid of the standalone Jaeger port move Jaeger auth
+// handling to here, possibly by dispatching to a callback based on the method
+// name and req.
+func newAuthUnaryServerInterceptor(builder *authorization.Builder) grpc.UnaryServerInterceptor {
+	authHandler := builder.ForPrivilege(authorization.PrivilegeEventWrite.Action)
+	return func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (resp interface{}, err error) {
+		if strings.HasPrefix(info.FullMethod, "/opentelemetry") {
+			if err := verifyGRPCAuthorization(ctx, authHandler); err != nil {
+				return nil, err
+			}
+		}
+		return handler(ctx, req)
+	}
+}
+
+func verifyGRPCAuthorization(ctx context.Context, authHandler *authorization.Handler) error {
+	var authHeader string
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		if values := md.Get(headers.Authorization); len(values) > 0 {
+			authHeader = values[0]
+		}
+	}
+	auth := authHandler.AuthorizationFor(authorization.ParseAuthorizationHeader(authHeader))
+	authorized, err := auth.AuthorizedFor(ctx, authorization.ResourceInternal)
+	if err != nil {
+		return err
+	}
+	if !authorized {
+		return status.Error(codes.Unauthenticated, "unauthorized")
+	}
+	return nil
+}

beater/otlp/grpc.go

@@ -0,0 +1,75 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package otlp
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver/trace"
+	"google.golang.org/grpc"
+
+	"github.com/elastic/apm-server/beater/request"
+	"github.com/elastic/apm-server/processor/otel"
+	"github.com/elastic/apm-server/publish"
+	"github.com/elastic/beats/v7/libbeat/logp"
+	"github.com/elastic/beats/v7/libbeat/monitoring"
+)
+
+var (
+	monitoringKeys = []request.ResultID{
+		request.IDRequestCount, request.IDResponseCount, request.IDResponseErrorsCount, request.IDResponseValidCount,
+	}
+
+	gRPCConsumerRegistry      = monitoring.Default.NewRegistry("apm-server.otlp.grpc.consumer")
+	gRPCConsumerMonitoringMap = request.MonitoringMapForRegistry(gRPCConsumerRegistry, monitoringKeys)
+)
+
+// RegisterGRPCServices registers OTLP consumer services with the given gRPC server.
+func RegisterGRPCServices(grpcServer *grpc.Server, reporter publish.Reporter, logger *logp.Logger) error {
+	consumer := &monitoredConsumer{
+		consumer: &otel.Consumer{Reporter: reporter},
+		logger:   logger,
+	}
+	// TODO(axw) add support for metrics to processer/otel.Consumer, and register a metrics receiver here.
+	traceReceiver := trace.New("otlp", consumer)
+	if err := otlpreceiver.RegisterTraceReceiver(context.Background(), traceReceiver, grpcServer, nil); err != nil {
+		return errors.Wrap(err, "failed to register OTLP trace receiver")
+	}
+	return nil
+}
+
+type monitoredConsumer struct {
+	consumer *otel.Consumer
+	logger   *logp.Logger
+}
+
+// ConsumeTraces consumes OpenTelemtry trace data.
+func (c *monitoredConsumer) ConsumeTraces(ctx context.Context, traces pdata.Traces) error {
+	gRPCConsumerMonitoringMap[request.IDRequestCount].Inc()
+	defer gRPCConsumerMonitoringMap[request.IDResponseCount].Inc()
+	if err := c.consumer.ConsumeTraces(ctx, traces); err != nil {
+		gRPCConsumerMonitoringMap[request.IDResponseErrorsCount].Inc()
+		c.logger.With(logp.Error(err)).Error("ConsumeTraces returned an error")
+		return err
+	}
+	gRPCConsumerMonitoringMap[request.IDResponseValidCount].Inc()
+	return nil
+}

beater/otlp/grpc_test.go

@@ -0,0 +1,135 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package otlp_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+
+	"github.com/elastic/apm-server/beater/otlp"
+	"github.com/elastic/apm-server/publish"
+	"github.com/elastic/apm-server/transform"
+	"github.com/elastic/beats/v7/libbeat/logp"
+	"github.com/elastic/beats/v7/libbeat/monitoring"
+)
+
+var (
+	exportTraceServiceRequestType  = proto.MessageType("opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest")
+	exportTraceServiceResponseType = proto.MessageType("opentelemetry.proto.collector.trace.v1.ExportTraceServiceResponse")
+)
+
+func TestConsumeTraces(t *testing.T) {
+	var events []transform.Transformable
+	var reportError error
+	report := func(ctx context.Context, req publish.PendingReq) error {
+		events = append(events, req.Transformables...)
+		return reportError
+	}
+
+	// Send a minimal trace to verify that everything is connected properly.
+	//
+	// We intentionally do not check the published event contents; those are
+	// tested in processor/otel.
+	cannedRequest := jsonExportTraceServiceRequest(`{
+"resource_spans": [
+  {
+    "instrumentation_library_spans": [
+      {
+        "spans": [
+	  {
+	    "trace_id": "0123456789abcdef0123456789abcdef",
+	    "span_id": "945254c567a5417e",
+	    "name": "operation_name"
+	  }
+	]
+      }
+    ]
+  }
+]
+}`)
+
+	conn := newServer(t, report)
+	err := conn.Invoke(
+		context.Background(), "/opentelemetry.proto.collector.trace.v1.TraceService/Export",
+		cannedRequest, newExportTraceServiceResponse(),
+	)
+	assert.NoError(t, err)
+	assert.Len(t, events, 1)
+
+	reportError = errors.New("failed to publish events")
+	err = conn.Invoke(
+		context.Background(), "/opentelemetry.proto.collector.trace.v1.TraceService/Export",
+		cannedRequest, newExportTraceServiceResponse(),
+	)
+	assert.Error(t, err)
+	errStatus := status.Convert(err)
+	assert.Equal(t, "failed to publish events", errStatus.Message())
+	assert.Len(t, events, 2)
+
+	actual := map[string]interface{}{}
+	monitoring.GetRegistry("apm-server.otlp.grpc.consumer").Do(monitoring.Full, func(key string, value interface{}) {
+		actual[key] = value
+	})
+	assert.Equal(t, map[string]interface{}{
+		"request.count":         int64(2),
+		"response.count":        int64(2),
+		"response.errors.count": int64(1),
+		"response.valid.count":  int64(1),
+	}, actual)
+}
+
+func jsonExportTraceServiceRequest(j string) interface{} {
+	request := reflect.New(exportTraceServiceRequestType.Elem()).Interface()
+	decoder := json.NewDecoder(strings.NewReader(j))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(request); err != nil {
+		panic(err)
+	}
+	return request
+}
+
+func newExportTraceServiceResponse() interface{} {
+	return reflect.New(exportTraceServiceResponseType.Elem()).Interface()
+}
+
+func newServer(t *testing.T, report publish.Reporter) *grpc.ClientConn {
+	lis, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+
+	srv := grpc.NewServer()
+	err = otlp.RegisterGRPCServices(srv, report, logp.NewLogger("otlp_test"))
+	require.NoError(t, err)
+
+	go srv.Serve(lis)
+	t.Cleanup(srv.GracefulStop)
+	conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())
+	require.NoError(t, err)
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}

beater/server.go

@@ -35,6 +35,7 @@ import (
 	"github.com/elastic/apm-server/beater/authorization"
 	"github.com/elastic/apm-server/beater/config"
 	"github.com/elastic/apm-server/beater/jaeger"
+	"github.com/elastic/apm-server/beater/otlp"
 	"github.com/elastic/apm-server/kibana"
 	"github.com/elastic/apm-server/publish"
 )
@@ -117,26 +118,27 @@ func newServer(logger *logp.Logger, cfg *config.Config, tracer *apm.Tracer, repo
 func newGRPCServer(
 	logger *logp.Logger, cfg *config.Config, tracer *apm.Tracer, reporter publish.Reporter, tlsConfig *tls.Config,
 ) (*grpc.Server, error) {
-	// NOTE(axw) even if TLS is enabled we should not use grpc.Creds,
-	// as TLS is handled by the net/http server.
-	grpcOptions := []grpc.ServerOption{grpc.UnaryInterceptor(apmgrpc.NewUnaryServerInterceptor(
-		apmgrpc.WithRecovery(),
-		apmgrpc.WithTracer(tracer))),
-	}
-	srv := grpc.NewServer(grpcOptions...)
-
 	// TODO(axw) share auth builder with beater/api.
 	authBuilder, err := authorization.NewBuilder(cfg)
 	if err != nil {
 		return nil, err
 	}
+
+	// NOTE(axw) even if TLS is enabled we should not use grpc.Creds, as TLS is handled by the net/http server.
+	apmInterceptor := apmgrpc.NewUnaryServerInterceptor(apmgrpc.WithRecovery(), apmgrpc.WithTracer(tracer))
+	authInterceptor := newAuthUnaryServerInterceptor(authBuilder)
+	srv := grpc.NewServer(grpc.ChainUnaryInterceptor(apmInterceptor, authInterceptor))
+
 	var kibanaClient kibana.Client
 	var agentcfgFetcher *agentcfg.Fetcher
 	if cfg.Kibana.Enabled {
 		kibanaClient = kibana.NewConnectingClient(&cfg.Kibana.ClientConfig)
 		agentcfgFetcher = agentcfg.NewFetcher(kibanaClient, cfg.AgentConfig.Cache.Expiration)
 	}
 	jaeger.RegisterGRPCServices(srv, authBuilder, jaeger.ElasticAuthTag, logger, reporter, kibanaClient, agentcfgFetcher)
+	if err := otlp.RegisterGRPCServices(srv, reporter, logger); err != nil {
+		return nil, err
+	}
 	return srv, nil
 }
 

beater/server_test.go

@@ -26,14 +26,19 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"reflect"
 	"runtime"
 	"testing"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/jaegertracing/jaeger/proto-gen/api_v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -404,6 +409,40 @@ func TestServerJaegerGRPC(t *testing.T) {
 	assert.NotNil(t, result)
 }
 
+func TestServerOTLPGRPC(t *testing.T) {
+	ucfg, err := common.NewConfigFrom(m{"secret_token": "abc123"})
+	assert.NoError(t, err)
+	server, err := setupServer(t, ucfg, nil, nil)
+	require.NoError(t, err)
+	defer server.Stop()
+
+	baseURL, err := url.Parse(server.baseURL)
+	require.NoError(t, err)
+	invokeExport := func(ctx context.Context, conn *grpc.ClientConn) error {
+		// We can't use go.opentelemetry.io/otel, as it has its own generated protobuf packages
+		// which which conflict with opentelemetry-collector's. Instead, use the types registered
+		// by the opentelemetry-collector packages.
+		requestType := proto.MessageType("opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest")
+		responseType := proto.MessageType("opentelemetry.proto.collector.trace.v1.ExportTraceServiceResponse")
+		request := reflect.New(requestType.Elem()).Interface()
+		response := reflect.New(responseType.Elem()).Interface()
+		return conn.Invoke(ctx, "/opentelemetry.proto.collector.trace.v1.TraceService/Export", request, response)
+	}
+
+	conn, err := grpc.Dial(baseURL.Host, grpc.WithInsecure())
+	require.NoError(t, err)
+	defer conn.Close()
+
+	ctx := context.Background()
+	err = invokeExport(ctx, conn)
+	assert.Error(t, err)
+	assert.Equal(t, codes.Unauthenticated, status.Code(err))
+
+	ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs("Authorization", "Bearer abc123"))
+	err = invokeExport(ctx, conn)
+	assert.NoError(t, err)
+}
+
 func TestServerConfigReload(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping server test")

changelogs/head.asciidoc

@@ -20,6 +20,7 @@ https://github.com/elastic/apm-server/compare/7.11\...master[View commits]
 * Support for reloading config in Fleet mode, gracefully stopping the HTTP server and starting a new one {pull}4623[4623]
 * Add a `_doc_count` field to transaction histogram docs {pull}4647[4647]
 * Upgrade Go to 1.15.7 {pull}4663[4663]
+* OpenTelemetry Protocol (OTLP) over gRPC is now supported on the standard endpoint (8200) {pull}4677[4677]
 
 [float]
 ==== Deprecated