【 完结福利 】 Python 实现实时文件监控

在我们的业务运维过程中，监控是无处不在的。我们需要对业务的运行状态，数据库的运行状态，Nginx的运行状态等等做监控。一旦有业务故障，或者业务即将发生故障的时候提前通知我们的运维或者开发人员。这样才能把损失和风险降到最低。

多维度详解
手把手入门
《从头解锁Python运维》，专栏完结福利，开启限时拼团>>>

当然要查看业务的运行状态是否正常，我们一般从以下几个方面来判断：

（1）业务接口的状态码是否正常

（2）业务接口的返回内容是否正常

（3）业务端口是否正常

（4）对业务程序的生成的日志内容进行判断

当然上面的4个判断准则，我们一般可以通过如下几个方法进行逐一实现：（当然这只是我们公司的实现思路举例）

（1）业务接口的状态码监控

1.1 我们可以通过Zabbix 的web监测来判断业务的状态码，并编写触发器实现监控告警。
1.2 我们也可以通过Zabbix的自定义Key，然后写脚本去添加监控项以及触发器实现监控告警。
1.3 Prometheus的blackbox_exporter 实现接口的状态码监控。

（2）业务接口的返回内容监控

• 2.1 我们可以通过Zabbix 的web监测来判断业务的状态码，并编写触发器实现监控告警。
• 2.2 我们也可以通过Zabbix的自定义Key，然后写脚本去添加监控项以及触发器实现监控告警。
• 2.3 Prometheus的blackbox_exporter 的fail_if_body_not_matches_regexp 等配置实现监控。

（3）业务端口监控

3.2 创建 zabbix的模板，使用zabbix的net.tcp.listen[port] 实现TCP端口监控。
3.2 我们也可以通过Zabbix的自定义Key，然后写脚本去添加监控项以及触发器实现监控告警。
3.3 Prometheus的blackbox_exporter实现。

（4）对业务程序的生成的日志内容进行判断

• 4.1 通过编写脚本,tail -C 50M xxx.log... 然后结合zabbix实现日志内容监控告警。
• 4.2 我们把所有日志（包括前置机Nginx的日志）都收集ELFK/ELK系统中，通过编写查询ElasticSearch的指定索引（当然这个指定索引是通过参数传入进来的）进行告警监控。并结合Zabbix实现定时查询并告警。
• 编写守护进程对日志内容进行告警，不依赖任何第三方的监控系统。

Python实现实时文件监控几种方案

上面介绍了常用的监控方法的实现。如果使用第三方的监控系统去实现日志内容的告警的话多少有一点延迟性，如果对于非常重要的业务，我们更加需要的是能实时监控日志内容并监测到异常就能告警出来。

本文讲如何使用Python去实现实时文件监控并告警。首先我们来看看下面的一个具体需求（只是一个举例）：

我们有一个支付的Java微服务，进程名为pay-1.0.0.jar, 它是个微服务。 同时它会打2个日志：

pay-api_all.log： 这个日志是info和error都会记录。
pay-api_error.log： 这个日志只会记录error日志。

现在我们需要监控 pay-api_error.log 日志一旦出现wechat 就告警。（也就是一旦微信支付失败立马邮件告警）

对于这个需求，我们来分析一下我们要做的具体步骤：

第一步: 编写好告邮件告警函数或者导入类。

第二步：我们要能实时监测文件，能实时的读到日志文件内容的最近一条的内容，这个监测进程是后台进程。

第三步： 一旦最后一条日志出现出现wechat 立马触发告警的函数。

开始我们的代码

我们先准备邮件发送类S_mail.py 代码如下：

#coding:utf-8 import smtplib from email.mime.text import MIMEText class SendEMail(object): # 定义第三方 SMTP 服务 def __init__(self): self.mail_host = "smtp.exmail.qq.com" # SMTP服务器 self.mail_user = "tech.sys@aa.cn" # 用户名 self.mail_pass = "aapwd" # 密码 self.sender = 'tech.sys@aa.cn' # 发件人邮箱 self.smtpObj = smtplib.SMTP_SSL(self.mail_host, 465) self.smtpObj.login(self.mail_user, self.mail_pass) # 登录验证 def sendmail(self, receivers, title, content): message = MIMEText(content, 'plain', 'utf-8') # 内容, 格式, 编码 message['From'] = "{}".format(self.sender) message['To'] = ",".join(receivers) message['Subject'] = title try: self.smtpObj.sendmail(self.sender, message['To'].split(','), message.as_string()) # 发送 print("mail has been send successfully.") except smtplib.SMTPException as e: print(e) if __name__ == '__main__': sm = SendEMail() sm.sendmail(['xx@qq.com'], '主题', '正文') 

