Python 检测系统时间,k8s版本,redis集群,etcd,mysql,ceph,kafka

• 2020 年 2 月 24 日
• 笔记

一、概述

线上有一套k8s集群，部署了很多应用。现在需要对一些基础服务做一些常规检测，比如：

• 系统时间，要求：k8s的每一个节点的时间，差值上下不超过2秒
• k8s版本，要求：k8s的每一个节点的版本必须一致
• redis集群，要求：1. 查看cluster nodes状态 2. AOF状态
• etcd，要求：访问etcd的api，能获取到版本信息， 说明etcd服务正常
• mysql，要求：获取mysql运行统计时间，能获取说明mysql服务正常
• ceph，要求：使用ceph osd tree命令查看ceph节点信息
• kafka，要求：使用生产者模式写入一个消息，消费者模式能得到同样的消息，则kafka服务正常

python远程执行命令

上面这么多要求，有一大部分，都需要远程执行命令。那么如何使用python来执行远程命令呢？

使用paramiko模块即可！

paramiko

安装paramiko模块

pip3 install paramiko

使用paramiko

这里，我封装了一个函数ssh2，代码如下：

import paramiko  def ssh2(ip, username, passwd, cmd):      """      使用ssh连接远程服务器执行命令      :param username: 用户名      :param passwd: 密码      :param cmd: 执行的命令      :return:      """      try:          ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例          # 设置host key,如果在"known_hosts"中没有保存相关的信息，SSHClient 默认行为是拒绝连接，会提示yes/no          ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())          ssh.connect(ip, 22, username, passwd, timeout=1)  # 连接远程服务器,超时时间1秒          stdin, stdout, stderr = ssh.exec_command(cmd)  # 执行命令          out = stdout.readlines()    # 执行结果,readlines会返回列表          ssh.close()  # 关闭ssh连接          return out      except Exception as e:          print(e)          return False

执行此函数，会返回一个列表。因为是使用readlines，将结果转换为列表了！

如果执行时，输出：

Please use EllipticCurvePublicKey.public_bytes to obtain both compressed and uncompressed point encoding.

原因

paramiko 2.4.2 依赖 cryptography，而最新的cryptography==2.5里有一些弃用的API。

解决

删掉cryptography，安装2.4.2，就不会报错了。

pip uninstall cryptography  pip install cryptography==2.4.2

本文参考链接：

https://blog.51cto.com/wangfeng7399/2376115

二、系统时间

获取时间戳

我们需要获取多台服务器的时间，并且还需要对比时间差。那么最简单办法，就是获取时间戳，它是一段数字，那么数字之间，就可以做减法了！

使用 date +%s 命令，就可以获取时间戳了

root@localhost:~# date +%s  1547546824

对比思路

怎么去对比，每一天服务器的时间戳呢？上面已经获取到时间戳了，关键问题是，如何对比？

构造字典

这里需要构造一个数据字典，将每一台服务器的ip以及时间戳存储一下，数据格式如下：

{    'ip地址': 时间戳,      ...  }

对比数据

首先从字典里面取出第一个值，由于python 3.5是无需的，所以取出的数据，每次可能不一样。不过没有关系，取出之后，再删除即可！

然后将取出的第一个值，和字典中的其他值，做对比即可！注意：时间戳要转换为int类型才行！

完整代码

ntp.py

