fix: fix zookeeper collect blocking (#965)

Author: tanxiao1990

inputs/zookeeper/zookeeper.go

@@ -110,6 +110,11 @@ func (ins *Instance) Gather(slist *types.SampleList) {
 
 func (ins *Instance) gatherOneHost(wg *sync.WaitGroup, slist *types.SampleList, zkHost string) {
 	defer wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			log.Println("E! Recovered in zookeeper gatherOneHost ", zkHost, r)
+		}
+	}()
 
 	tags := map[string]string{"zk_host": zkHost, "zk_cluster": ins.ClusterName}
 	begun := time.Now()
@@ -120,27 +125,22 @@ func (ins *Instance) gatherOneHost(wg *sync.WaitGroup, slist *types.SampleList,
 		slist.PushFront(types.NewSample("", "zk_scrape_use_seconds", use, tags))
 	}(begun)
 
-	// zk_up
-	mntrConn, err := ins.ZkConnect(zkHost)
-	if err != nil {
-		slist.PushFront(types.NewSample("", "zk_up", 0, tags))
-		log.Println("E! failed to connect zookeeper:", zkHost, "error:", err)
-		return
-	}
-
-	defer mntrConn.Close()
-	ins.gatherMntrResult(mntrConn, slist, tags)
+	// zk_up  zk_ruok
+	conn, err := ins.ZkConnect(zkHost)
 
-	// zk_ruok
-	ruokConn, err := ins.ZkConnect(zkHost)
 	if err != nil {
+		slist.PushFront(types.NewSample("", "zk_up", 0, tags))
 		slist.PushFront(types.NewSample("", "zk_ruok", 0, tags))
 		log.Println("E! failed to connect zookeeper:", zkHost, "error:", err)
 		return
 	}
+	defer conn.Close()
+
+	// prevent blocking
+	conn.SetDeadline(time.Now().Add(time.Duration(ins.Timeout) * time.Second))
 
-	defer ruokConn.Close()
-	ins.gatherRuokResult(ruokConn, slist, tags)
+	ins.gatherMntrResult(conn, slist, tags)
+	ins.gatherRuokResult(conn, slist, tags)
 }
 
 func (ins *Instance) gatherMntrResult(conn net.Conn, slist *types.SampleList, globalTags map[string]string) {
@@ -226,12 +226,14 @@ func (ins *Instance) gatherRuokResult(conn net.Conn, slist *types.SampleList, gl
 func sendZookeeperCmd(conn net.Conn, cmd string) string {
 	_, err := conn.Write([]byte(cmd))
 	if err != nil {
-		log.Println("E! failed to exec Zookeeper command:", cmd)
+		log.Printf("E! failed to exec Zookeeper command: %s response from '%s': %s", cmd, conn.RemoteAddr().String(), err)
+		return ""
 	}
 
 	res, err := ioutil.ReadAll(conn)
 	if err != nil {
 		log.Printf("E! failed read Zookeeper command: '%s' response from '%s': %s", cmd, conn.RemoteAddr().String(), err)
+		return ""
 	}
 	return string(res)
 }