微服务状态之python巡查脚本开发

背景

由于后端微服务架构，于是各种业务被拆分为多个服务，服务之间的调用采用RPC接口，而Nacos作为注册中心，可以监听多个服务的状态，比如某个服务是否down掉了、某个服务的访问地址是否改变、以及流量分配降级等等。（nacos注册中心的由来及原理）

运维架构用的是K8S，由于之前没有做服务高可用的特性，K8S上服务挂掉后不会自动重启。

所以测试过程中，某个接口有问题，可能最先需要排查该服务或依赖服务是否正常可用。

问题

上面说到，测试过程中还需要查看服务是否正常，这是很浪费时间的，特别是环境不稳定的情况下，也许今天下班前测得好好的，明天或者后天就会发现部分服务就挂掉了（毕竟六七十个服务人工筛查太麻烦了）。能否把这个问题前置解决？

解决方案

我的解决方法就是，利用python爬虫思路，每天获取当前nacos注册的服务与nacos配置的服务列表对比，如果注册的服务小于Nacos的服务，就说明部分服务不可用，这时就可以先解决后再介入测试，再搭配Jenkins，可以早晚巡查服务是否正常。（当然我们后面K8S已经实现了服务的自动重启，所以该脚本也算服务检查的补充方案了）

设计如下

首先定义一个测试类checkJob。

第一步，登录nacos

1     # 登录nocas

2     def con_nacos(self):

3         data = {"username":"nacos","password":"Billion123!@#"}

4         res = requests.post(url = self.urlLogin, headers = self.headers, data=data).json()['accessToken']

5         return res

第二步，获取当前namespace服务配置文件

1     # 获取目前环境服务配置列表

2     def get_job_config(self):

3         self.joblist = []

4         params = self.config_params

5         res = requests.get(url= self.urlGetConfig, headers = self.headers, params = params).json()["pageItems"]

6         for i in res:

7             self.joblist.append(i.get("dataId"))

8         print("-------目前的服务配置有：", len(self.joblist))

9         return self.joblist

第三步，获取当前正常注册的服务

1     # 获取目前所有服务列表

2     def get_job(self):

3         self.getjoblist = []

4         res = requests.get(url = self.urlGetJob,headers = self.headers, params = self.params).json()['serviceList']

5         for i in res:

6             self.getjoblist.append(i.get("name"))

7         return self.getjoblist

第四步，比对实际注册的服务与配置的服务并打印输出

 1     #迭代比对当前服务列表

 2     def check_job(self):

 3         tmp = set(self.joblist).difference(self.getjoblist)

 4         if tmp:

 5             return tmp

 6         else:

 7             return ""

 8

 9     #对未存在的服务通知告警

10     def msg(self,tmp):

11         for i in self.ignore:

12             if i in tmp:

13                 tmp.remove(i)

14                 # print(i)

15         if tmp:

16             print(f"缺少{len(tmp)}个服务，如下：", tmp)

17         else:

18             print(f"{len(self.joblist)}个服务正常：", self.joblist)