#!/usr/bin/env python3  # coding: utf-8  import json  import paramiko  def ssh2(ip, username, passwd, cmd):      """      使用ssh连接远程服务器执行命令      :param username: 用户名      :param passwd: 密码      :param cmd: 执行的命令      :return:      """      try:          ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例          # 设置host key,如果在"known_hosts"中没有保存相关的信息，SSHClient 默认行为是拒绝连接，会提示yes/no          ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())          ssh.connect(ip, 22, username, passwd, timeout=1)  # 连接远程服务器,超时时间1秒          stdin, stdout, stderr = ssh.exec_command(cmd)  # 执行命令          out = stdout.readlines()    # 执行结果,readlines会返回列表          ssh.close()  # 关闭ssh连接          return out      except Exception as e:          print(e)          return False    # 服务器列表  ip_list = ["192.168.0.172","192.168.0.173"]  username = "root"  passwd = "root"  cmd = "date +%s"  # 获取时间戳命令    result_dic = {}  # 执行结果    for ip in ip_list:  # 遍历ip列表      if not result_dic.get(ip):          # 添加键值对          res = ssh2(ip,username,passwd,cmd)          if res:              res = json.loads(res[0])              result_dic[ip] = res              print("添加键值,ip: ", ip,"value: ",res)          else:              print("ssh连接服务器失败，ip: %s" %ip)    print("结果字典",result_dic)    # 获取第一个ip以及时间戳  first_ip = ""  first_val = ""    for i in result_dic:  # 遍历结果字典      # 获取第一个ip以及时间戳      first_ip = i      res = ssh2(i,username,passwd,cmd)      first_val = json.loads(res[0])      print("第一个ip: ", i,"value: ",first_val)        result_dic.pop(i)  # 删除key,避免下面的for循环重复      break    for ip in result_dic:      d_value = int(first_val) - int(result_dic[ip])      # 判断差值上下不超过2      if d_value <= -2 or d_value >= 2:          print("错误, 上下差值超过2 !!!","ip:",ip,"value:",d_value)      else:          print("正常",ip,"差值为: ",d_value)

执行脚本，输出如下：

添加键值,ip:  192.168.0.172 value:  1547540144  添加键值,ip:  192.168.0.173 value:  1547540145  ...  第一个ip:  192.168.0.172 value:  1547540144  正常 192.168.0.173 差值为:  1

三、k8s版本

获取版本

查看k8s版本，使用命令 kubectl version 

root@localhost:~# kubectl version  Client Version: version.Info{Major:"1", Minor:"11", GitVersion:"v1.11.2", GitCommit:"bb9ffb1654d4a729bb4cec18ff088eacc153c239", GitTreeState:"clean", BuildDate:"2018-08-07T23:17:28Z", GoVersion:"go1.10.3", Compiler:"gc", Platform:"linux/amd64"}

那么要获取到 "v1.11.2" ，还需要进一步过滤

root@localhost:~# kubectl version | awk '{print $5}' | cut -d ':' -f 2 | head -1 | cut -d ',' -f 1  "v1.11.2"

对比版本

使用上面的ssh2函数之后，输出的值，是这样的

["v1.11.2n"]

这里会有一个换行符，为了避免这种问题，使用 json.loads() 反序列一下，就可以还原为 v1.11.2，连双引号也没有了！

完整代码

kube_v.py

#!/usr/bin/env python3  # coding: utf-8  # 验证k8s版本    import json  import paramiko  def ssh2(ip, username, passwd, cmd):      """      使用ssh连接远程服务器执行命令      :param username: 用户名      :param passwd: 密码      :param cmd: 执行的命令      :return:      """      try:          ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例          # 设置host key,如果在"known_hosts"中没有保存相关的信息，SSHClient 默认行为是拒绝连接，会提示yes/no          ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())          ssh.connect(ip, 22, username, passwd, timeout=1)  # 连接远程服务器,超时时间1秒          stdin, stdout, stderr = ssh.exec_command(cmd)  # 执行命令          out = stdout.readlines()    # 执行结果,readlines会返回列表          ssh.close()  # 关闭ssh连接          return out      except Exception as e:          print(e)          return False    # 服务器列表  ip_list = ["192.168.0.172","192.168.0.173"]  username = "root"  passwd = "root"    cmd = "kubectl version | awk '{print $5}' | cut -d ':' -f 2 | head -1 | cut -d ',' -f 1"  # 获取时间戳命令  result_dic = {}  # 执行结果    for ip in ip_list:  # 遍历ip列表      if not result_dic.get(ip):          # 添加键值对          res = ssh2(ip,username,passwd,cmd)          # print("res",res)          if res:  # 判断不为None              res = json.loads(res[0])  # 反序列化第一行结果            result_dic[ip] = res          print("添加键值", ip,"value",res)    print("结果字典",result_dic)  # 获取第一个ip以及时间戳  first_ip = ""  first_val = ""    for i in result_dic:  # 遍历结果字典      # 获取第一个ip以及时间戳      first_ip = i      first_val = ssh2(i,username,passwd,cmd)        if first_val:          first_val = json.loads(first_val[0])      print("第一个ip", i, "value", first_val)        result_dic.pop(i)  # 删除key，比较下面的for循环重复      break    for ip in result_dic:      d_value = result_dic[ip]      # print("其他服务器,ip",ip,"value",d_value)      # 判断版本是否一致      if first_val == d_value:          print("正常", ip, "版本是", d_value)      else:          print("错误, 版本不一致!!!", "ip:", ip, "value:", d_value)

