fix(grpc): use grpc-consul-resolver to connect service

src/utils/consul/register.go

@@ -6,9 +6,7 @@ import (
 	"fmt"
 	capi "github.com/hashicorp/consul/api"
 	log "github.com/sirupsen/logrus"
-	"math/rand"
 	"strconv"
-	"time"
 )
 
 var consulClient *capi.Client
@@ -51,41 +49,3 @@ func RegisterConsul(name string, port string) error {
 	}
 	return nil
 }
-
-func ResolveService(serviceName string) (*capi.CatalogService, error) {
-	serviceName = config.EnvCfg.ConsulAnonymityPrefix + serviceName
-
-	for {
-		instances, err := getServiceInstances(serviceName)
-		if err != nil || len(instances) == 0 {
-			logging.Logger.Errorf("Cannot find service: %s", serviceName)
-		}
-
-		selectedInstance := roundRobin(instances)
-		if selectedInstance != nil {
-			return selectedInstance, nil
-		}
-		time.Sleep(time.Second)
-	}
-}
-
-func getServiceInstances(serviceName string) ([]*capi.CatalogService, error) {
-
-	services, _, err := consulClient.Catalog().Service(serviceName, "", nil)
-	if err != nil {
-		return nil, err
-	}
-
-	return services, nil
-}
-
-func roundRobin(instances []*capi.CatalogService) *capi.CatalogService {
-	if len(instances) == 0 {
-		return nil
-	}
-
-	rand.NewSource(time.Now().UnixNano())
-	index := rand.Intn(len(instances))
-
-	return instances[index]
-}

src/utils/grpc/connection.go

@@ -1,10 +1,10 @@
 package grpc
 
 import (
-	"GuGoTik/src/utils/consul"
+	"GuGoTik/src/constant/config"
 	"GuGoTik/src/utils/logging"
 	"fmt"
-	capi "github.com/hashicorp/consul/api"
+	_ "github.com/mbobakov/grpc-consul-resolver"
 	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
@@ -14,37 +14,27 @@ import (
 )
 
 func Connect(serviceName string) (conn *grpc.ClientConn) {
-	service, err := consul.ResolveService(serviceName)
-	if err != nil {
-		logging.Logger.WithFields(logrus.Fields{
-			"service": serviceName,
-			"err":     err,
-		}).Errorf("Cannot find %v rpc server", serviceName)
-	}
-
-	logging.Logger.Debugf("Found service %v in %v:%v", service.ServiceName, service.Address, service.ServicePort)
-
-	conn, err = connect(service)
-	if err != nil {
-		logging.Logger.WithFields(logrus.Fields{
-			"service": service.ServiceName,
-			"err":     err,
-		}).Errorf("Cannot connect to %v rpc server in %v:%v", service.ServiceName, service.Address, service.ServicePort)
-	}
-	return
-}
-
-func connect(service *capi.CatalogService) (conn *grpc.ClientConn, err error) {
-	var kacp = keepalive.ClientParameters{
+	kacp := keepalive.ClientParameters{
 		Time:                10 * time.Second, // send pings every 10 seconds if there is no activity
 		Timeout:             time.Second,      // wait 1 second for ping ack before considering the connection dead
 		PermitWithoutStream: false,            // send pings even without active streams
 	}
 
-	conn, err = grpc.Dial(fmt.Sprintf("%v:%v", service.Address, service.ServicePort),
+	conn, err := grpc.Dial(
+		fmt.Sprintf("consul://%s/%s?wait=15s", config.EnvCfg.ConsulAddr, config.EnvCfg.ConsulAnonymityPrefix+serviceName),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithKeepaliveParams(kacp),
 		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
-		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()))
+		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
+		grpc.WithKeepaliveParams(kacp),
+	)
+
+	logging.Logger.Debugf("connect")
+
+	if err != nil {
+		logging.Logger.WithFields(logrus.Fields{
+			"service": config.EnvCfg.ConsulAnonymityPrefix + serviceName,
+			"err":     err,
+		}).Errorf("Cannot connect to %v service", config.EnvCfg.ConsulAnonymityPrefix+serviceName)
+	}
 	return
 }