邮件发送类的代码，在之前的章节中已经讲过了，这里不再解释代码。为了测试方便，我们直接把邮件类，代码以及日志放在同一个文件夹下测试。

$ tree /home/www /home/www ├── pay-api_error.log ├── S_mail.py └── v1.py 0 directories, 3 files

下面我们来实现文件监测的逻辑。

方案一之调用Linux的tailf实现

我们知道Linux的tailf命令可以实时获取文件的内容，我们尝试调用Linux的Shell去实现，我们暂且命名为v1.py，代码如下：

import subprocess from S_mail import SendEMail # 定义变量 logfile = "pay-api_error.log" cmd = 'tailf -1 {0}'.format(logfile) key_word="wechat" pp = subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,shell=True) while True: line = pp.stdout.readline().strip() line = line.decode() #编码成字符串 if key_word in line: print("有{0},发送告警".format(key_word)) sm = SendEMail() sm.sendmail(['xx@qq.com'], '主题', '正文')

代码解析：

第1-2行： 导入subprocess 用于调用shell，from S_mail import SendEMail 用于导入邮件类。

第4-7行： 定义变量，包括shell命令，文件路径以及关键字。

第10-17行： 调用shell去执行，strip() 方法用于移除字符串头尾指定的字符（默认为空格或换行符）或字符序列。

line = pp.stdout.readline().strip() 得到是byte类型，我们需要将它decode转成字符串类型。然后再去判断这个字符串有没有关键字然后再触发告警。

我们测试一下：

$ echo "aa" >> pay-api_error.log $ echo "wechat" >> pay-api_error.log $ python3 v1.py 有wechat,发送告警 发送成功

当然我们可以使用nohup丢到后台去执行。

nohup python3 v1.py &

方案二之使用Python的File方法

采用 python 对文件的操作来实现，用文件对象的 tell(), seek() 方法分别得到当前文件位置和要移动到的位置，我们暂且命名为v2.py，代码如下：

#!/usr/bin/env python import time from S_mail import SendEMail file = open("pay-api_error.log") key_word="wechat" while True: where = file.tell() line = file.readline() if not line: time.sleep(1) file.seek(where) else: if line.find(key_word) >=0 : sm = SendEMail() sm.sendmail(['xx@qq.com'], '主题', '正文') print(line)

解析：

第2行： 导入time模块，为了后续休眠1s用。

第8行： 因为要成为后台进程，这里使用while True 永远为真的形式丢到后台。

第9行： file.tell() 方法返回文件的当前位置，即文件指针指向当前位置。

第10行： file.readline() 方法用于从文件读取整行，包括 "\n" 字符

第11-13行： 如果文件里没有写入内容，我们就让休眠1秒。并且file.seek(),并且移动指针到这个位置。

第14-18行： 如果文件有内容的话，在字符串line里看能不能找到关键字key_word，当然你也可以用 v1的代码xx in line 代替，效果都是一样的。都是为了判断字符串是否还有指定的字符。如果有指定的字符实例化邮件里，然后调用里面的邮件发送函数。

我们测试一下：

$ echo "cc" >> pay-api_error.log $ echo "wechat" >> pay-api_error.log $ python3 v2.py 发送成功 wechat

方案三之使用Python的生成器方法

利用 python 的 yield 来实现一个生成器函数，然后调用这个生成器函数，这样当日志文件有变化并且含有关键字的时候就调用邮件发送类进行告警。我们暂且命名为v3.py，具体的功能代码实现如下：

#!/usr/bin/env python import time from S_mail import SendEMail # 定义变量 file_path="pay-api_error.log" key_word="wechat" def follow(thefile): thefile.seek(0,2) while True: line = thefile.readline() if not line: time.sleep(1) continue yield line if __name__ == '__main__': logfile = open(file_path,'r') logline_xx = follow(logfile) for line in logline_xx: if key_word in line: sm = SendEMail() sm.sendmail(['xx@qq.com'], '主题', '正文') #print(line)

解析：