当然可能有多个环境，优化一下代码，全部代码如下：

  1 import requests

  2

  3 class CheckJob():

  4     #定义所有服务列表

  5     joblist = []

  6     urlLogin = "http://xxx.xxx.xxx.xxx:10086/nacos/v1/auth/users/login"

  7     urlGetConfig = "http://xxx.xxx.xxx.xxx:10086/nacos/v1/cs/configs"

  8     urlGetJob = "http://xxx.xxx.xxx.xxx:10086/nacos/v1/ns/catalog/services"

  9     headers = {

 10         'Accept': 'application/json, text/plain, */*',

 11         'Accept-Encoding': 'gzip, deflate',

 12         'Accept-Language': 'zh-CN,zh;q=0.9',

 13         'Connection': 'keep-alive',

 14         "Host": "nacos.bb.local",

 15         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36',

 16     }

 17     # 忽略部分服务列表

 18     ignore = ["member", "zuul", "order", "b-shop"]

 19     def __init__(self):

 20         self.accessToken = self.con_nacos()

 21         for str in ["d","t","p","pro"]:

 22             if str.strip() == "p":

 23                 # 预发环境

 24                 self.params = "hasIpCount=true&withInstances=false&pageNo=1&pageSize=100&serviceNameParam=&groupNameParam=&namespaceId=4dcd1ab3-9c07-4f2b-9ed9-9c1f67f4de9e&accessToken=$accessToken"

 25                 self.config_params = "dataId=&group=&appName=&config_tags=&pageNo=1&pageSize=100&tenant=4dcd1ab3-9c07-4f2b-9ed9-9c1f67f4de9e&search=accurate&accessToken=$accessToken"

 26                 print("预发环境",end="")

 27             elif str.strip() == "t":

 28                 # 测试环境

 29                 self.params = "hasIpCount=true&withInstances=false&pageNo=1&pageSize=100&serviceNameParam=&groupNameParam=&namespaceId=0f4106a4-acad-4327-a1f5-54ee78b342ea&accessToken=$accessToken"

 30                 self.config_params = "dataId=&group=&appName=&config_tags=&pageNo=1&pageSize=100&tenant=0f4106a4-acad-4327-a1f5-54ee78b342ea&search=accurate&accessToken=&accessToken"

 31                 print("测试环境", end="")

 32             elif str.strip() == "d":

 33                 # 开发环境

 34                 self.params = "hasIpCount=true&withInstances=false&pageNo=1&pageSize=100&serviceNameParam=&groupNameParam=&accessToken=$accessToken&namespaceId=ddd4b051-5a76-417d-bef8-09f7ff941bbd"

 35                 self.config_params = "dataId=&group=&appName=&config_tags=&pageNo=1&pageSize=100&tenant=ddd4b051-5a76-417d-bef8-09f7ff941bbd&search=accurate&accessToken=&accessToken"

 36                 print("开发环境",end="")

 37             elif str.strip() == "pro":

 38                 # 线上环境

 39                 self.params = "hasIpCount=true&withInstances=false&pageNo=1&pageSize=100&serviceNameParam=&groupNameParam=&accessToken=&accessToken&namespaceId=1b1a6ba8-2d21-4cd6-b7b2-ecd3b838c387"

 40                 self.config_params = "dataId=&group=&appName=&config_tags=&pageNo=1&pageSize=100&tenant=1b1a6ba8-2d21-4cd6-b7b2-ecd3b838c387&search=accurate&accessToken=&accessToken"

 41                 print("线上",end="")

 42             else:

 43                 print("输入不合法")

 44             if self.params:

 45                 self.params = self.params.replace("$accessToken", self.accessToken)

 46                 if self.get_job_config():

 47                     if self.get_job():

 48                         self.msg(self.check_job())

 49                     else:

 50                         print("获取nacos服务失败")

 51                 else:

 52                     print("获取nacos服务配置失败")

 53             else:

 54                 print("nacos登录失败")

 55

 56

 57     # 登录nocas

 58     def con_nacos(self):

 59         data = {"username":"naco登录名","password":"nacos登录密码"}

 60         res = requests.post(url = self.urlLogin, headers = self.headers, data=data).json()['accessToken']

 61         return res

 62

 63     # 获取目前环境服务配置列表

 64     def get_job_config(self):

 65         self.joblist = []

 66         params = self.config_params

 67         res = requests.get(url= self.urlGetConfig, headers = self.headers, params = params).json()["pageItems"]

 68         for i in res:

 69             self.joblist.append(i.get("dataId"))

 70         print("-------目前的服务配置有：", len(self.joblist))

 71         return self.joblist

 72

 73     # 获取目前所有服务列表

 74     def get_job(self):

 75         self.getjoblist = []

 76         res = requests.get(url = self.urlGetJob,headers = self.headers, params = self.params).json()['serviceList']

 77         for i in res:

 78             self.getjoblist.append(i.get("name"))

 79         return self.getjoblist

 80

 81     #迭代比对当前服务列表

 82     def check_job(self):

 83         tmp = set(self.joblist).difference(self.getjoblist)

 84         if tmp:

 85             return tmp

 86         else:

 87             return ""

 88

 89     #对未存在的服务通知告警

 90     def msg(self,tmp):

 91         for i in self.ignore:

 92             if i in tmp:

 93                 tmp.remove(i)

 94                 # print(i)

 95         if tmp:

 96             print(f"缺少{len(tmp)}个服务，如下：", tmp)

 97         else:

 98             print(f"{len(self.joblist)}个服务正常：", self.joblist)

 99

