#4147 added tenant_id tag to tracing spans

CHANGELOG.md

@@ -18,6 +18,7 @@
   * `cortex_alertmanager_state_persist_failed_total`
 * [ENHANCEMENT] Blocks storage: support ingesting exemplars.  Enabled by setting new CLI flag `-blocks-storage.tsdb.max-exemplars=<n>` or config option `blocks_storage.tsdb.max_exemplars` to positive value. #4124
 * [ENHANCEMENT] Distributor: Added distributors ring status section in the admin page. #4151
+* [ENHANCEMENT] Added `tenant_ids` tag to tracing spans #4147
 * [BUGFIX] Purger: fix `Invalid null value in condition for column range` caused by `nil` value in range for WriteBatch query. #4128
 * [BUGFIX] Ingester: fixed infrequent panic caused by a race condition between TSDB mmap-ed head chunks truncation and queries. #4176
 

pkg/querier/tenantfederation/merge_queryable.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util/concurrency"
+	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
 const (
@@ -96,6 +97,9 @@ type mergeQueryable struct {
 // Querier returns a new mergeQuerier, which aggregates results from multiple
 // underlying queriers into a single result.
 func (m *mergeQueryable) Querier(ctx context.Context, mint int64, maxt int64) (storage.Querier, error) {
+	// TODO: it's necessary to think how to override context inside querier
+	//  to mark spans created inside querier as child of a span created inside
+	//  methods of merged querier.
 	ids, queriers, err := m.callback(ctx, mint, maxt)
 	if err != nil {
 		return nil, err
@@ -133,6 +137,8 @@ type mergeQuerier struct {
 // For the label "original_" + `idLabelName it will return all the values
 // of the underlying queriers for `idLabelName`.
 func (m *mergeQuerier) LabelValues(name string, matchers ...*labels.Matcher) ([]string, storage.Warnings, error) {
+	log, _ := spanlogger.New(m.ctx, "mergeQuerier.LabelValues")
+	defer log.Span.Finish()
 	if name == m.idLabelName {
 		return m.ids, nil, nil
 	}
@@ -143,7 +149,8 @@ func (m *mergeQuerier) LabelValues(name string, matchers ...*labels.Matcher) ([]
 		name = m.idLabelName
 	}
 
-	return m.mergeDistinctStringSlice(func(ctx context.Context, q storage.Querier) ([]string, storage.Warnings, error) {
+	return m.mergeDistinctStringSlice(func(ctx context.Context,
+		q storage.Querier) ([]string, storage.Warnings, error) {
 		return q.LabelValues(name, matchers...)
 	})
 }
@@ -152,6 +159,8 @@ func (m *mergeQuerier) LabelValues(name string, matchers ...*labels.Matcher) ([]
 // queriers. It also adds the `idLabelName` and if present in the original
 // results the original `idLabelName`.
 func (m *mergeQuerier) LabelNames() ([]string, storage.Warnings, error) {
+	log, _ := spanlogger.New(m.ctx, "mergeQuerier.LabelNames")
+	defer log.Span.Finish()
 	labelNames, warnings, err := m.mergeDistinctStringSlice(func(ctx context.Context, q storage.Querier) ([]string, storage.Warnings, error) {
 		return q.LabelNames()
 	})
@@ -272,6 +281,8 @@ type selectJob struct {
 // matching. The forwarded labelSelector is not containing those that operate
 // on `idLabelName`.
 func (m *mergeQuerier) Select(sortSeries bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	log, ctx := spanlogger.New(m.ctx, "mergeQuerier.Select")
+	defer log.Span.Finish()
 	matchedValues, filteredMatchers := filterValuesByMatchers(m.idLabelName, m.ids, matchers...)
 	var jobs = make([]interface{}, len(matchedValues))
 	var seriesSets = make([]storage.SeriesSet, len(matchedValues))
@@ -305,7 +316,7 @@ func (m *mergeQuerier) Select(sortSeries bool, hints *storage.SelectHints, match
 		return nil
 	}
 
-	err := concurrency.ForEach(m.ctx, jobs, maxConcurrency, run)
+	err := concurrency.ForEach(ctx, jobs, maxConcurrency, run)
 	if err != nil {
 		return storage.ErrSeriesSet(err)
 	}

pkg/querier/tenantfederation/merge_queryable_test.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"sort"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/mocktracer"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/storage"
@@ -18,6 +21,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/querier/series"
 	"github.com/cortexproject/cortex/pkg/tenant"
+	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
 const (
@@ -37,7 +41,11 @@ func (m *mockTenantQueryableWithFilter) Querier(ctx context.Context, _, _ int64)
 		return nil, err
 	}
 
-	q := mockTenantQuerier{tenant: tenantIDs[0], extraLabels: m.extraLabels}
+	q := mockTenantQuerier{
+		tenant:      tenantIDs[0],
+		extraLabels: m.extraLabels,
+		ctx:         ctx,
+	}
 
 	// set warning if exists
 	if m.warningsByTenant != nil {
@@ -66,6 +74,7 @@ type mockTenantQuerier struct {
 
 	warnings storage.Warnings
 	queryErr error
+	ctx      context.Context
 }
 
 func (m mockTenantQuerier) matrix() model.Matrix {
@@ -135,6 +144,8 @@ func (m *mockSeriesSet) Warnings() storage.Warnings {
 }
 
 func (m mockTenantQuerier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet {
+	log, _ := spanlogger.New(m.ctx, "mockTenantQuerier.select")
+	defer log.Span.Finish()
 	var matrix model.Matrix
 
 	for _, s := range m.matrix() {
@@ -478,3 +489,66 @@ func TestSetLabelsRetainExisting(t *testing.T) {
 		assert.Equal(t, tc.expected, setLabelsRetainExisting(tc.labels, tc.additionalLabels...))
 	}
 }
+
+func TestTracingMergeQueryable(t *testing.T) {
+	mockTracer := mocktracer.New()
+	opentracing.SetGlobalTracer(mockTracer)
+	ctx := user.InjectOrgID(context.Background(), "team-a|team-b")
+
+	// set a multi tenant resolver
+	tenant.WithDefaultResolver(tenant.NewMultiResolver())
+	filter := mockTenantQueryableWithFilter{}
+	q := NewQueryable(&filter, false)
+	// retrieve querier if set
+	querier, err := q.Querier(ctx, mint, maxt)
+	require.NoError(t, err)
+
+	seriesSet := querier.Select(true, &storage.SelectHints{Start: mint,
+		End: maxt})
+
+	require.NoError(t, seriesSet.Err())
+	spans := mockTracer.FinishedSpans()
+	assertSpanExist(t, spans, "mergeQuerier.Select", expectedTag{spanlogger.TenantIDTagName,
+		[]string{"team-a", "team-b"}})
+	assertSpanExist(t, spans, "mockTenantQuerier.select", expectedTag{spanlogger.TenantIDTagName,
+		[]string{"team-a"}})
+	assertSpanExist(t, spans, "mockTenantQuerier.select", expectedTag{spanlogger.TenantIDTagName,
+		[]string{"team-b"}})
+}
+
+func assertSpanExist(t *testing.T,
+	actualSpans []*mocktracer.MockSpan,
+	name string,
+	tag expectedTag) {
+	for _, span := range actualSpans {
+		if span.OperationName == name && containsTags(span, tag) {
+			return
+		}
+	}
+	require.FailNow(t, "can not find span matching params",
+		"expected span with name `%v` and with "+
+			"tags %v to be present but it was not. actual spans: %+v",
+		name, tag, extractNameWithTags(actualSpans))
+}
+
+func extractNameWithTags(actualSpans []*mocktracer.MockSpan) []spanWithTags {
+	result := make([]spanWithTags, len(actualSpans))
+	for i, span := range actualSpans {
+		result[i] = spanWithTags{span.OperationName, span.Tags()}
+	}
+	return result
+}
+
+func containsTags(span *mocktracer.MockSpan, expectedTag expectedTag) bool {
+	return reflect.DeepEqual(span.Tag(expectedTag.key), expectedTag.values)
+}
+
+type spanWithTags struct {
+	name string
+	tags map[string]interface{}
+}
+
+type expectedTag struct {
+	key    string
+	values []string
+}

pkg/util/spanlogger/spanlogger.go

@@ -9,11 +9,16 @@ import (
 	"github.com/opentracing/opentracing-go/ext"
 	otlog "github.com/opentracing/opentracing-go/log"
 
+	"github.com/cortexproject/cortex/pkg/tenant"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 
 type loggerCtxMarker struct{}
 
+const (
+	TenantIDTagName = "tenant_ids"
+)
+
 var (
 	loggerCtxKey = &loggerCtxMarker{}
 )
@@ -34,6 +39,9 @@ func New(ctx context.Context, method string, kvps ...interface{}) (*SpanLogger,
 // retrieved with FromContext or FromContextWithFallback.
 func NewWithLogger(ctx context.Context, l log.Logger, method string, kvps ...interface{}) (*SpanLogger, context.Context) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, method)
+	if ids, _ := tenant.TenantIDs(ctx); len(ids) > 0 {
+		span.SetTag(TenantIDTagName, ids)
+	}
 	logger := &SpanLogger{
 		Logger: log.With(util_log.WithContext(ctx, l), "method", method),
 		Span:   span,

pkg/util/spanlogger/spanlogger_test.go

@@ -5,8 +5,11 @@ import (
 	"testing"
 
 	"github.com/go-kit/kit/log"
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/mocktracer"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/common/user"
 )
 
 func TestSpanLogger_Log(t *testing.T) {
@@ -44,6 +47,27 @@ func TestSpanLogger_CustomLogger(t *testing.T) {
 	require.Equal(t, expect, logged)
 }
 
+func TestSpanCreatedWithTenantTag(t *testing.T) {
+	mockSpan := createSpan(user.InjectOrgID(context.Background(), "team-a"))
+
+	require.Equal(t, []string{"team-a"}, mockSpan.Tag(TenantIDTagName))
+}
+
+func TestSpanCreatedWithoutTenantTag(t *testing.T) {
+	mockSpan := createSpan(context.Background())
+
+	_, exist := mockSpan.Tags()[TenantIDTagName]
+	require.False(t, exist)
+}
+
+func createSpan(ctx context.Context) *mocktracer.MockSpan {
+	mockTracer := mocktracer.New()
+	opentracing.SetGlobalTracer(mockTracer)
+
+	logger, _ := New(ctx, "name")
+	return logger.Span.(*mocktracer.MockSpan)
+}
+
 type funcLogger func(keyvals ...interface{}) error
 
 func (f funcLogger) Log(keyvals ...interface{}) error {