执行输出：

添加键值 192.168.0.172 value v1.11.2  添加键值 192.168.0.173 value v1.11.2  ...  第一个ip 192.168.0.172 value v1.11.2  正常 192.168.0.173 版本是 v1.11.2

四、redis集群

cluster nodes

查看cluster nodes信息，使用命令

redis-cli -c -h 192.168.0.172 -p 7201 cluster nodes

请确保服务器，已经安装了redis，可以执行redis-cli命令

执行输出：

2a77efc52a1dfec83130b908dbd3809e057956f6 192.168.0.172:7201@17201 slave c036906bf3079fa3da15ef43df545b17c3efa144 0 1547548520000 19 connected  db60724fdae507eda5d72fd1181559bc6da33097 192.168.0.169:7201@17201 slave ed5cfdfd021dcc821fc0749d318e8ac420cc5c14 0 1547548519607 18 connected  ed5cfdfd021dcc821fc0749d318e8ac420cc5c14 192.168.0.143:7201@17201 myself,master - 0 1547548519000 18 connected 0-5460  d4f928c75a2a17d9c30ec77115628eb1840f6ee4 192.168.0.171:7201@17201 slave b8c692d0eff3a5d1ee059878a4213844c6d82ddf 0 1547548520610 21 connected  b8c692d0eff3a5d1ee059878a4213844c6d82ddf 192.168.0.170:7201@17201 master - 0 1547548521614 21 connected 10923-16383  c036906bf3079fa3da15ef43df545b17c3efa144 192.168.0.168:7201@17201 master - 0 1547548518604 19 connected 5461-10922

从上面的输出信息，可以看出，有3个master节点，3个slave节点。其中紫色部分，如果id一致，表示这是一组服务器！

请确保这一组服务器不能同时挂掉，否则会造成数据丢失！

默认redis集群要求至少6个节点，当redis集群中的master节点，挂掉一半时，集群不可用。

也就是说，目前有3个master节点，最多允许1台mater节点挂掉！

判断依据

还有一点，当有任意一个节点状态为fail时，也表示集群不正常！说明：mater节点已经挂掉了一半！

因此，使用python来判断集群是否状态的关键点，就是判断状态中是否有fail即可！

aof状态

在redis.conf配置文件中， 有一个参数 appendonly ，表示是否开启aof，默认是关闭的。

那么使用redis-cli可以获取它的状态

redis-cli -c -h 192.168.0.172 -p 7201 config get appendonly

执行输出：

1) "appendonly"  2) "no"

上面这段输出，表示 没有开启aof

完整代码

redis.py

#!/usr/bin/env python3  # coding: utf-8  # 检测redis集群    import paramiko    def ssh2(ip, username, passwd, cmd):      """      使用ssh连接远程服务器执行命令      :param username: 用户名      :param passwd: 密码      :param cmd: 执行的命令      :return:      """      try:          ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例          # 设置host key,如果在"known_hosts"中没有保存相关的信息，SSHClient 默认行为是拒绝连接，会提示yes/no          ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())          ssh.connect(ip, 22, username, passwd, timeout=1)  # 连接远程服务器,超时时间1秒          stdin, stdout, stderr = ssh.exec_command(cmd)  # 执行命令          out = stdout.readlines()    # 执行结果,readlines会返回列表          ssh.close()  # 关闭ssh连接          return out      except Exception as e:          print(e)          return False    # 服务器列表  redis_apps = ["192.168.0.168","192.168.0.172"]    username = "root"  passwd = "root"  port = "6379"  cmd1 = "cluster nodes"  # cluster nodes状态  cmd2 = "config get appendonly"  # AOF状态    result_dic = {}  # 执行结果    for ip in redis_apps:  # 遍历ip列表      if not result_dic.get(ip):          result_dic[ip] = {}          # nodes          if not result_dic[ip].get('nodes'):              # 添加键值对，统一由192.168.0.167这台服务器执行redis-cli，因为它安装了redis-cli              res = ssh2("192.168.0.167",username,passwd,"redis-cli -c -h {} -p {} {}".format(ip,port,cmd1))              result_dic[ip]['nodes'] = res              print("添加键值,ip: ", ip,"value: ",res)            # aof          if not result_dic[ip].get('aof'):              # 添加键值对              res = ssh2("192.168.0.167",username,passwd,"redis-cli -c -h {} -p {} {}".format(ip,port,cmd2))              # print(res,type(res))              res = res[1].split("n")[0]  # 获取选项值              result_dic[ip]['aof'] = res              print("添加键值,ip: ", ip,"value: ",res)    print("结果字典",result_dic)      # 标志位  flag_nodes = True  for i in result_dic:      for j in result_dic[i]['nodes']:  # 遍历结果列表          if "fail" in j:  # 判断是否有fail状态              print("状态异常: ",j)              flag_nodes = False    # 判断标志位  if flag_nodes:      print("redis cluster nodes 状态正常")  else:      print("redis cluster nodes 状态异常")    # 输出aof状态  print("redis aof状态如下: ")  for i in result_dic:      print("ip: {} 状态: {}".format(i,result_dic[i]['aof']))