第10-17行： seek() 函数接收 2 个参数：file.seek(off, whence=0 )，从文件中移动 off 个操作标记（文件指针），正数往结束方向移动，负数往开始方向移动。如果设定了 whence 参数，就以 whence 设定的起始位为准，0 代表从头开始，1 代表当前位置，2 代表文件最末尾位置。

line = thefile.readline() 用于读取文件内容，如果没有内容的话执行休眠并跳过本轮循环，然后进行下一轮循环。yield line 表示如果有内容的话就输出文件内容。这是一个生成器。

第20-21行： 读取文件内容并执行follow函数。

第22-25行： for循环遍历生成器的内容，使用if条件句，如果有关键字key_word就发送邮件

执行结果如下：

$echo "cc" >> pay-api_error.log $echo "wechat" >> pay-api_error.log $ python3 v3.py 发送成功

方案四之使用第三方库pyinotify实现

pyinotify模块用来监测文件系统的变化，依赖于Linux内核的inotify功能，inotify是一个事件驱动的通知器，其通知接口从内核空间到用户空间通过三个系统调用。pyinotify结合这些系统调用，提供一个顶级的抽象和一个通用的方式来处理这些功能。在代码开始之前，我们先来看看pynotify的用法。

安装文档: https://pypi.org/project/pyinotify/

官方文档： https://github.com/seb-m/pyinotify

API文档： http://seb.dbzteam.org/pyinotify/

pip3 install pyinotify # 安装方法 ##创建目录用于后续测试 mkdir /media/tmp

Notifier是pyinotify模块最重要的类，用来读取通知和处理事件，默认情况下，Notifier处理事件的方式是打印事件。

Notifier类在初始化时接受多个参数，但是只有WatchManager对象是必须传递的参数，WatchManager对象保存了需要监视的文件和目录，以及监视文件和目录的哪些事件，Notifier类根据WatchManager对象中的配置来决定如何处理事件。我们做一个简单的测试，暂且命名测试代码为 test.py。

#!/usr/bin/env python3 import pyinotify path="/media/tmp" wm = pyinotify.WatchManager() # 创建WatchManager对象 wm.add_watch(path,pyinotify.ALL_EVENTS) # 添加要监控的目录，以及要监控的事件，这里ALL_EVENT表示所有事件 notifier = pyinotify.Notifier(wm) # 交给Notifier进行处理 notifier.loop() # 循环处理事件

直接结果如下：

$ touch /media/tmp/b $ python3 test.py <Event dir=True mask=0x40000020 maskname=IN_OPEN|IN_ISDIR name='' path=/media/tmp pathname=/media/tmp wd=1 > <Event dir=True mask=0x40000010 maskname=IN_CLOSE_NOWRITE|IN_ISDIR name='' path=/media/tmp pathname=/media/tmp wd=1 > <Event dir=False mask=0x100 maskname=IN_CREATE name=b path=/media/tmp pathname=/media/tmp/b wd=1 > <Event dir=False mask=0x20 maskname=IN_OPEN name=b path=/media/tmp pathname=/media/tmp/b wd=1 >

事件标志：

pyinotify 仅仅是对 inotify 的Python封装，inotify提供了多种事件，基本上事件名称和含义都是相同的。常用的事件标志有：

事件标志	事件含义
IN_ACCESS	被监控项目或者被监控目录中的文件被访问，比如一个文件被读取
IN_MODIFY	被监控项目或者被监控目录中的文件被修改
IN_ATTRIB	被监控项目或者被监控目录中的文件的元数据被修改
IN_CLOSE_WRITE	一个打开切等待写入的文件或者目录被关闭
IN_CLOSE_NOWRITE	一个以只读方式打开的文件或者目录被关闭
IN_OPEN	文件或者目录被打开
IN_MOVED_FROM	被监控项目或者目录中的文件被移除监控区域
IN_MOVED_TO	文件或目录被移入监控区域
IN_CREATE	在所监控的目录中创建子目录或文件
IN_DELETE	在所监控的目录中删除目录或文件
IN_MOVE	文件被移动,等同于IN_CLOSE_NOWRITE

上面列举的是事件的标志位，我们可以用'与'来关联监控多个事件。

multi_event = pyinotify.IN_OPEN | pyinotify.IN_CLOSE_NOWRITE 

到此 基础知识就讲解到这里，我们来开始我们的代码，我们先来一个简单的，先看能不能通过这个库监测到文件的内容。暂且命名我们的代码为v4-1.py。