100 if __name__ == '__main__':

101     cj= CheckJob()

Jenkins持续集成

第一步，上传到gitlab

这一步就不说了，自行上传即可（记得拷贝git链接），没用gitlab的可以直接放到Jenkins的服务器上。

第二步，配置jenkins

新建一个任务，选择github项目，输入项目地址

在源码管理这里配置的git登录用户证书信息，分支直接选master即可

构建触发器这里填入cron定时任务，这里代表周一到周六每天早上8点和下午15点各执行一次。

构建命令这里输入python命令即可（注意Jenkins的服务器上得有python的运行环境）

最后，构建后自动发送邮件（需要安装Editable Email Notification插件）

以下收件人和回复人得邮件地址采用了变量方式，也可直接输入邮箱地址

Content Type选择HTML（text/html）作为邮箱格式，Default Subject输入模板标题，Default Content作为邮件内容模板，参考如下

 1 <!DOCTYPE html>

 2 <html>

 3 <head>

 4 <meta charset="UTF-8">

 5 <title>${ENV, var="JOB_NAME"}-第${BUILD_NUMBER}次构建日志</title>

 6 </head>

 7

 8 <body leftmargin="8" marginwidth="0" topmargin="8" marginheight="4"

 9     offset="0">

10     <table width="95%" cellpadding="0" cellspacing="0"  style="font-size: 11pt; font-family: Tahoma, Arial, Helvetica, sans-serif">

11         <tr>

12             本邮件由系统自动发出，无需回复！<br/>

13             各位同事，大家好，以下为${PROJECT_NAME }项目构建信息</br>

14             <td><font color="#2D5900">构建结果 - ${BUILD_STATUS}</font></td>

15         </tr>

16         <tr>

17             <td><br />

18             <b><font color="#0B610B">构建信息</font></b>

19             <hr size="2" width="100%" align="center" /></td>

20         </tr>

21         <tr>

22             <td>

23                 <ul>

24

25                     <li>账        号 ： admin</li>

26                     <li>密        码 ： admin123</li>

27                     <li>项目名称 ： ${PROJECT_NAME}</li>

28                     <li>构建编号 ： 第${BUILD_NUMBER}次构建</li>

29                     <li>触发原因： ${CAUSE}</li>

30                     <li>构建状态： ${BUILD_STATUS}</li>

31                     <li>项目  Url ： <a href="${PROJECT_URL}">${PROJECT_URL}</a></li>

32                     <li>构建信息： ${BUILD_LOG, maxLines = 10, escapeHtml = false}</li>

33                 </ul>

34

35 <h4><font color="#0B610B">失败用例</font></h4>

36 <hr size="2" width="100%" />

37 $FAILED_TESTS<br/>

38

39 <h4><font color="#0B610B">最近提交(#$SVN_REVISION)</font></h4>

40 <hr size="2" width="100%" />

41 <ul>

42 ${CHANGES_SINCE_LAST_SUCCESS, reverse=true, format="%c", changesFormat="<li>%d [%a] %m</li>"}

43 </ul>

44             </td>

45         </tr>

46     </table>

47 </body>

48 </html>

Attachments作为邮件附件，直接输入report/*.html，Attach Build Log这栏选择Attach Build Log选项

保存构建项目，查看结果如下，详细配置邮件可参考这个文章：https://zhuanlan.zhihu.com/p/161627231

巴特西