执行程序，输出：

添加键值,ip:  192.168.0.168 value: [...]  ...  redis cluster nodes 状态正常  redis aof状态如下:   ip: 192.168.0.168 状态: no  ip: 192.168.0.172 状态: no

五、etcd

etcd api

判断etcd工作是否正常，只需要能访问到api地址，就说明正常，url如下：

http://192.168.0.169:2380/version

完整代码

etcd.py

#!/usr/bin/env python3  # coding: utf-8  # 检测etcd状态    import json  import requests      etcd_list = ["192.168.0.169","192.168.0.168","192.168.0.167"]    for ip in etcd_list:      # 访问api接口，查看版本      response=requests.get('http://%s:2380/version' %ip)      # print(response.content)      res = (response.content).decode('utf-8')      res_dict = json.loads(res)      # print(res_dict,type(res_dict))      print("ip: {} etcd版本为: {}".format(ip,res_dict['etcdserver']))

执行输出：

ip: 192.168.0.169 etcd版本为: 3.3.0  ip: 192.168.0.168 etcd版本为: 3.3.0  ip: 192.168.0.167 etcd版本为: 3.3.0

六、mysql

运行统计时间

查看mysql的运行统计时间，使用命令

show status like "uptime"

完整代码

#!/usr/bin/env python3  # coding: utf-8    import pymysql    conn = pymysql.connect(      host="192.168.0.179",  # mysql ip地址      user="root",      passwd="root",      port=3306  # mysql 端口号,注意：必须是int类型  )    cur = conn.cursor()  # 创建游标    # 查询当前MySQL本次启动后的运行统计时间  cur.execute('show status like "uptime"')    data_all = cur.fetchall()  # 获取执行的返回结果  print(data_all)

执行输出：

(('Uptime', '941067'),)

七、ceph

节点信息

查看节点信息，需要在 主节点操作

ceph osd tree

完整代码

#!/usr/bin/env python3  # coding: utf-8    import paramiko    def ssh2(ip, username, passwd, cmd):      """      使用ssh连接远程服务器执行命令      :param username: 用户名      :param passwd: 密码      :param cmd: 执行的命令      :return:      """      try:          ssh = paramiko.SSHClient()  # 创建一个新的SSHClient实例          # 设置host key,如果在"known_hosts"中没有保存相关的信息，SSHClient 默认行为是拒绝连接，会提示yes/no          ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())          ssh.connect(ip, 22, username, passwd, timeout=1)  # 连接远程服务器,超时时间1秒          stdin, stdout, stderr = ssh.exec_command(cmd)  # 执行命令          out = stdout.readlines()    # 执行结果,readlines会返回列表          ssh.close()  # 关闭ssh连接          return out      except Exception as e:          print(e)          return False    svr_list = ["192.168.0.181"]  username = "root"  passwd = "root"  cmd = "ceph osd tree"  # 查看节点信息    for ip in svr_list:      res = ssh2(ip,username,passwd,cmd)      print(ip,"节点信息如下：")      print("".join(res))

执行输出：

