Zookeeper-监控（Monitor）

运维小路

发布于 2025-05-18 01:43:51

7200

代码可运行

文章被收录于专栏：运维小路运维小路

运行总次数：0

代码可运行

作者介绍：简历上没有一个精通的运维工程师。请点击上方的蓝色《运维小路》关注我，下面的思维导图也是预计更新的内容和当前进度(不定时更新)。

前面我们介绍介绍了几个常用的代理服务器，本章节我们讲来讲解Zookeeper这个中间件。

我们虽然前面讲过4字命令和自带的Shell脚本可以读取ZooKeeper的运行状态，但是它并不成体系。还有就是目前云原生的情况下，我们的的监控也需要接入Prometheus和Grafana，所以我们下面就来讲解下我们的监控。目前官方可以直接下载的ZooKeeper版本是3.7.2 和 3.8.4 和3.9.3从配置文件来看都是支持通过metrics方式暴露监控指标。

#取消所有注释，重启服务则可以看到
## Metrics Providers
#
# https://prometheus.io Metrics Exporter
metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
metricsProvider.httpHost=0.0.0.0
metricsProvider.httpPort=7000
metricsProvider.exportJvmInfo=true

#访问地址
http://ip地址:7000/metrics

这里的监控默认包含两个方面的内容：一个就是ZooKeeper的指标（类似前面的四字命令输出）。另外一个它作为使用JAVA语言的应用，内部都有一个JAVA虚拟机，如果我们要分析性能问题就必须分析JAVA虚拟机的堆栈信息(所有的JAVA应用都有)。针对JAVA虚拟机都有一个jvm监控，如果要关闭必须明确关闭设置。

#必须显示设置关闭，注释都不行 
metricsProvider.exportJvmInfo=false

#如果开启了可以从日志里面看出来 
2025-04-28 00:36:23,701 [myid:] - INFO  [main:PrometheusMetricsProvider@74] - Initializing metrics, configuration: {httpPort=7000}
2025-04-28 00:36:23,701 [myid:] - INFO  [main:PrometheusMetricsProvider@82] - Starting /metrics HTTP endpoint at port 7000 exportJvmInfo: true

下面这个图就是我从Grafana官方网站找出来的通过metrics的指标在grafana上的图。

#下载地址
https://grafana.com/grafana/dashboards/12338-zookeeper/

当然如果只是简单监控，未接入云原生则可以采用下面的脚本，这里不仅仅需要监控集群状态，还需要监控znode和zxid，因为我是遇到过集群状态正常，但是zxid不一致的情况。

当然下面的代码是我让DeepSeek生成的，格式还有功能可以根据自己需要调整。并且还可以进行调整作为检查本机的ZooKeeper是否正常的的脚本，然后如果不正常还可以对它进行重启。

import socket
from concurrent.futures import ThreadPoolExecutor
import concurrent.futures

def parse_zoo_cfg(file_path):
    """解析zoo.cfg文件，获取服务器列表和客户端端口"""
    client_port = 2181  # 默认端口
    servers = []
    with open(file_path, 'r') as f:
        for line in f:
            line = line.strip()
            if line.startswith('#') or not line:
                continue
            if '=' in line:
                key, value = line.split('=', 1)
                key = key.strip()
                value = value.strip()
                if key == 'clientPort':
                    try:
                        client_port = int(value)
                    except ValueError:
                        pass  # 保持默认端口
                elif key.startswith('server.'):
                    parts = value.split(':')
                    if parts:
                        host = parts[0]
                        servers.append(host)
    return servers, client_port

def get_node_stat(host, port, timeout=5):
    """通过四字命令获取节点状态"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)
            s.connect((host, port))
            s.sendall(b'stat\n')
            response = b''
            while True:
                data = s.recv(1024)
                if not data:
                    break
                response += data
            return response.decode('utf-8')
    except Exception as e:
        return f"Error: {str(e)}"

def parse_stat_response(response):
    """解析stat命令的响应"""
    stats = {}
    lines = response.split('\n')
    for line in lines:
        if line.startswith('Mode:'):
            stats['mode'] = line.split(':', 1)[1].strip()
        elif line.startswith('Zxid:'):
            stats['zxid'] = line.split(':', 1)[1].strip()
        elif line.startswith('Node count:'):
            stats['node_count'] = line.split(':', 1)[1].strip()
    return stats

def main():
    import sys
    cfg_file = sys.argv[1] if len(sys.argv) > 1 else 'zoo.cfg'
    servers, client_port = parse_zoo_cfg(cfg_file)

    if not servers:
        print("未找到服务器配置。")
        return

    print(f"检查 {len(servers)} 个节点...\n")

    results = []
    with ThreadPoolExecutor(max_workers=3) as executor:
        future_to_host = {executor.submit(get_node_stat, host, client_port): host for host in servers}
        for future in concurrent.futures.as_completed(future_to_host):
            host = future_to_host[future]
            try:
                response = future.result()
                if response.startswith("Error"):
                    stats = {'error': response}
                else:
                    stats = parse_stat_response(response)
                results.append((host, stats))
            except Exception as e:
                stats = {'error': str(e)}
                results.append((host, stats))

    # 打印结果
    for host, stats in results:
        print(f"节点: {host}:{client_port}")
        if 'error' in stats:
            print(f"  错误: {stats['error']}")
        else:
            print(f"  状态: {stats.get('mode', 'N/A')}")
            print(f"  Znode数量: {stats.get('node_count', 'N/A')}")
            print(f"  ZXID: {stats.get('zxid', 'N/A')}")
        print()

if __name__ == "__main__":
    main()

当前集群检查就是不符合预期的。监控的目的是让我们及时发现异常，然后根据异常情况及时做除修复操作，避免更大的故障。

检查 3 个节点...

节点: 192.168.31.140:2181
  状态: follower
  Znode数量: 9
  ZXID: 0x2000000d1

节点: 192.168.31.141:2181
  状态: leader
  Znode数量: 9
  ZXID: 0x500000000

节点: 192.168.31.142:2181
  状态: follower
  Znode数量: 9
  ZXID: 0x500000000