Fix multitenancy when using OTLP HTTP

CHANGELOG.md

@@ -37,11 +37,12 @@ query_frontend:
 ```
 * [ENHACEMENT] Update OTel collector to v0.57.2 [#1757](https://github.com/grafana/tempo/pull/1757) (@mapno)
 * [ENHANCEMENT] Vulture now has improved distribution of the random traces it searches. [#1763](https://github.com/grafana/tempo/pull/1763) (@rfratto)
-* [BUGFIX] Honor caching and buffering settings when finding traces by id [#1697](https://github.com/grafana/tempo/pull/1697) (@joe-elliott)
-* [BUGFIX] Correctly propagate errors from the iterator layer up through the queriers [#1723](https://github.com/grafana/tempo/pull/1723) (@joe-elliott)
 * [ENHANCEMENT] Upgrade opentelemetry-proto submodule to v0.18.0 [#1754](https://github.com/grafana/tempo/pull/1754) (@mapno)
 Internal types are updated to use `scope` instead of `instrumentation_library`. This is a breaking change in trace by ID queries if JSON is requested.
 * [ENHANCEMENT] Metrics generator: extract `status_message` field from spans [#1786](https://github.com/grafana/tempo/pull/1786) (@stoewer)
+* [BUGFIX] Honor caching and buffering settings when finding traces by id [#1697](https://github.com/grafana/tempo/pull/1697) (@joe-elliott)
+* [BUGFIX] Correctly propagate errors from the iterator layer up through the queriers [#1723](https://github.com/grafana/tempo/pull/1723) (@joe-elliott)
+* [BUGFIX] Make multitenancy work with HTTP [#1781](https://github.com/grafana/tempo/pull/1781) (@gouthamve)
 
 ## v1.5.0 / 2022-08-17
 

modules/distributor/receiver/middleware.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/weaveworks/common/user"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 
@@ -56,8 +57,20 @@ func (m *multiTenancyMiddleware) Wrap(next consumer.Traces) consumer.Traces {
 		var err error
 		_, ctx, err = user.ExtractFromGRPCRequest(ctx)
 		if err != nil {
-			log.Logger.Log("msg", "failed to extract org id", "err", err)
-			return err
+			// Maybe its a HTTP request.
+			info := client.FromContext(ctx)
+			orgIDs := info.Metadata.Get(user.OrgIDHeaderName)
+			if len(orgIDs) == 0 {
+				log.Logger.Log("msg", "failed to extract org id from both grpc and HTTP", "err", err)
+				return err
+			}
+
+			if len(orgIDs) > 1 {
+				log.Logger.Log("msg", "more than one orgID found", "orgIDs", orgIDs)
+				return err
+			}
+
+			ctx = user.InjectOrgID(ctx, orgIDs[0])
 		}
 		return next.ConsumeTraces(ctx, td)
 	})

modules/distributor/receiver/middleware_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
+	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"google.golang.org/grpc/metadata"
@@ -54,7 +55,7 @@ func TestFakeTenantMiddleware(t *testing.T) {
 func TestMultiTenancyMiddleware(t *testing.T) {
 	m := MultiTenancyMiddleware()
 
-	t.Run("injects org id", func(t *testing.T) {
+	t.Run("injects org id grpc", func(t *testing.T) {
 		tenantID := "test-tenant-id"
 
 		consumer := newAssertingConsumer(t, func(t *testing.T, ctx context.Context) {
@@ -70,6 +71,25 @@ func TestMultiTenancyMiddleware(t *testing.T) {
 		require.NoError(t, m.Wrap(consumer).ConsumeTraces(ctx, ptrace.Traces{}))
 	})
 
+	t.Run("injects org id http", func(t *testing.T) {
+		tenantID := "test-tenant-id"
+
+		consumer := newAssertingConsumer(t, func(t *testing.T, ctx context.Context) {
+			orgID, err := user.ExtractOrgID(ctx)
+			require.NoError(t, err)
+			require.Equal(t, orgID, tenantID)
+		})
+
+		info := client.Info{
+			Metadata: client.NewMetadata(map[string][]string{
+				"x-scope-OrgID": {tenantID},
+			}),
+		}
+
+		ctx := client.NewContext(context.Background(), info)
+		require.NoError(t, m.Wrap(consumer).ConsumeTraces(ctx, ptrace.Traces{}))
+	})
+
 	t.Run("returns error if org id cannot be extracted", func(t *testing.T) {
 		// no need to assert anything, because the wrapped function is never called
 		consumer := newAssertingConsumer(t, func(t *testing.T, ctx context.Context) {})

modules/distributor/receiver/shim.go

@@ -184,6 +184,32 @@ func New(receiverCfg map[string]interface{}, pusher BatchPusher, middleware Midd
 			return nil, fmt.Errorf("receiver factory not found for type: %s", componentID.Type())
 		}
 
+		// Make sure that the headers are added to context. Required for Authentication.
+		switch componentID.Type() {
+		case "otlp":
+			otlpRecvCfg := cfg.(*otlpreceiver.Config)
+
+			if otlpRecvCfg.HTTP != nil {
+				otlpRecvCfg.HTTP.IncludeMetadata = true
+				cfg = otlpRecvCfg
+			}
+
+		case "zipkin":
+			zipkinRecvCfg := cfg.(*zipkinreceiver.Config)
+
+			zipkinRecvCfg.HTTPServerSettings.IncludeMetadata = true
+			cfg = zipkinRecvCfg
+
+		case "jaeger":
+			jaegerRecvCfg := cfg.(*jaegerreceiver.Config)
+
+			if jaegerRecvCfg.ThriftHTTP != nil {
+				jaegerRecvCfg.ThriftHTTP.IncludeMetadata = true
+			}
+
+			cfg = jaegerRecvCfg
+		}
+
 		receiver, err := factoryBase.CreateTracesReceiver(ctx, params, cfg, middleware.Wrap(shim))
 		if err != nil {
 			return nil, err