192.168.0.181 节点信息如下：  ID WEIGHT  TYPE NAME           UP/DOWN REWEIGHT PRIMARY-AFFINITY   -1 1.02695 root default                                            -2 0.18399     host xx-node13                                     0.18399         osd.0            up  1.00000          1.00000   -3 0.18399     host xx-node15                                     0.18399         osd.1            up  1.00000          1.00000   -4 0.18399     host xx-node17                                     0.18399         osd.2            up  1.00000          1.00000   -5 0.18399     host xx-node18                                     0.18399         osd.3            up  1.00000          1.00000   -6 0.10699     host xx-node19                                     0.10699         osd.4            up  1.00000          1.00000   -7 0.18399     host xx-node14                                     0.18399         osd.5            up  1.00000          1.00000

八、kafka

创建topic

手动创建名为test的topic

bin/kafka-topics.sh --create --zookeeper zookeeper-1.default.svc.cluster.local:2181,zookeeper-2.default.svc.cluster.local:2128,zookeeper-3.default.svc.cluster.local:2128 --topic test --partitions 1 --replication-factor 1

因为python中的kakfa包无法直接创建 topic，所以需要手动创建

完整代码

由于线上kafka启动了ACL，那么请确保相关用户设置了ACL规则，关于acl的配置，请参考链接：

https://www.cnblogs.com/xiao987334176/p/10110389.html

为了方便，这里直接使用超级用户。注意：超级用户是不需要设置ACL规则的，拥有所有权限

#!/usr/bin/env python3  # coding: utf-8  # 注意：需要手动创建topic才行执行此脚本    import time  from kafka import KafkaProducer  from kafka import KafkaConsumer      class KafkaClient(object):  # kafka客户端程序      def __init__(self, kafka_server, port, topic,content,username,password):          self.kafka_server = kafka_server  # kafka服务器ip地址          self.port = port  # kafka端口          self.topic = topic  # topic名          self.content = content  # 发送内容          self.username = username          self.password = password        def producer(self):          """          生产者模式          :return: object          """            # 连接kafka服务器,比如['192.138.150.193:9092']          producer = KafkaProducer(bootstrap_servers=['%s:%s' % (self.kafka_server, self.port)],                                   security_protocol="SASL_PLAINTEXT",  # 指定SASL安全协议                                   sasl_mechanism='PLAIN',  # 配置SASL机制                                   sasl_plain_username=self.username,  # 认证用户名                                   sasl_plain_password=self.password,  # 密码                                   )            producer.send(self.topic, self.content)  # 发送消息,必须是二进制          producer.flush()  # flush确保所有meg都传送给broker          producer.close()          return producer        def consumer(self):          """          消费者模式          :return: object          """            # 连接kafka,指定组为test_group          consumer = KafkaConsumer(topic, group_id='test_group', bootstrap_servers=['%s:%s' % (kafka_server, port)],                                   sasl_mechanism="PLAIN",                                   security_protocol='SASL_PLAINTEXT',                                   sasl_plain_username=self.username,                                   sasl_plain_password=self.password,                                   )            return consumer          # for msg in consumer:          #     recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)          #     print(recv)        def main(self):          startime = time.time()  # 开始时间            client = KafkaClient(self.kafka_server, self.port, self.topic, self.content,self.username,self.password)  # 实例化客户端          client.producer()  # 执行生产者          print('执行生产者')          consumer = client.consumer()  # 执行消费者          print('执行消费者')          print('等待结果....')          flag = False            for msg in consumer:              # recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)              # 判断生产的消息和消费的消息是否一致              print(msg.value)              # print(self.content)              if msg.value == self.content:                  flag = True                  break            consumer.close()  # 关闭消费者对象          endtime = time.time()  # 结束时间            if flag:              # %.2f %(xx) 表示保留小数点2位              return "kafka验证消息成功,花费时间", '%.2f 秒' % (endtime - startime)          else:              return "kafka验证消息失败,花费时间", '%.2f 秒' % (endtime - startime)      if __name__ == '__main__':      kafka_server = "kafka-1.default.svc.cluster.local"      port = "9092"      topic = "test"      # 测试消息      content = "hello honey".encode('utf-8')        username = "admin"      password = "admin"        client = KafkaClient(kafka_server,port,topic,content,username,password)  # 实例化客户端      print(client.main())

执行程序，输出：

执行生产者  执行消费者  等待结果....  b'hello honey'('kafka验证消息成功,花费时间', '3.61 秒')

注意：第一次执行时，会卡住1分钟。多执行几次就会很快了，至于什么原因，不知道！

就是在执行这一行代码时，会卡住

for msg in consumer: