Implement client throttler

Signed-off-by: Isaac Hier <isaachier@gmail.com>

config/config.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2017 Uber Technologies, Inc.
+// Copyright (c) 2017-2018 Uber Technologies, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import (
 
 	"github.com/uber/jaeger-client-go"
 	"github.com/uber/jaeger-client-go/internal/baggage/remote"
+	throttler "github.com/uber/jaeger-client-go/internal/throttler/remote"
 	"github.com/uber/jaeger-client-go/rpcmetrics"
 )
 
@@ -49,6 +50,7 @@ type Configuration struct {
 	Reporter            *ReporterConfig            `yaml:"reporter"`
 	Headers             *jaeger.HeadersConfig      `yaml:"headers"`
 	BaggageRestrictions *BaggageRestrictionsConfig `yaml:"baggage_restrictions"`
+	Throttler           *ThrottlerConfig           `yaml:"throttler"`
 }
 
 // SamplerConfig allows initializing a non-default sampler.  All fields are optional.
@@ -125,6 +127,24 @@ type BaggageRestrictionsConfig struct {
 	RefreshInterval time.Duration `yaml:"refreshInterval"`
 }
 
+// ThrottlerConfig configures the throttler which can be used to throttle the
+// rate at which the client may send debug requests.
+type ThrottlerConfig struct {
+	// HostPort of jaeger-agent's credit server.
+	HostPort string `yaml:"hostPort"`
+
+	// RefreshInterval controls how often the throttler will poll jaeger-agent
+	// for more throttling credits.
+	RefreshInterval time.Duration `yaml:"refreshInterval"`
+
+	// SynchronousInitialization determines whether or not the throttler should
+	// synchronously fetch credits from the agent when an operation is seen for
+	// the first time. This should be set to true if the client will be used by
+	// a short lived service that needs to ensure that credits are fetched
+	// upfront such that sampling or throttling occurs.
+	SynchronousInitialization bool `yaml:"synchronousInitialization"`
+}
+
 type nullCloser struct{}
 
 func (*nullCloser) Close() error { return nil }
@@ -238,6 +258,21 @@ func (c Configuration) NewTracer(options ...Option) (opentracing.Tracer, io.Clos
 		tracerOptions = append(tracerOptions, jaeger.TracerOptions.BaggageRestrictionManager(mgr))
 	}
 
+	if c.Throttler != nil {
+		debugThrottler := throttler.NewThrottler(
+			c.ServiceName,
+			throttler.Options.Metrics(tracerMetrics),
+			throttler.Options.Logger(opts.logger),
+			throttler.Options.HostPort(c.Throttler.HostPort),
+			throttler.Options.RefreshInterval(c.Throttler.RefreshInterval),
+			throttler.Options.SynchronousInitialization(
+				c.Throttler.SynchronousInitialization,
+			),
+		)
+
+		tracerOptions = append(tracerOptions, jaeger.TracerOptions.DebugThrottler(debugThrottler))
+	}
+
 	tracer, closer := jaeger.NewTracer(
 		c.ServiceName,
 		sampler,

config/config_test.go

@@ -495,3 +495,19 @@ func TestServiceNameViaConfiguration(t *testing.T) {
 	assert.NoError(t, err)
 	defer closer.Close()
 }
+
+func TestTracerTags(t *testing.T) {
+	cfg := &Configuration{Tags: []opentracing.Tag{{Key: "test", Value: 123}}}
+	_, closer, err := cfg.New("test-service")
+	assert.NoError(t, err)
+	defer closer.Close()
+}
+
+func TestThrottlerDefaultConfig(t *testing.T) {
+	cfg := &Configuration{
+		Throttler: &ThrottlerConfig{},
+	}
+	_, closer, err := cfg.New("test-service")
+	assert.NoError(t, err)
+	defer closer.Close()
+}

constants.go

@@ -38,6 +38,9 @@ const (
 	// TracerIPTagKey used to report ip of the process.
 	TracerIPTagKey = "ip"
 
+	// TracerUUIDTagKey used to report UUID of the client process.
+	TracerUUIDTagKey = "uuid"
+
 	// SamplerTypeTagKey reports which sampler was used on the root span.
 	SamplerTypeTagKey = "sampler.type"
 

crossdock/endtoend/handler.go

@@ -22,16 +22,19 @@ import (
 	"time"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 
 	"github.com/uber/jaeger-client-go"
 	"github.com/uber/jaeger-client-go/config"
 	"github.com/uber/jaeger-client-go/crossdock/common"
 	"github.com/uber/jaeger-client-go/crossdock/log"
 )
 
-var (
+const (
 	defaultSamplerType = jaeger.SamplerTypeRemote
+)
 
+var (
 	endToEndConfig = config.Configuration{
 		Disabled: false,
 		Sampler: &config.SamplerConfig{
@@ -42,6 +45,10 @@ var (
 		Reporter: &config.ReporterConfig{
 			BufferFlushInterval: time.Second,
 		},
+		Throttler: &config.ThrottlerConfig{
+			SynchronousInitialization: true,
+			HostPort:                  "agent:5778",
+		},
 	}
 )
 
@@ -110,7 +117,7 @@ func (h *Handler) getTracer(samplerType string) opentracing.Tracer {
 	if !ok {
 		endToEndConfig.Sampler.Type = samplerType
 		if err := h.init(endToEndConfig); err != nil {
-			log.Printf("Failed to create tracer: %s", err.Error())
+			log.Printf("Failed to create tracer: %s\n", err.Error())
 			return nil
 		}
 		tracer, _ = h.tracers[samplerType]
@@ -138,7 +145,11 @@ func generateTraces(tracer opentracing.Tracer, r *traceRequest) {
 	for i := 0; i < r.Count; i++ {
 		span := tracer.StartSpan(r.Operation)
 		for k, v := range r.Tags {
-			span.SetTag(k, v)
+			if k == string(ext.SamplingPriority) && v == "1" {
+				span.SetTag(k, uint16(1))
+			} else {
+				span.SetTag(k, v)
+			}
 		}
 		span.Finish()
 	}

crossdock/log/logger.go

@@ -24,6 +24,6 @@ var Enabled bool
 // Printf delegates to log.Printf if Enabled == true
 func Printf(msg string, args ...interface{}) {
 	if Enabled {
-		real_log.Printf(msg, args)
+		real_log.Printf(msg, args...)
 	}
 }

glide.yaml

@@ -28,9 +28,10 @@ import:
   - metrics
 - package: github.com/pkg/errors
   version: ~0.8.0
+- package: github.com/satori/go.uuid
+  version: ^1.2.0
 testImport:
 - package: github.com/stretchr/testify
-  version: ^1.1.3
   subpackages:
   - assert
   - require