Adds support for agent level tag

cmd/agent/app/builder.go

@@ -230,7 +230,7 @@ func CreateCollectorProxy(
 	}
 	switch opts.ReporterType {
 	case reporter.GRPC:
-		return grpc.NewCollectorProxy(grpcRepOpts, mFactory, logger)
+		return grpc.NewCollectorProxy(grpcRepOpts, opts.AgentTags, mFactory, logger)
 	case reporter.TCHANNEL:
 		return tchannel.NewCollectorProxy(tchanRep, mFactory, logger)
 	default:

cmd/agent/app/reporter/flags.go

@@ -17,12 +17,17 @@ package reporter
 import (
 	"flag"
 	"fmt"
+	"os"
+	"strings"
 
 	"github.com/spf13/viper"
 )
 
 const (
+	// Whether to use grpc or tchannel reporter.
 	reporterType = "reporter.type"
+	// Agent tags
+	agentTags = "jaeger.tags"
 	// TCHANNEL is name of tchannel reporter.
 	TCHANNEL Type = "tchannel"
 	// GRPC is name of gRPC reporter.
@@ -35,15 +40,45 @@ type Type string
 // Options holds generic reporter configuration.
 type Options struct {
 	ReporterType Type
+	AgentTags    map[string]string
 }
 
 // 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, "", fmt.Sprintf(`One or more tags to be added to the Process tags of all spans passing through this agent.
+		Ex: --jaeger.tags=key1=value1,key2=${envVar:defaultValue}`))
 }
 
 // InitFromViper initializes Options with properties retrieved from Viper.
 func (b *Options) InitFromViper(v *viper.Viper) *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, "}") {
+			ed := strings.SplitN(string(v[2:len(v)-1]), ":", 2)
+			e, d := ed[0], ed[1]
+			v = os.Getenv(e)
+			if v == "" && d != "" {
+				v = d
+			}
+		}
+
+		tags[k] = v
+	}
+
+	return tags
+}

cmd/agent/app/reporter/flags_test.go

@@ -16,6 +16,7 @@ package reporter
 
 import (
 	"flag"
+	"os"
 	"testing"
 
 	"github.com/spf13/cobra"
@@ -24,6 +25,25 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestBindFlags_NoJaegerTags(t *testing.T) {
+	v := viper.New()
+	command := cobra.Command{}
+	flags := &flag.FlagSet{}
+	AddFlags(flags)
+	command.PersistentFlags().AddGoFlagSet(flags)
+	v.BindPFlags(command.PersistentFlags())
+
+	err := command.ParseFlags([]string{
+		"--reporter.type=grpc",
+	})
+	require.NoError(t, err)
+
+	b := &Options{}
+	b.InitFromViper(v)
+	assert.Equal(t, Type("grpc"), b.ReporterType)
+	assert.Equal(t, parseAgentTags(""), b.AgentTags)
+}
+
 func TestBindFlags(t *testing.T) {
 	v := viper.New()
 	command := cobra.Command{}
@@ -34,10 +54,20 @@ func TestBindFlags(t *testing.T) {
 
 	err := command.ParseFlags([]string{
 		"--reporter.type=grpc",
+		"--jaeger.tags=key=value,envVar1=${envKey1:defaultVal1},envVar2=${envKey2:defaultVal2}",
 	})
 	require.NoError(t, err)
 
 	b := &Options{}
+	os.Setenv("envKey1", "envVal1")
 	b.InitFromViper(v)
+
+	agentTags := make(map[string]string)
+	agentTags["key"] = "value"
+	agentTags["envVar1"] = "envVal1"
+	agentTags["envVar2"] = "defaultVal2"
+
 	assert.Equal(t, Type("grpc"), b.ReporterType)
+	assert.Equal(t, agentTags, b.AgentTags)
+	os.Unsetenv("envKey1")
 }

cmd/agent/app/reporter/grpc/collector_proxy.go

@@ -42,7 +42,7 @@ type ProxyBuilder struct {
 var systemCertPool = x509.SystemCertPool // to allow overriding in unit test
 
 // NewCollectorProxy creates ProxyBuilder
-func NewCollectorProxy(o *Options, mFactory metrics.Factory, logger *zap.Logger) (*ProxyBuilder, error) {
+func NewCollectorProxy(o *Options, agentTags map[string]string, mFactory metrics.Factory, logger *zap.Logger) (*ProxyBuilder, error) {
 	if len(o.CollectorHostPort) == 0 {
 		return nil, errors.New("could not create collector proxy, address is missing")
 	}
@@ -87,7 +87,7 @@ func NewCollectorProxy(o *Options, mFactory metrics.Factory, logger *zap.Logger)
 	grpcMetrics := mFactory.Namespace(metrics.NSOptions{Name: "", Tags: map[string]string{"protocol": "grpc"}})
 	return &ProxyBuilder{
 		conn:     conn,
-		reporter: aReporter.WrapWithMetrics(NewReporter(conn, logger), grpcMetrics),
+		reporter: aReporter.WrapWithMetrics(NewReporter(conn, agentTags, logger), grpcMetrics),
 		manager:  configmanager.WrapWithMetrics(grpcManager.NewConfigManager(conn), grpcMetrics)}, nil
 }
 

cmd/agent/app/reporter/grpc/collector_proxy_test.go

@@ -50,7 +50,7 @@ iPKnCkzNgxMzQtwdgpAOXIAqXyNibvyOAv1C+3QSMLKbuPEHaIxlCuvl1suX/g25
 -----END CERTIFICATE-----`
 
 func TestProxyBuilderMissingAddress(t *testing.T) {
-	proxy, err := NewCollectorProxy(&Options{}, metrics.NullFactory, zap.NewNop())
+	proxy, err := NewCollectorProxy(&Options{}, nil, metrics.NullFactory, zap.NewNop())
 	require.Nil(t, proxy)
 	assert.EqualError(t, err, "could not create collector proxy, address is missing")
 }
@@ -99,7 +99,7 @@ func TestProxyBuilder(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			proxy, err := NewCollectorProxy(test.proxyOptions, metrics.NullFactory, zap.NewNop())
+			proxy, err := NewCollectorProxy(test.proxyOptions, nil, metrics.NullFactory, zap.NewNop())
 			if test.expectError {
 				require.Error(t, err)
 			} else {
@@ -125,7 +125,7 @@ func TestSystemCertPoolError(t *testing.T) {
 	_, err := NewCollectorProxy(&Options{
 		CollectorHostPort: []string{"foo", "bar"},
 		TLS:               true,
-	}, nil, nil)
+	}, nil, nil, nil)
 	assert.Equal(t, fakeErr, err)
 }
 
@@ -142,7 +142,7 @@ func TestMultipleCollectors(t *testing.T) {
 	defer s2.Stop()
 
 	mFactory := metricstest.NewFactory(time.Microsecond)
-	proxy, err := NewCollectorProxy(&Options{CollectorHostPort: []string{addr1.String(), addr2.String()}}, mFactory, zap.NewNop())
+	proxy, err := NewCollectorProxy(&Options{CollectorHostPort: []string{addr1.String(), addr2.String()}}, nil, mFactory, zap.NewNop())
 	require.NoError(t, err)
 	require.NotNil(t, proxy)
 	assert.NotNil(t, proxy.GetReporter())

cmd/agent/app/reporter/grpc/reporter.go

@@ -32,14 +32,16 @@ import (
 // Reporter reports data to collector over gRPC.
 type Reporter struct {
 	collector api_v2.CollectorServiceClient
+	agentTags []model.KeyValue
 	logger    *zap.Logger
 	sanitizer zipkin2.Sanitizer
 }
 
 // NewReporter creates gRPC reporter.
-func NewReporter(conn *grpc.ClientConn, logger *zap.Logger) *Reporter {
+func NewReporter(conn *grpc.ClientConn, agentTags map[string]string, logger *zap.Logger) *Reporter {
 	return &Reporter{
 		collector: api_v2.NewCollectorServiceClient(conn),
+		agentTags: makeModelKeyValue(agentTags),
 		logger:    logger,
 		sanitizer: zipkin2.NewChainedSanitizer(zipkin2.StandardSanitizers...),
 	}
@@ -63,6 +65,7 @@ func (r *Reporter) EmitZipkinBatch(zSpans []*zipkincore.Span) error {
 }
 
 func (r *Reporter) send(spans []*model.Span, process *model.Process) error {
+	spans, process = addProcessTags(spans, process, r.agentTags)
 	batch := model.Batch{Spans: spans, Process: process}
 	req := &api_v2.PostSpansRequest{Batch: batch}
 	_, err := r.collector.PostSpans(context.Background(), req)
@@ -71,3 +74,27 @@ func (r *Reporter) send(spans []*model.Span, process *model.Process) error {
 	}
 	return err
 }
+
+// addTags appends jaeger tags for the agent to every span it sends to the collector.
+func addProcessTags(spans []*model.Span, process *model.Process, agentTags []model.KeyValue) ([]*model.Span, *model.Process) {
+	if process != nil {
+		process.Tags = append(process.Tags, agentTags...)
+		return spans, process
+	}
+	for _, span := range spans {
+		if len(agentTags) > 0 {
+			span.Process.Tags = append(span.Process.Tags, agentTags...)
+		}
+	}
+	return spans, nil
+}
+
+func makeModelKeyValue(agentTags map[string]string) []model.KeyValue {
+	tags := []model.KeyValue{}
+	for k, v := range agentTags {
+		tag := model.String(k, v)
+		tags = append(tags, tag)
+	}
+
+	return tags
+}

cmd/agent/app/reporter/grpc/reporter_test.go

@@ -58,7 +58,8 @@ func TestReporter_EmitZipkinBatch(t *testing.T) {
 	conn, err := grpc.Dial(addr.String(), grpc.WithInsecure())
 	defer conn.Close()
 	require.NoError(t, err)
-	rep := NewReporter(conn, zap.NewNop())
+
+	rep := NewReporter(conn, nil, zap.NewNop())
 
 	tm := time.Unix(158, 0)
 	a := tm.Unix() * 1000 * 1000
@@ -71,7 +72,45 @@ func TestReporter_EmitZipkinBatch(t *testing.T) {
 		{in: &zipkincore.Span{Name: "jonatan", TraceID: 1, ID: 2, Timestamp: &a, Annotations: []*zipkincore.Annotation{{Value: zipkincore.CLIENT_SEND, Host: &zipkincore.Endpoint{ServiceName: "spring"}}}},
 			expected: model.Batch{
 				Spans: []*model.Span{{TraceID: model.NewTraceID(0, 1), SpanID: model.NewSpanID(2), OperationName: "jonatan", Duration: time.Microsecond * 1,
-					Tags: model.KeyValues{{Key: "span.kind", VStr: "client", VType: model.StringType}}, Process: &model.Process{ServiceName: "spring"}, StartTime: tm.UTC()}}}},
+					Tags:    model.KeyValues{{Key: "span.kind", VStr: "client", VType: model.StringType}},
+					Process: &model.Process{ServiceName: "spring"}, StartTime: tm.UTC()}}}},
+	}
+	for _, test := range tests {
+		err = rep.EmitZipkinBatch([]*zipkincore.Span{test.in})
+		if test.err != "" {
+			assert.EqualError(t, err, test.err)
+		} else {
+			assert.Equal(t, 1, len(handler.requests))
+			assert.Equal(t, test.expected, handler.requests[0].GetBatch())
+		}
+	}
+}
+
+func TestReporter_EmitZipkinBatch_WithAgentTags(t *testing.T) {
+	handler := &mockSpanHandler{}
+	s, addr := initializeGRPCTestServer(t, func(s *grpc.Server) {
+		api_v2.RegisterCollectorServiceServer(s, handler)
+	})
+	defer s.Stop()
+	conn, err := grpc.Dial(addr.String(), grpc.WithInsecure())
+	defer conn.Close()
+	require.NoError(t, err)
+
+	agentTags := make(map[string]string)
+	agentTags["hello"] = "world"
+	rep := NewReporter(conn, agentTags, zap.NewNop())
+
+	tm := time.Unix(158, 0)
+	a := tm.Unix() * 1000 * 1000
+	tests := []struct {
+		in       *zipkincore.Span
+		expected model.Batch
+		err      string
+	}{
+		{in: &zipkincore.Span{TraceID: 1, ID: 2, Timestamp: &a, Annotations: []*zipkincore.Annotation{{Host: &zipkincore.Endpoint{ServiceName: "spring"}}}},
+			expected: model.Batch{
+				Spans: []*model.Span{{TraceID: model.NewTraceID(0, 1), SpanID: model.NewSpanID(2), Duration: time.Microsecond * 1,
+					Process: &model.Process{ServiceName: "spring", Tags: []model.KeyValue{model.String("hello", "world")}}, StartTime: tm.UTC()}}}},
 	}
 	for _, test := range tests {
 		err = rep.EmitZipkinBatch([]*zipkincore.Span{test.in})
@@ -93,7 +132,7 @@ func TestReporter_EmitBatch(t *testing.T) {
 	conn, err := grpc.Dial(addr.String(), grpc.WithInsecure())
 	defer conn.Close()
 	require.NoError(t, err)
-	rep := NewReporter(conn, zap.NewNop())
+	rep := NewReporter(conn, nil, zap.NewNop())
 
 	tm := time.Unix(158, 0)
 	tests := []struct {
@@ -115,10 +154,55 @@ func TestReporter_EmitBatch(t *testing.T) {
 	}
 }
 
+func TestReporter_EmitBatch_WithAgentTags(t *testing.T) {
+	handler := &mockSpanHandler{}
+	s, addr := initializeGRPCTestServer(t, func(s *grpc.Server) {
+		api_v2.RegisterCollectorServiceServer(s, handler)
+	})
+	defer s.Stop()
+	conn, err := grpc.Dial(addr.String(), grpc.WithInsecure())
+	defer conn.Close()
+	require.NoError(t, err)
+
+	agentTags := make(map[string]string)
+	agentTags["hello"] = "world"
+	rep := NewReporter(conn, agentTags, zap.NewNop())
+
+	tm := time.Unix(158, 0)
+	tests := []struct {
+		in       *jThrift.Batch
+		expected model.Batch
+		err      string
+	}{
+		{in: &jThrift.Batch{Process: &jThrift.Process{ServiceName: "node"}, Spans: []*jThrift.Span{{OperationName: "foo", StartTime: int64(model.TimeAsEpochMicroseconds(tm))}}},
+			expected: model.Batch{Process: &model.Process{ServiceName: "node", Tags: []model.KeyValue{model.String("hello", "world")}},
+				Spans: []*model.Span{{OperationName: "foo", StartTime: tm.UTC()}}}},
+	}
+	for _, test := range tests {
+		err = rep.EmitBatch(test.in)
+		if test.err != "" {
+			assert.EqualError(t, err, test.err)
+		} else {
+			assert.Equal(t, 1, len(handler.requests))
+			assert.Equal(t, test.expected, handler.requests[0].GetBatch())
+		}
+	}
+}
+
 func TestReporter_SendFailure(t *testing.T) {
 	conn, err := grpc.Dial("", grpc.WithInsecure())
 	require.NoError(t, err)
-	rep := NewReporter(conn, zap.NewNop())
+	rep := NewReporter(conn, nil, zap.NewNop())
 	err = rep.send(nil, nil)
 	assert.EqualError(t, err, "rpc error: code = Unavailable desc = all SubConns are in TransientFailure, latest connection error: connection error: desc = \"transport: Error while dialing dial tcp: missing address\"")
 }
+
+func TestReporter_MakeModelKeyValue(t *testing.T) {
+	stringTags := make(map[string]string)
+	stringTags["hello"] = "world"
+	jaegerTags := makeModelKeyValue(stringTags)
+	expected := []model.KeyValue{model.String("hello", "world")}
+
+	assert.Equal(t, 1, len(jaegerTags))
+	assert.Equal(t, expected, jaegerTags)
+}

cmd/agent/app/reporter/metrics_test.go

@@ -104,8 +104,7 @@ func TestMetricsReporter(t *testing.T) {
 			err := reporter.EmitBatch(&jaeger.Batch{Spans: []*jaeger.Span{{}}})
 			require.Error(t, err)
 		}, rep: &noopReporter{err: errors.New("foo")}},
-		{expectedCounters:
-			[]metricstest.ExpectedMetric{
+		{expectedCounters: []metricstest.ExpectedMetric{
 			{Name: "reporter.batches.failures", Tags: map[string]string{"format": "zipkin"}, Value: 1},
 			{Name: "reporter.spans.failures", Tags: map[string]string{"format": "zipkin"}, Value: 2},
 		}, expectedGauges: []metricstest.ExpectedMetric{