gRPC load balancing (#7133)

* gRPC load balancing
* Fix typo
* Merge branch 'master' into 5132-gRPC-client-load-balancing
* Add doc about load balancer
* Update validator/client/multiple_endpoints_grpc_resolver.go
* Merge branch 'master' into 5132-gRPC-client-load-balancing
* gofmt
* Merge branch 'master' into 5132-gRPC-client-load-balancing
* Merge branch 'master' into 5132-gRPC-client-load-balancing

Author: eklm

validator/client/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "log.go",
         "metrics.go",
         "mock_validator.go",
+        "multiple_endpoints_grpc_resolver.go",
         "propose.go",
         "propose_protect.go",
         "runner.go",
@@ -60,6 +61,7 @@ go_library(
         "@org_golang_google_grpc//codes:go_default_library",
         "@org_golang_google_grpc//credentials:go_default_library",
         "@org_golang_google_grpc//metadata:go_default_library",
+        "@org_golang_google_grpc//resolver:go_default_library",
         "@org_golang_google_grpc//status:go_default_library",
     ],
 )

validator/client/multiple_endpoints_grpc_resolver.go

@@ -0,0 +1,46 @@
+package client
+
+import (
+	"strings"
+
+	"google.golang.org/grpc/resolver"
+)
+
+// Modification of a default grpc passthrough resolver (google.golang.org/grpc/resolver/passthrough) allowing to use multiple addresses
+// in grpc endpoint. Example:
+// conn, err := grpc.DialContext(ctx, "127.0.0.1:4000,127.0.0.1:4001", grpc.WithInsecure(), grpc.WithResolvers(&multipleEndpointsGrpcResolverBuilder{}))
+// It can be used with any grpc load balancer (pick_first, round_robin). Default is pick_first.
+// Round robin can be used by adding the following option:
+// grpc.WithDefaultServiceConfig("{\"loadBalancingConfig\":[{\"round_robin\":{}}]}")
+type multipleEndpointsGrpcResolverBuilder struct{}
+
+func (*multipleEndpointsGrpcResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	r := &multipleEndpointsGrpcResolver{
+		target: target,
+		cc:     cc,
+	}
+	r.start()
+	return r, nil
+}
+
+func (*multipleEndpointsGrpcResolverBuilder) Scheme() string {
+	return resolver.GetDefaultScheme()
+}
+
+type multipleEndpointsGrpcResolver struct {
+	target resolver.Target
+	cc     resolver.ClientConn
+}
+
+func (r *multipleEndpointsGrpcResolver) start() {
+	endpoints := strings.Split(r.target.Endpoint, ",")
+	var addrs []resolver.Address
+	for _, endpoint := range endpoints {
+		addrs = append(addrs, resolver.Address{Addr: endpoint})
+	}
+	r.cc.UpdateState(resolver.State{Addresses: addrs})
+}
+
+func (*multipleEndpointsGrpcResolver) ResolveNow(o resolver.ResolveNowOptions) {}
+
+func (*multipleEndpointsGrpcResolver) Close() {}

validator/client/service.go

@@ -346,6 +346,7 @@ func ConstructDialOptions(
 			grpc_prometheus.StreamClientInterceptor,
 			grpc_retry.StreamClientInterceptor(),
 		),
+		grpc.WithResolvers(&multipleEndpointsGrpcResolverBuilder{}),
 	}
 
 	dialOpts = append(dialOpts, extraOpts...)