Add grpc_resolver using external discovery service

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

@@ -27,6 +27,7 @@ import (
 	"google.golang.org/grpc/resolver/manual"
 
 	"github.com/jaegertracing/jaeger/pkg/discovery"
+	"github.com/jaegertracing/jaeger/pkg/discovery/grpcresolver"
 )
 
 // ConnBuilder Struct to hold configurations
@@ -39,15 +40,9 @@ type ConnBuilder struct {
 	TLSCA         string
 	TLSServerName string
 
-	notifier discovery.Notifier
-}
-
-// WithDiscoveryNotifier sets service discovery notifier
-// TODO User should provide their own notifier so that notifier can push address updates to grpc resolver by invoking notifier.Notify(instances []string)
-// We will add integation code with custom notifier and resolver in next PR
-func (b *ConnBuilder) WithDiscoveryNotifier(n discovery.Notifier) *ConnBuilder {
-	b.notifier = n
-	return b
+	DiscoveryMinPeers int
+	Notifier          discovery.Notifier
+	Discoverer        discovery.Discoverer
 }
 
 // NewConnBuilder creates a new grpc connection builder.
@@ -81,26 +76,28 @@ func (b *ConnBuilder) CreateConnection(logger *zap.Logger) (*grpc.ClientConn, er
 		dialOptions = append(dialOptions, grpc.WithInsecure())
 	}
 
-	if b.notifier != nil {
-		return nil, errors.New("not implemented")
-	}
-	if b.CollectorHostPorts == nil {
-		return nil, errors.New("at least one collector hostPort address is required when resolver is not available")
-	}
-	if len(b.CollectorHostPorts) > 1 {
-		r, _ := manual.GenerateAndRegisterManualResolver()
-		var resolvedAddrs []resolver.Address
-		for _, addr := range b.CollectorHostPorts {
-			resolvedAddrs = append(resolvedAddrs, resolver.Address{Addr: addr})
-		}
-		r.InitialState(resolver.State{Addresses: resolvedAddrs})
-		dialTarget = r.Scheme() + ":///round_robin"
-		logger.Info("Agent is connecting to a static list of collectors", zap.String("dialTarget", dialTarget), zap.String("collector hosts", strings.Join(b.CollectorHostPorts, ",")))
+	if b.Notifier != nil && b.Discoverer != nil {
+		logger.Info("Using external discovery service with roundrobin load balancer")
+		grpcResolver := grpcresolver.New(b.Notifier, b.Discoverer, logger, b.DiscoveryMinPeers)
+		dialTarget = grpcResolver.Scheme() + ":///round_robin"
 	} else {
-		dialTarget = b.CollectorHostPorts[0]
+		if b.CollectorHostPorts == nil {
+			return nil, errors.New("at least one collector hostPort address is required when resolver is not available")
+		}
+		if len(b.CollectorHostPorts) > 1 {
+			r, _ := manual.GenerateAndRegisterManualResolver()
+			var resolvedAddrs []resolver.Address
+			for _, addr := range b.CollectorHostPorts {
+				resolvedAddrs = append(resolvedAddrs, resolver.Address{Addr: addr})
+			}
+			r.InitialState(resolver.State{Addresses: resolvedAddrs})
+			dialTarget = r.Scheme() + ":///round_robin"
+			logger.Info("Agent is connecting to a static list of collectors", zap.String("dialTarget", dialTarget), zap.String("collector hosts", strings.Join(b.CollectorHostPorts, ",")))
+		} else {
+			dialTarget = b.CollectorHostPorts[0]
+		}
 	}
 	dialOptions = append(dialOptions, grpc.WithBalancerName(roundrobin.Name))
-
 	dialOptions = append(dialOptions, grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(grpc_retry.WithMax(b.MaxRetry))))
 	return grpc.Dial(dialTarget, dialOptions...)
 }

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

