Support collector tags, similar to agent tags

cmd/agent/app/builder_test.go

@@ -241,7 +241,7 @@ func TestCreateCollectorProxy(t *testing.T) {
 		err := command.ParseFlags(test.flags)
 		require.NoError(t, err)
 
-		rOpts := new(reporter.Options).InitFromViper(v)
+		rOpts := new(reporter.Options).InitFromViper(v, zap.NewNop())
 		tchan := tchannel.NewBuilder().InitFromViper(v, zap.NewNop())
 		grpcBuilder := grpc.NewConnBuilder().InitFromViper(v)
 

cmd/agent/app/reporter/flags.go

@@ -17,17 +17,20 @@ package reporter
 import (
 	"flag"
 	"fmt"
-	"os"
-	"strings"
 
 	"github.com/spf13/viper"
+	"go.uber.org/zap"
+
+	"github.com/jaegertracing/jaeger/cmd/all-in-one/setupcontext"
+	"github.com/jaegertracing/jaeger/cmd/flags"
 )
 
 const (
 	// Whether to use grpc or tchannel reporter.
 	reporterType = "reporter.type"
 	// Agent tags
-	agentTags = "jaeger.tags"
+	agentTagsDeprecated = "jaeger.tags"
+	agentTags           = "agent.tags"
 	// TCHANNEL is name of tchannel reporter.
 	TCHANNEL Type = "tchannel"
 	// GRPC is name of gRPC reporter.
@@ -46,51 +49,23 @@ type Options struct {
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
 	flags.String(reporterType, string(GRPC), fmt.Sprintf("Reporter type to use e.g. %s, %s", string(GRPC), string(TCHANNEL)))
-	flags.String(agentTags, "", "One or more tags to be added to the Process tags of all spans passing through this agent. Ex: key1=value1,key2=${envVar:defaultValue}")
+	if !setupcontext.IsAllInOne() {
+		flags.String(agentTagsDeprecated, "", "(deprecated) see --"+agentTags)
+		flags.String(agentTags, "", "One or more tags to be added to the Process tags of all spans passing through this agent. Ex: key1=value1,key2=${envVar:defaultValue}")
+	}
 }
 
 // InitFromViper initializes Options with properties retrieved from Viper.
-func (b *Options) InitFromViper(v *viper.Viper) *Options {
+func (b *Options) InitFromViper(v *viper.Viper, logger *zap.Logger) *Options {
 	b.ReporterType = Type(v.GetString(reporterType))
-	b.AgentTags = parseAgentTags(v.GetString(agentTags))
-	return b
-}
-
-// Parsing logic borrowed from jaegertracing/jaeger-client-go
-func parseAgentTags(agentTags string) map[string]string {
-	if agentTags == "" {
-		return nil
-	}
-	tagPairs := strings.Split(string(agentTags), ",")
-	tags := make(map[string]string)
-	for _, p := range tagPairs {
-		kv := strings.SplitN(p, "=", 2)
-		k, v := strings.TrimSpace(kv[0]), strings.TrimSpace(kv[1])
-
-		if strings.HasPrefix(v, "${") && strings.HasSuffix(v, "}") {
-			skipWhenEmpty := false
-
-			ed := strings.SplitN(string(v[2:len(v)-1]), ":", 2)
-			if len(ed) == 1 {
-				// no default value specified, set to empty
-				skipWhenEmpty = true
-				ed = append(ed, "")
-			}
-
-			e, d := ed[0], ed[1]
-			v = os.Getenv(e)
-			if v == "" && d != "" {
-				v = d
-			}
-
-			// no value is set, skip this entry
-			if v == "" && skipWhenEmpty {
-				continue
-			}
+	if !setupcontext.IsAllInOne() {
+		if len(v.GetString(agentTagsDeprecated)) > 0 {
+			logger.Warn("Using deprecated configuration", zap.String("option", agentTagsDeprecated))
+			b.AgentTags = flags.ParseJaegerTags(v.GetString(agentTagsDeprecated))
+		}
+		if len(v.GetString(agentTags)) > 0 {
+			b.AgentTags = flags.ParseJaegerTags(v.GetString(agentTags))
 		}
-
-		tags[k] = v
 	}
-
-	return tags
+	return b
 }

cmd/agent/app/reporter/flags_test.go

@@ -24,6 +24,9 @@ import (
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/jaegertracing/jaeger/cmd/all-in-one/setupcontext"
 )
 
 func TestBindFlags_NoJaegerTags(t *testing.T) {
@@ -40,7 +43,7 @@ func TestBindFlags_NoJaegerTags(t *testing.T) {
 	require.NoError(t, err)
 
 	b := &Options{}
-	b.InitFromViper(v)
+	b.InitFromViper(v, zap.NewNop())
 	assert.Equal(t, Type("grpc"), b.ReporterType)
 	assert.Len(t, b.AgentTags, 0)
 }
@@ -53,7 +56,7 @@ func TestBindFlags(t *testing.T) {
 	command.PersistentFlags().AddGoFlagSet(flags)
 	v.BindPFlags(command.PersistentFlags())
 
-	jaegerTags := fmt.Sprintf("%s,%s,%s,%s,%s,%s",
+	agentTags := fmt.Sprintf("%s,%s,%s,%s,%s,%s",
 		"key=value",
 		"envVar1=${envKey1:defaultVal1}",
 		"envVar2=${envKey2:defaultVal2}",
@@ -64,7 +67,7 @@ func TestBindFlags(t *testing.T) {
 
 	err := command.ParseFlags([]string{
 		"--reporter.type=grpc",
-		"--jaeger.tags=" + jaegerTags,
+		"--agent.tags=" + agentTags,
 	})
 	require.NoError(t, err)
 
@@ -75,7 +78,7 @@ func TestBindFlags(t *testing.T) {
 	os.Setenv("envKey4", "envVal4")
 	defer os.Unsetenv("envKey4")
 
-	b.InitFromViper(v)
+	b.InitFromViper(v, zap.NewNop())
 
 	expectedTags := map[string]string{
 		"key":     "value",
@@ -88,3 +91,31 @@ func TestBindFlags(t *testing.T) {
 	assert.Equal(t, Type("grpc"), b.ReporterType)
 	assert.Equal(t, expectedTags, b.AgentTags)
 }
+
+func TestBindFlagsAllInOne(t *testing.T) {
+
+	setupcontext.SetAllInOne()
+	defer setupcontext.UnsetAllInOne()
+
+	v := viper.New()
+	command := cobra.Command{}
+	flags := &flag.FlagSet{}
+	AddFlags(flags)
+	command.PersistentFlags().AddGoFlagSet(flags)
+	v.BindPFlags(command.PersistentFlags())
+
+	agentTags := fmt.Sprintf("%s,%s,%s,%s,%s,%s",
+		"key=value",
+		"envVar1=${envKey1:defaultVal1}",
+		"envVar2=${envKey2:defaultVal2}",
+		"envVar3=${envKey3}",
+		"envVar4=${envKey4}",
+		"envVar5=${envVar5:}",
+	)
+
+	err := command.ParseFlags([]string{
+		"--reporter.type=grpc",
+		"--agent.tags=" + agentTags,
+	})
+	require.Error(t, err)
+}

cmd/agent/main.go

@@ -56,7 +56,7 @@ func main() {
 				Namespace(metrics.NSOptions{Name: "jaeger"}).
 				Namespace(metrics.NSOptions{Name: "agent"})
 
-			rOpts := new(reporter.Options).InitFromViper(v)
+			rOpts := new(reporter.Options).InitFromViper(v, logger)
 			tchanBuilder := tchannel.NewBuilder().InitFromViper(v, logger)
 			grpcBuilder := grpc.NewConnBuilder().InitFromViper(v)
 			cp, err := app.CreateCollectorProxy(rOpts, tchanBuilder, grpcBuilder, logger, mFactory)

cmd/all-in-one/main.go

@@ -41,6 +41,7 @@ import (
 	agentRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
 	agentGrpcRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
 	agentTchanRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
+	"github.com/jaegertracing/jaeger/cmd/all-in-one/setupcontext"
 	basic "github.com/jaegertracing/jaeger/cmd/builder"
 	collectorApp "github.com/jaegertracing/jaeger/cmd/collector/app"
 	collector "github.com/jaegertracing/jaeger/cmd/collector/app/builder"
@@ -71,6 +72,9 @@ import (
 
 // all-in-one/main is a standalone full-stack jaeger backend, backed by a memory store
 func main() {
+
+	setupcontext.SetAllInOne()
+
 	svc := flags.NewService(ports.CollectorAdminHTTP)
 
 	if os.Getenv(storage.SpanStorageTypeEnvVar) == "" {
@@ -121,7 +125,7 @@ by default uses only in-memory database.`,
 			strategyStore := initSamplingStrategyStore(strategyStoreFactory, metricsFactory, logger)
 
 			aOpts := new(agentApp.Builder).InitFromViper(v)
-			repOpts := new(agentRep.Options).InitFromViper(v)
+			repOpts := new(agentRep.Options).InitFromViper(v, logger)
 			tchanBuilder := agentTchanRep.NewBuilder().InitFromViper(v, logger)
 			grpcBuilder := agentGrpcRep.NewConnBuilder().InitFromViper(v)
 			cOpts := new(collector.CollectorOptions).InitFromViper(v)

cmd/all-in-one/setupcontext/setupcontext.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2019 The Jaeger Authors.
+//
+// Licensed 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 setupcontext
+
+var isAllInOne bool
+
+// SetAllInOne sets the internal flag to all in one on.
+func SetAllInOne() {
+	isAllInOne = true
+}
+
+// UnsetAllInOne unsets the internal all-in-one flag.
+func UnsetAllInOne() {
+	isAllInOne = false
+}
+
+// IsAllInOne returns true when all in one mode is on.
+func IsAllInOne() bool {
+	return isAllInOne
+}

cmd/all-in-one/setupcontext/setupcontext_test.go

@@ -0,0 +1,28 @@
+// Copyright (c) 2019 The Jaeger Authors.
+//
+// Licensed 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 setupcontext
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSetupContext(t *testing.T) {
+	// purely for code coverage
+	SetAllInOne()
+	defer UnsetAllInOne()
+	assert.True(t, IsAllInOne())
+}

cmd/collector/app/builder/builder_flags.go

@@ -21,6 +21,7 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/jaegertracing/jaeger/cmd/collector/app"
+	"github.com/jaegertracing/jaeger/cmd/flags"
 	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/ports"
 )
@@ -31,6 +32,7 @@ const (
 	collectorPort                 = "collector.port"
 	collectorHTTPPort             = "collector.http-port"
 	collectorGRPCPort             = "collector.grpc-port"
+	collectorTags                 = "collector.tags"
 	collectorZipkinHTTPort        = "collector.zipkin.http-port"
 	collectorZipkinAllowedOrigins = "collector.zipkin.allowed-origins"
 	collectorZipkinAllowedHeaders = "collector.zipkin.allowed-headers"
@@ -56,6 +58,8 @@ type CollectorOptions struct {
 	CollectorGRPCPort int
 	// TLS configures secure transport
 	TLS tlscfg.Options
+	// CollectorTags is the string representing collector tags to append to each and every span
+	CollectorTags map[string]string
 	// CollectorZipkinHTTPPort is the port that the Zipkin collector service listens in on for http requests
 	CollectorZipkinHTTPPort int
 	// CollectorZipkinAllowedOrigins is a list of origins a cross-domain request to the Zipkin collector service can be executed from
@@ -71,6 +75,7 @@ func AddFlags(flags *flag.FlagSet) {
 	flags.Int(collectorPort, ports.CollectorTChannel, "The TChannel port for the collector service")
 	flags.Int(collectorHTTPPort, ports.CollectorHTTP, "The HTTP port for the collector service")
 	flags.Int(collectorGRPCPort, ports.CollectorGRPC, "The gRPC port for the collector service")
+	flags.String(collectorTags, "", "One or more tags to be added to the Process tags of all spans passing through this collector. Ex: key1=value1,key2=${envVar:defaultValue}")
 	flags.Int(collectorZipkinHTTPort, 0, "The HTTP port for the Zipkin collector service e.g. 9411")
 	flags.String(collectorZipkinAllowedOrigins, "*", "Comma separated list of allowed origins for the Zipkin collector service, default accepts all")
 	flags.String(collectorZipkinAllowedHeaders, "content-type", "Comma separated list of allowed headers for the Zipkin collector service, default content-type")
@@ -84,6 +89,7 @@ func (cOpts *CollectorOptions) InitFromViper(v *viper.Viper) *CollectorOptions {
 	cOpts.CollectorPort = v.GetInt(collectorPort)
 	cOpts.CollectorHTTPPort = v.GetInt(collectorHTTPPort)
 	cOpts.CollectorGRPCPort = v.GetInt(collectorGRPCPort)
+	cOpts.CollectorTags = flags.ParseJaegerTags(v.GetString(collectorTags))
 	cOpts.CollectorZipkinHTTPPort = v.GetInt(collectorZipkinHTTPort)
 	cOpts.CollectorZipkinAllowedOrigins = v.GetString(collectorZipkinAllowedOrigins)
 	cOpts.CollectorZipkinAllowedHeaders = v.GetString(collectorZipkinAllowedHeaders)

cmd/collector/app/builder/span_handler_builder.go

@@ -67,6 +67,7 @@ func (spanHb *SpanHandlerBuilder) BuildHandlers() (
 		app.Options.SpanFilter(defaultSpanFilter),
 		app.Options.NumWorkers(spanHb.collectorOpts.NumWorkers),
 		app.Options.QueueSize(spanHb.collectorOpts.QueueSize),
+		app.Options.CollectorTags(spanHb.collectorOpts.CollectorTags),
 	)
 
 	return app.NewZipkinSpanHandler(spanHb.logger, spanProcessor, zs.NewChainedSanitizer(zs.StandardSanitizers...)),

cmd/collector/app/options.go

@@ -43,6 +43,7 @@ type options struct {
 	queueSize        int
 	reportBusy       bool
 	extraFormatTypes []SpanFormat
+	collectorTags    map[string]string
 }
 
 // Option is a function that sets some option on StorageBuilder.
@@ -135,6 +136,13 @@ func (options) ExtraFormatTypes(extraFormatTypes []SpanFormat) Option {
 	}
 }
 
+// CollectorTags creates an Option that initializes the extra tags to append to the spans flowing through this collector
+func (options) CollectorTags(extraTags map[string]string) Option {
+	return func(b *options) {
+		b.collectorTags = extraTags
+	}
+}
+
 func (o options) apply(opts ...Option) options {
 	ret := options{}
 	for _, opt := range opts {

cmd/collector/app/options_test.go

@@ -40,15 +40,18 @@ func TestAllOptionSet(t *testing.T) {
 		Options.Sanitizer(func(span *model.Span) *model.Span { return span }),
 		Options.QueueSize(10),
 		Options.PreSave(func(span *model.Span) {}),
+		Options.CollectorTags(map[string]string{"extra": "tags"}),
 	)
 	assert.EqualValues(t, 5, opts.numWorkers)
 	assert.EqualValues(t, 10, opts.queueSize)
+	assert.EqualValues(t, map[string]string{"extra": "tags"}, opts.collectorTags)
 }
 
 func TestNoOptionsSet(t *testing.T) {
 	opts := Options.apply()
 	assert.EqualValues(t, DefaultNumWorkers, opts.numWorkers)
 	assert.EqualValues(t, 0, opts.queueSize)
+	assert.Nil(t, opts.collectorTags)
 	assert.False(t, opts.reportBusy)
 	assert.False(t, opts.blockingSubmit)
 	assert.NotPanics(t, func() { opts.preProcessSpans(nil) })