fix: keeper connect timeout if not leader && FetchSchemerFromOtherNode not work if zoopath exists

Author: YenchangChan

192.168.110.48:7890

@@ -44,7 +44,7 @@ func ZCntHandle(opts ZCntOpts) {
 		zkPort, _ = strconv.Atoi(port)
 	}
 
-	service, err := zookeeper.NewZkService(hosts, zkPort, 300)
+	service, err := zookeeper.NewZkService(hosts, zkPort, 30)
 	if err != nil {
 		log.Logger.Fatalf("can't create zookeeper instance:%v", err)
 	}

cmd/znodes/count.go

@@ -36,7 +36,7 @@ func SuballHandle(opts ZSuballOpts) {
 		log.Logger.Fatalf("get cluster %s failed:%v", opts.ClusterName, err)
 	}
 	nodes, port := zookeeper.GetZkInfo(&cluster)
-	service, err := zookeeper.NewZkService(nodes, port, 300)
+	service, err := zookeeper.NewZkService(nodes, port, 10)
 	if err != nil {
 		log.Logger.Fatalf("can't create zookeeper instance:%v", err)
 	}

cmd/znodes/suball.go

@@ -1929,7 +1929,7 @@ func (controller *ClickHouseController) StartNode(c *gin.Context) {
 	}
 	conf.Watch(ip)
 	if host != "" {
-		ckService := clickhouse.NewCkService(&conf)
+		ckService := clickhouse.NewCkService(&con)
 		ckService.InitCkService()
 		err := ckService.FetchSchemerFromOtherNode(host)
 		if err != nil {

controller/clickhouse.go

@@ -588,15 +588,17 @@ func (ck *CkService) FetchSchemerFromOtherNode(host string) error {
 	}
 
 	num := len(names)
+	var lastErr error
 	for i := 0; i < num; i++ {
 		log.Logger.Debugf("statement: %s", statements[i])
 		if err := ck.Conn.Exec(statements[i]); err != nil {
 			log.Logger.Warnf("execute [%s] failed: %v", statements[i], err)
+			lastErr = err
 			continue
 		}
 	}
 
-	return nil
+	return lastErr
 }
 
 func (ck *CkService) ShowCreateTable(tbname, database string) (string, error) {

service/clickhouse/clickhouse_service.go

@@ -37,10 +37,27 @@ func NewZkService(nodes []string, port int, sessionTimeout int) (*ZkService, err
 		SessionTimeout: sessionTimeout,
 	}
 
-	servers := make([]string, len(nodes))
-	for index, node := range nodes {
-		servers[index] = net.JoinHostPort(node, fmt.Sprint(port))
+	var leader string
+	for _, node := range nodes {
+		tmp := model.ZkStatusRsp{
+			Host: node,
+		}
+		body, err := ZkMetric(node, port, "mntr")
+		if err != nil {
+			continue
+		}
+		err = json.Unmarshal(body, &tmp)
+		if err == nil {
+			if tmp.ServerState == "leader" {
+				leader = node
+				break
+			}
+		}
+	}
+	if leader == "" {
+		return nil, errors.New("no leader zookeeper node")
 	}
+	servers := []string{net.JoinHostPort(leader, fmt.Sprint(port))}
 
 	c, e, err := zk.Connect(servers, time.Duration(sessionTimeout)*time.Second)
 	if err != nil {