@@ -78,6 +78,7 @@ func TestBuilderWithCollectors(t *testing.T) {
 		hostPorts       []string
 		checkSuffixOnly bool
 		notifier        discovery.Notifier
+		discoverer      discovery.Discoverer
 		err             error
 	}{
 		{
@@ -86,28 +87,31 @@ func TestBuilderWithCollectors(t *testing.T) {
 			hostPorts:       []string{"127.0.0.1:9876", "127.0.0.1:9877", "127.0.0.1:9878"},
 			checkSuffixOnly: true,
 			notifier:        nil,
+			discoverer:      nil,
 		},
 		{
 			target:          "127.0.0.1:9876",
 			name:            "with single host",
 			hostPorts:       []string{"127.0.0.1:9876"},
 			checkSuffixOnly: false,
 			notifier:        nil,
+			discoverer:      nil,
 		},
 		{
-			target:          "dns://random_stuff",
-			name:            "with custom resolver",
+			target:          "///round_robin",
+			name:            "with custom resolver and fixed discoverer",
 			hostPorts:       []string{"dns://random_stuff"},
-			checkSuffixOnly: false,
+			checkSuffixOnly: true,
 			notifier:        noopNotifier{},
-			err:             errors.New("not implemented"),
+			discoverer:      discovery.FixedDiscoverer{},
 		},
 		{
 			target:          "",
 			name:            "without collectorPorts and resolver",
 			hostPorts:       nil,
 			checkSuffixOnly: false,
 			notifier:        nil,
+			discoverer:      nil,
 			err:             errors.New("at least one collector hostPort address is required when resolver is not available"),
 		},
 	}
@@ -117,7 +121,8 @@ func TestBuilderWithCollectors(t *testing.T) {
 			// Use NewBuilder for code coverage consideration
 			cfg := NewConnBuilder()
 			cfg.CollectorHostPorts = test.hostPorts
-			cfg.WithDiscoveryNotifier(test.notifier)
+			cfg.Notifier = test.notifier
+			cfg.Discoverer = test.discoverer
 
 			conn, err := cfg.CreateConnection(zap.NewNop())
 			if err != nil {

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

@@ -29,6 +29,7 @@ const (
 	collectorTLS           = gRPCPrefix + "tls"
 	collectorTLSCA         = gRPCPrefix + "tls.ca"
 	collectorTLSServerName = gRPCPrefix + "tls.server-name"
+	discoveryMinPeers      = gRPCPrefix + "discovery.min-peers"
 )
 
 // AddFlags adds flags for Options.
@@ -38,6 +39,7 @@ func AddFlags(flags *flag.FlagSet) {
 	flags.Bool(collectorTLS, false, "Enable TLS.")
 	flags.String(collectorTLSCA, "", "Path to a TLS CA file. (default use the systems truststore)")
 	flags.String(collectorTLSServerName, "", "Override the TLS server name.")
+	flags.Int(discoveryMinPeers, 3, "Max number of collectors to which the agent will try to connect at any given time")
 }
 
 // InitFromViper initializes Options with properties retrieved from Viper.
@@ -50,5 +52,6 @@ func (b *ConnBuilder) InitFromViper(v *viper.Viper) *ConnBuilder {
 	b.TLS = v.GetBool(collectorTLS)
 	b.TLSCA = v.GetString(collectorTLSCA)
 	b.TLSServerName = v.GetString(collectorTLSServerName)
+	b.DiscoveryMinPeers = v.GetInt(discoveryMinPeers)
 	return b
 }

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

@@ -30,9 +30,11 @@ func TestBindFlags(t *testing.T) {
 		expected *ConnBuilder
 	}{
 		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111", "--reporter.grpc.retry.max=15"},
-			expected: &ConnBuilder{CollectorHostPorts: []string{"localhost:1111"}, MaxRetry: 15}},
+			expected: &ConnBuilder{CollectorHostPorts: []string{"localhost:1111"}, MaxRetry: 15, DiscoveryMinPeers: 3}},
 		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111,localhost:2222"},
-			expected: &ConnBuilder{CollectorHostPorts: []string{"localhost:1111", "localhost:2222"}, MaxRetry: defaultMaxRetry}},
+			expected: &ConnBuilder{CollectorHostPorts: []string{"localhost:1111", "localhost:2222"}, MaxRetry: defaultMaxRetry, DiscoveryMinPeers: 3}},
+		{cOpts: []string{"--reporter.grpc.host-port=localhost:1111,localhost:2222", "--reporter.grpc.discovery.min-peers=5"},
+			expected: &ConnBuilder{CollectorHostPorts: []string{"localhost:1111", "localhost:2222"}, MaxRetry: defaultMaxRetry, DiscoveryMinPeers: 5}},
 	}
 	for _, test := range tests {
 		v := viper.New()

cmd/agent/main.go

@@ -54,9 +54,9 @@ func main() {
 				Namespace(metrics.NSOptions{Name: "agent"})
 
 			rOpts := new(reporter.Options).InitFromViper(v)
-			tChanOpts := tchannel.NewBuilder().InitFromViper(v, logger)
-			grpcOpts := grpc.NewConnBuilder().InitFromViper(v)
-			cp, err := app.CreateCollectorProxy(rOpts, tChanOpts, grpcOpts, logger, mFactory)
+			tchanBuilder := tchannel.NewBuilder().InitFromViper(v, logger)
+			grpcBuilder := grpc.NewConnBuilder().InitFromViper(v)
+			cp, err := app.CreateCollectorProxy(rOpts, tchanBuilder, grpcBuilder, logger, mFactory)
 			if err != nil {
 				logger.Fatal("Could not create collector proxy", zap.Error(err))
 			}

cmd/all-in-one/main.go

@@ -119,12 +119,12 @@ func main() {
 
 			aOpts := new(agentApp.Builder).InitFromViper(v)
 			repOpts := new(agentRep.Options).InitFromViper(v)
-			tchannelRepOpts := agentTchanRep.NewBuilder().InitFromViper(v, logger)
-			grpcRepOpts := agentGrpcRep.NewConnBuilder().InitFromViper(v)
+			tchanBuilder := agentTchanRep.NewBuilder().InitFromViper(v, logger)
+			grpcBuilder := agentGrpcRep.NewConnBuilder().InitFromViper(v)
 			cOpts := new(collector.CollectorOptions).InitFromViper(v)
 			qOpts := new(queryApp.QueryOptions).InitFromViper(v)
 
-			startAgent(aOpts, repOpts, tchannelRepOpts, grpcRepOpts, cOpts, logger, metricsFactory)
+			startAgent(aOpts, repOpts, tchanBuilder, grpcBuilder, cOpts, logger, metricsFactory)
 			collectorSrv := startCollector(cOpts, spanWriter, logger, metricsFactory, strategyStore, svc.HC())
 			querySrv := startQuery(
 				svc, qOpts, archiveOptions(storageFactory, logger),