Python 检测系统时间,k8s版本,redis集群,etcd,mysql,ceph,kafka
发布时间：2020-02-19 19:34:18编辑：admin阅读（2690）
系统时间，要求：k8s的每一个节点的时间，差值上下不超过2秒
k8s版本，要求：k8s的每一个节点的版本必须一致
redis集群，要求：1. 查看cluster nodes状态 2. AOF状态
etcd，要求：访问etcd的api，能获取到版本信息， 说明etcd服务正常
mysql，要求：获取mysql运行统计时间，能获取说明mysql服务正常
ceph，要求：使用ceph osd tree命令查看ceph节点信息
kafka，要求：使用生产者模式写入一个消息，消费者模式能得到同样的消息，则kafka服务正常
一、概述
线上有一套k8s集群，部署了很多应用。现在需要对一些基础服务做一些常规检测，比如：
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.2\n"]
这里会有一个换行符，为了避免这种问题，使用 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:
上一篇： Grafana 监控系统是否重启
下一篇： docker 运行Django项目
- openvpn linux客户端使用
47606
- H3C基本命令大全
45987
- openvpn windows客户端使用
36910
- H3C IRF原理及 配置
34470
- Python exit()函数
29082
- openvpn mac客户端使用
25713
- python全系列官方中文文档
24566
- 1.常用turtle功能函数
19715
- python 获取网卡实时流量
19248
- python 获取Linux和Windows硬件信息
17756
- python3-scipy-ndimage平滑
5566°
- python3-scikit-image平滑去噪
6155°
- python3-PIL非线性噪声平滑
5691°
- python3-Scipy-ndimage进行盒核与高斯核平滑比较
5737°
- python3-PIL高斯模糊滤波器平滑
6706°
- python3-PIL基于盒模糊核均值化平滑
5483°
- python3-PIL线性噪声平滑
5485°
- python3-scikit-image直方图匹配
5989°
- python3-scikit-image对比拉伸和直方图均衡化
5965°
- python3-PIL二值化
7295°
- 姓名：Run
- 职业：谜
- 邮箱：383697894@qq.com
- 定位：上海 · 松江