#!/usr/bin/env python import pyinotify import time import os class ProcessTransientFile(pyinotify.ProcessEvent): def process_IN_MODIFY(self, event): line = file.readline() if line: print(line) # 已经有内容 filename = '/media/tmp/test.txt' file = open(filename,'r') #找到文件的大小并移动到末尾 st_results = os.stat(filename) st_size = st_results[6] file.seek(st_size) wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm) wm.watch_transient_file(filename, pyinotify.IN_MODIFY, ProcessTransientFile) notifier.loop()

代码解析：

1-4行： 导入需要使用的类库。

第6行： 定制化事件处理类，注意是继承关系。

第7行： 定义process_IN_MODIFY 函数，函数名称必须为process_事件名称，event表示事件对象。

第8行： line = file.readline() 每次读取一行，返回的是一个字符串对象。

第9-10行： if条件句，如果有内容就打印文件内容。

第12行： 定义文件的路径。

第13行： 打开文件，其中r 表示以只读方式打开文件。文件的指针将会放在文件的开头，这是也是默认模式。

第16行： os.stat() 方法用于在给定的路径上执行一个系统 stat 的调用。结果如下（以下是ipython的执行结果）：

In [3]: st_results = os.stat("/media/tmp/test.txt") In [4]: print(st_results) os.stat_result(st_mode=33188, st_ino=917693, st_dev=64769, st_nlink=1, st_uid=1144, st_gid=40001, st_size=15, st_atime=1589537157, st_mtime=1589537157, st_ctime=1589537157) In [5]: print(st_results[6]) 15 In [6]: print(type(st_results)) <class 'os.stat_result'>

上面的结果和Linux的命令一致如下,，返回的内容是Size的大小：

$ stat test.txt File: ‘test.txt’ Size: 15 Blocks: 8 IO Block: 4096 regular file Device: fd01h/64769d Inode: 917693 Links: 1 Access: (0644/-rw-r--r--) Uid: ( 1144/knight.zhou) Gid: (40001/ sa) Access: 2020-05-15 18:05:57.554290444 +0800 Modify: 2020-05-15 18:05:57.554290444 +0800 Change: 2020-05-15 18:05:57.554290444 +0800 Birth: -

第18行： file.seek() 方法用于移动文件读取指针到指定位置。

第20行： 创建监控实例。

第21行: notifier = pyinotify.Notifier(wm) 用于绑定一个事件。

第22行： wm.watch_transient_file(filename, pyinotify.IN_MODIFY, ProcessTransientFile) 表示 添加监控的对象，我们使用的标志事件为IN_MODIFY （被监控项目或者被监控目录中的文件被修改）。

最后notifier.loop() 运行监控。

执行结果如下：

$touch /media/tmp/test.txt $echo "bb" >> test.txt $echo "cc" >> test.txt $ python3 v4-1.py bb cc

上面的基本功能是实现了，我们现在来结合我们的需求来实现我们的代码。我们暂且命名我们的最终代码为v4-last.py。

#!/usr/bin/env python import pyinotify import time import os from S_mail import SendEMail ## 定义变量 key_word="wechat" class ProcessTransientFile(pyinotify.ProcessEvent): def process_IN_MODIFY(self, event): line = file.readline() if key_word in line: print("有{0},发送告警".format(key_word)) sm = SendEMail() sm.sendmail(['xx@qq.com'], '主题', '正文') filename = '/media/tmp/pay-api_error.log' file = open(filename,'r') #找到文件的大小并移动到末尾 st_results = os.stat(filename) st_size = st_results[6] file.seek(st_size) wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm) wm.watch_transient_file(filename, pyinotify.IN_MODIFY, ProcessTransientFile) notifier.loop()

代码解析：

第14-17行： 判断读取的每一行里是否有关键字，有关键字就发送邮件。

其余代码功能和之前的一样。

执行结果如下：

$ echo "xx" > pay-api_error.log $ echo "wechat" >> pay-api_error.log $ python3 v4-last.py 有wechat,发送告警 发送成功

总结

实时监控文件，我们介绍了以上4种方法去实现，个人觉得第4种方法更加科学。就像我们搭建RSYNC服务器一样，如果想实时同步文件，我们一般结合 rsync+inotify 去实现。虽然这种方法难以理解一点但还是推荐使用。

多维度详解
手把手入门
《从头解锁Python运维》，专栏完结福利，开启限时拼团>>>