Added round_robin balancer as an option to gRPC client settings (#1353)

* Added round_robin balancer as an option to gRPC client settings

* Added documentation changes`

* Changed the balancerName setting from bool to string to accomodate new balancers in future

Setting invalid balancer is panicking. Hence validated the same & thrown an error

* Fixed test

* Fixed tests

* Replaced grpc.WithBalancerName with grpc.WithDefaultServiceConfig

* Validated the balancerName using a var string array instead of error control flow

* Fixed lint errors

* Fixed lint errors

* typo fix in documentation

config/configgrpc/configgrpc.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/keepalive"
@@ -45,6 +46,9 @@ var (
 	}
 )
 
+// Allowed balancer names to be set in grpclb_policy to discover the servers
+var allowedBalancerNames = []string{roundrobin.Name, grpc.PickFirstBalancerName}
+
 // KeepaliveClientConfig exposes the keepalive.ClientParameters to be used by the exporter.
 // Refer to the original data-structure for the meaning of each parameter:
 // https://godoc.org/google.golang.org/grpc/keepalive#ClientParameters
@@ -89,6 +93,10 @@ type GRPCClientSettings struct {
 
 	// PerRPCAuth parameter configures the client to send authentication data on a per-RPC basis.
 	PerRPCAuth *PerRPCAuthConfig `mapstructure:"per_rpc_auth"`
+
+	// Sets the balancer in grpclb_policy to discover the servers. Default is pick_first
+	// https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md
+	BalancerName string `mapstructure:"balancer_name"`
 }
 
 type KeepaliveServerConfig struct {
@@ -154,7 +162,6 @@ type GRPCServerSettings struct {
 // ToServerOption maps configgrpc.GRPCClientSettings to a slice of dial options for gRPC
 func (gcs *GRPCClientSettings) ToDialOptions() ([]grpc.DialOption, error) {
 	opts := []grpc.DialOption{}
-
 	if gcs.Compression != "" {
 		if compressionKey := GetGRPCCompressionKey(gcs.Compression); compressionKey != CompressionUnsupported {
 			opts = append(opts, grpc.WithDefaultCallOptions(grpc.UseCompressor(compressionKey)))
@@ -200,9 +207,26 @@ func (gcs *GRPCClientSettings) ToDialOptions() ([]grpc.DialOption, error) {
 		}
 	}
 
+	if gcs.BalancerName != "" {
+		valid := validateBalancerName(gcs.BalancerName)
+		if !valid {
+			return nil, fmt.Errorf("invalid balancer_name: %s", gcs.BalancerName)
+		}
+		opts = append(opts, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, gcs.BalancerName)))
+	}
+
 	return opts, nil
 }
 
+func validateBalancerName(balancerName string) bool {
+	for _, item := range allowedBalancerNames {
+		if item == balancerName {
+			return true
+		}
+	}
+	return false
+}
+
 func (gss *GRPCServerSettings) ToListener() (net.Listener, error) {
 	return gss.NetAddr.Listen()
 }

config/configgrpc/configgrpc_test.go

@@ -60,10 +60,11 @@ func TestAllGrpcClientSettings(t *testing.T) {
 		WriteBufferSize: 1024,
 		WaitForReady:    true,
 		PerRPCAuth:      nil,
+		BalancerName:    "round_robin",
 	}
 	opts, err := gcs.ToDialOptions()
 	assert.NoError(t, err)
-	assert.Len(t, opts, 5)
+	assert.Len(t, opts, 6)
 }
 
 func TestDefaultGrpcServerSettings(t *testing.T) {
@@ -143,10 +144,34 @@ func TestGRPCClientSettingsError(t *testing.T) {
 				Keepalive: nil,
 			},
 		},
+		{
+			err: "invalid balancer_name: test",
+			settings: GRPCClientSettings{
+				Headers: map[string]string{
+					"test": "test",
+				},
+				Endpoint:    "localhost:1234",
+				Compression: "gzip",
+				TLSSetting: configtls.TLSClientSetting{
+					Insecure: false,
+				},
+				Keepalive: &KeepaliveClientConfig{
+					Time:                time.Second,
+					Timeout:             time.Second,
+					PermitWithoutStream: true,
+				},
+				ReadBufferSize:  1024,
+				WriteBufferSize: 1024,
+				WaitForReady:    true,
+				BalancerName:    "test",
+			},
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.err, func(t *testing.T) {
-			_, err := test.settings.ToDialOptions()
+			opts, err := test.settings.ToDialOptions()
+			assert.Nil(t, opts)
+			assert.Error(t, err)
 			assert.Regexp(t, test.err, err)
 		})
 	}

exporter/jaegerexporter/README.md

@@ -18,6 +18,8 @@ connection. See [grpc.WithInsecure()](https://godoc.org/google.golang.org/grpc#W
 [grpc.WithKeepaliveParams()](https://godoc.org/google.golang.org/grpc#WithKeepaliveParams).
 - `server_name_override`: If set to a non empty string, it will override the virtual host name 
 of authority (e.g. :authority header field) in requests (typically used for testing).
+- `balancer_name`(default = pick_first): Sets the balancer in grpclb_policy to discover the servers.
+See [grpc loadbalancing example](https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md).
 
 Example:
 

exporter/jaegerexporter/config_test.go

@@ -53,6 +53,7 @@ func TestLoadConfig(t *testing.T) {
 	e1 := cfg.Exporters["jaeger/2"]
 	assert.Equal(t, "jaeger/2", e1.(*Config).Name())
 	assert.Equal(t, "a.new.target:1234", e1.(*Config).Endpoint)
+	assert.Equal(t, "round_robin", e1.(*Config).GRPCClientSettings.BalancerName)
 	params := component.ExporterCreateParams{Logger: zap.NewNop()}
 	te, err := factory.CreateTraceExporter(context.Background(), params, e1)
 	require.NoError(t, err)

exporter/jaegerexporter/testdata/config.yaml

@@ -10,6 +10,8 @@ exporters:
     insecure: true
   jaeger/2:
     endpoint: "a.new.target:1234"
+    balancer_name: "round_robin"
+
 
 service:
   pipelines:

exporter/opencensusexporter/README.md

@@ -25,6 +25,8 @@ The following settings can be optionally configured:
   Optional.
 - `reconnection_delay` (default = unset): time period between each reconnection
   performed by the exporter.
+- `balancer_name`(default = pick_first): Sets the balancer in grpclb_policy to discover the servers.
+See [grpc loadbalancing example](https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md).
 
 Example:
 

exporter/opencensusexporter/config_test.go

@@ -68,6 +68,7 @@ func TestLoadConfig(t *testing.T) {
 					Timeout:             30,
 				},
 				WriteBufferSize: 512 * 1024,
+				BalancerName:    "round_robin",
 			},
 			NumWorkers:        123,
 			ReconnectionDelay: 15,

exporter/opencensusexporter/testdata/config.yaml

@@ -16,6 +16,7 @@ exporters:
       header1: 234
       another: "somevalue"
     reconnection_delay: 15
+    balancer_name: "round_robin"
     keepalive:
       time: 20
       timeout: 30

exporter/otlpexporter/README.md

@@ -26,6 +26,8 @@ The following settings can be optionally configured:
 - `insecure`: whether to enable client transport security for the exporter's
   gRPC connection. See
   [grpc.WithInsecure()](https://godoc.org/google.golang.org/grpc#WithInsecure).
+- `balancer_name`(default = pick_first): Sets the balancer in grpclb_policy to discover the servers.
+See [grpc loadbalancing example](https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md).
 
 Example:
 

exporter/otlpexporter/config_test.go

@@ -73,6 +73,7 @@ func TestLoadConfig(t *testing.T) {
 					AuthType:    "bearer",
 					BearerToken: "some-token",
 				},
+				BalancerName: "round_robin",
 			},
 		})
 }

exporter/otlpexporter/testdata/config.yaml

@@ -21,6 +21,7 @@ exporters:
       time: 20s
       timeout: 30s
       permit_without_stream: true
+    balancer_name: "round_robin"
 
 service:
   pipelines: