起因：

本公司使用canal+kafka+自研中间件同步指定表数据到下游。

在每个环境都有端口监控，但是有天突然被反馈异常，”下游节点看不到上游更新的数据“，奇怪的是并没有监控报警，也就是说各个环境的组件都还是在运行的。这就很尴尬了，肯定是有个环节出了bug或异常但是程序没有退出。

挨着排查后确定是canalGG了，具体问题是canal在网络波动时与下游kafka失联了，在超时后canal开始不断报错”无法找到kafka“，但是此时canal认为是下游kafka出了问题，本身是没有问题的，所以canal只是日志报警但并未触发异常退出。

好吧，那就只能简单的撸个py脚本了，监测一下日志中是否出现error（正常情况下canal是没有什么日志的）

python脚本扫描日志并发送dingding告警

 \# -\*- coding: utf-8 -\*-  
 import socket  
 import time  
 import hmac  
 import hashlib  
 import base64  
 import requests  
 import urllib  
 import paramiko  
   
 #发送钉钉告警  
 def ding(title, options: dict):  
 content = '\\n'.join(\[f"+ \*\*{k}\*\*: {v}" for k, v in options.items()\])  
 secret = 'xxxxxx'  
 token = 'xxxxxx'  
 timestamp = int(time.time() \* 1000)  
 data = (str(timestamp) + '\\n' + secret).encode('utf-8')  
 hmac\_code = hmac.new(secret.encode('utf-8'), data, digestmod=hashlib.sha256).digest()  
 sign = urllib.parse.quote\_plus(base64.b64encode(hmac\_code))  
 url = f'https://oapi.dingtalk.com/robot/send?access\_token={token}×tamp={timestamp}&sign={sign}'  
 r = requests.post(url, json={  
 "msgtype": "markdown",  
 "markdown": {  
 "title": title,  
 "text": content,  
 }  
 })  
   
 #监控日志文件  
 # SSH连接远程服务器  
 # 获取命令输出结果  
 last\_status = 1  
 while True:  
 # 对结果进行判断  
 ssh = paramiko.SSHClient()  
 ssh.set\_missing\_host\_key\_policy(paramiko.AutoAddPolicy())  
 ssh.connect('192.168.xxx.xxx', username='root', password='xxxxxx')  
 stdin, stdout, stderr = ssh.exec\_command('grep error /opt/canal/logs/dtk/dtk.log')  
 output = stdout.read().decode('utf-8')  
 if 'error' in output:  
 #print("日志文件中包含 error")  
 ding("canal状态监控", {"canal异常":"日志文件出现ERROR"})  
 last\_status = 0  
 if 'error' not in output and last\_status == 0:  
 ding("canal状态监控", {"canal正常":"canal已恢复正常"})  
 last\_status=1  
 ssh.close()  
 time.sleep(60)