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:/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:/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:/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:/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: