Make grpc reporter default and add retry (jaegertracing#1384)

* Make grpc reporter default and add retry Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Polish Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix port Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Polish Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Use higher retry Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Increase retry to 100 Signed-off-by: Pavol Loffay <ploffay@redhat.com>
annanay25 · Mar 1, 2019 · 25767e9 · 25767e9
1 parent d087908
commit 25767e9
Show file tree

Hide file tree

Showing 9 changed files with 46 additions and 21 deletions.
diff --git a/Gopkg.lock b/Gopkg.lock
diff --git a/Gopkg.toml b/Gopkg.toml
@@ -122,3 +122,7 @@ required = [
 [[constraint]]
   name = "github.com/Shopify/sarama"
   version = "1.20.1"
+
+[[constraint]]
+  name = "github.com/grpc-ecosystem/go-grpc-middleware"
+  version = "1.0.0"
diff --git a/cmd/agent/app/reporter/flags.go b/cmd/agent/app/reporter/flags.go
@@ -39,7 +39,7 @@ type Options struct {
 
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
-	flags.String(reporterType, string(TCHANNEL), fmt.Sprintf("Reporter type to use e.g. %s, %s", string(TCHANNEL), string(GRPC)))
+	flags.String(reporterType, string(GRPC), fmt.Sprintf("Reporter type to use e.g. %s, %s", string(GRPC), string(TCHANNEL)))
 }
 
 // InitFromViper initializes Options with properties retrieved from Viper.

diff --git a/cmd/agent/app/reporter/grpc/collector_proxy.go b/cmd/agent/app/reporter/grpc/collector_proxy.go
@@ -17,6 +17,7 @@ package grpc
 import (
 	"errors"
 
+	"github.com/grpc-ecosystem/go-grpc-middleware/retry"
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -41,19 +42,23 @@ func NewCollectorProxy(o *Options, mFactory metrics.Factory, logger *zap.Logger)
 	if len(o.CollectorHostPort) == 0 {
 		return nil, errors.New("could not create collector proxy, address is missing")
 	}
-	var conn *grpc.ClientConn
+	var target string
 	if len(o.CollectorHostPort) > 1 {
 		r, _ := manual.GenerateAndRegisterManualResolver()
 		var resolvedAddrs []resolver.Address
 		for _, addr := range o.CollectorHostPort {
 			resolvedAddrs = append(resolvedAddrs, resolver.Address{Addr: addr})
 		}
 		r.InitialAddrs(resolvedAddrs)
-		conn, _ = grpc.Dial(r.Scheme()+":///round_robin", grpc.WithInsecure(), grpc.WithBalancerName(roundrobin.Name))
+		target = r.Scheme() + ":///round_robin"
 	} else {
-		// It does not return error if the collector is not running
-		conn, _ = grpc.Dial(o.CollectorHostPort[0], grpc.WithInsecure(), grpc.WithBalancerName(roundrobin.Name))
+		target = o.CollectorHostPort[0]
 	}
+	// It does not return error if the collector is not running
+	conn, _ := grpc.Dial(target,
+		grpc.WithInsecure(),
+		grpc.WithBalancerName(roundrobin.Name),
+		grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(grpc_retry.WithMax(o.MaxRetry))))
 	grpcMetrics := mFactory.Namespace(metrics.NSOptions{Name: "", Tags: map[string]string{"protocol": "grpc"}})
 	return &ProxyBuilder{
 		conn:     conn,

diff --git a/cmd/agent/app/reporter/grpc/flags.go b/cmd/agent/app/reporter/grpc/flags.go
@@ -24,17 +24,21 @@ import (
 const (
 	gRPCPrefix        = "reporter.grpc."
 	collectorHostPort = gRPCPrefix + "host-port"
+	retry             = gRPCPrefix + "retry.max"
+	defaultMaxRetry   = 3
 )
 
 // Options Struct to hold configurations
 type Options struct {
 	// CollectorHostPort is list of host:port Jaeger Collectors.
 	CollectorHostPort []string
+	MaxRetry          uint
 }
 
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
-	flags.String(collectorHostPort, "", "(experimental) Comma-separated string representing host:port of a static list of collectors to connect to directly.")
+	flags.String(collectorHostPort, "", "Comma-separated string representing host:port of a static list of collectors to connect to directly.")
+	flags.Uint(retry, defaultMaxRetry, "Sets the maximum number of retries for a call.")
 }
 
 // InitFromViper initializes Options with properties retrieved from Viper.
@@ -43,5 +47,6 @@ func (o *Options) InitFromViper(v *viper.Viper) *Options {
 	if hostPorts != "" {
 		o.CollectorHostPort = strings.Split(hostPorts, ",")
 	}
+	o.MaxRetry = uint(v.GetInt(retry))
 	return o
 }
diff --git a/cmd/agent/app/reporter/grpc/flags_test.go b/cmd/agent/app/reporter/grpc/flags_test.go
@@ -25,23 +25,23 @@ import (
 )
 
 func TestBingFlags(t *testing.T) {
-	v := viper.New()
-	command := cobra.Command{}
-	flags := &flag.FlagSet{}
-	AddFlags(flags)
-	command.PersistentFlags().AddGoFlagSet(flags)
-	v.BindPFlags(command.PersistentFlags())
-
 	tests := []struct {
 		cOpts    []string
 		expected *Options
 	}{
-		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111"},
-			expected: &Options{CollectorHostPort: []string{"localhost:1111"}}},
+		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111", "--reporter.grpc.retry.max=15"},
+			expected: &Options{CollectorHostPort: []string{"localhost:1111"}, MaxRetry:15}},
 		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111,localhost:2222"},
-			expected: &Options{CollectorHostPort: []string{"localhost:1111", "localhost:2222"}}},
+			expected: &Options{CollectorHostPort: []string{"localhost:1111", "localhost:2222"}, MaxRetry:defaultMaxRetry}},
 	}
 	for _, test := range tests {
+		v := viper.New()
+		command := cobra.Command{}
+		flags := &flag.FlagSet{}
+		AddFlags(flags)
+		command.PersistentFlags().AddGoFlagSet(flags)
+		v.BindPFlags(command.PersistentFlags())
+
 		err := command.ParseFlags(test.cOpts)
 		require.NoError(t, err)
 		b := new(Options).InitFromViper(v)

diff --git a/cmd/collector/app/builder/builder_flags.go b/cmd/collector/app/builder/builder_flags.go
@@ -59,7 +59,7 @@ func AddFlags(flags *flag.FlagSet) {
 	flags.Int(collectorNumWorkers, app.DefaultNumWorkers, "The number of workers pulling items from the queue")
 	flags.Int(collectorPort, defaultTChannelPort, "The TChannel port for the collector service")
 	flags.Int(collectorHTTPPort, defaultHTTPPort, "The HTTP port for the collector service")
-	flags.Int(collectorGRPCPort, defaultGRPCPort, "(experimental) The gRPC port for the collector service")
+	flags.Int(collectorGRPCPort, defaultGRPCPort, "The gRPC port for the collector service")
 	flags.Int(collectorZipkinHTTPort, 0, "The HTTP port for the Zipkin collector service e.g. 9411")
 }
 

diff --git a/crossdock/jaeger-docker-compose.yml b/crossdock/jaeger-docker-compose.yml
@@ -30,9 +30,7 @@ services:
 
     jaeger-agent:
       image: jaegertracing/jaeger-agent
-      # FIXME temporarily switching back to tchannel
-      # https://github.com/jaegertracing/jaeger/issues/1229
-      command: ["--reporter.tchannel.host-port=jaeger-collector:14267"]
+      command: ["--reporter.grpc.host-port=jaeger-collector:14250", "--reporter.grpc.retry.max=100"]
       ports:
         - "5775:5775/udp"
         - "6831:6831/udp"

diff --git a/docker-compose/jaeger-docker-compose.yml b/docker-compose/jaeger-docker-compose.yml
@@ -26,7 +26,7 @@ services:
 
     jaeger-agent:
       image: jaegertracing/jaeger-agent
-      command: ["--reporter.type=grpc", "--reporter.grpc.host-port=jaeger-collector:14250"]
+      command: ["--reporter.grpc.host-port=jaeger-collector:14250"]
       ports:
         - "5775:5775/udp"
         - "6831:6831/udp"