MySQL Error log监控

阅读量

MySQL Error log监控

告警日志监控其实比较简单了, 怎么做都行. 目前我们这里是使用下面的方法
FileBeat采集日志 -> Kafka -> 自己写脚本消费出来 -> 企业微信机器人告警
image
效果如下
image

这里给出FileBeat配置, 很简单

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
[root@node002142 filebeat-7.2.0-dba]# cat filebeat.yml
logging:
  level: warning
  json: true
filebeat.config.inputs:
  enabled: true
  path: configs/*.yml
  reload.enabled: true
  reload.period: 10s
http:
  enabled: true
  host: "0.0.0.0"
  port: 5066
processors:
  - drop_fields:
      fields: ["beat.name", "beat.version", "input_type", "offset"]
  - add_host_metadata:
      netinfo.enabled: true
output.kafka:
  hosts: ["192.168.x.xx:9092","192.168.x.xx:9092","192.168.x.xx:9092"]
  topic: '%{[kafka_topic]}'
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  max_message_bytes: 8388608
  compression: gzip
  bulk_max_size: 2048
  worker: 6
  keep_alive: 600
  channel_buffer_size: 2560
  version: 2.0.0  --filebeat7.2才支持最新版本的kafka, 虽然我们的kafka是2.1.2 但是这里也得写2.0.0

[root@node002142 configs]# ll
total 8
-rw-r--r-- 1 root root 247 Jul 15 18:40 mysql_error_log.yml

[root@node002142 filebeat-7.2.0-dba]# cat configs/mysql_error_log.yml
- type: log
  paths:
   - /data/mysql_*/logs/*.err
  fields:
    type: mysql_error_log
    format: plain
    kafka_topic: log_mysql_error_log
  fields_under_root: true
  max_backoff: 3s

写入Kafka的消息如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
{
    "@timestamp": "2019-07-21T01:43:22.091Z",
    "@metadata": {
        "beat": "filebeat",
        "type": "_doc",
        "version": "7.2.0",
        "topic": "log_mysql_error_log"
    },
    "input": {
        "type": "log"
    },
    "kafka_topic": "log_mysql_error_log",
    "type": "mysql_error_log",
    "format": "plain",
    "host": {
        "name": "node00xxx",
        "id": "ea3afe477be14c22abf234dd3cb80f55",
        "containerized": false,
        "ip": ["10.1.x.xx, "fe80::xx:xx:xx", "192.168.x.xx", "fe80::xx:xx:xx"],
        "mac": ["80:18:xx:xx:xx", "80:18:xx:xx:xx", "80:18:44:xx:xx:xx", "80:18:44:xx:xx:xx"],
        "hostname": "node002111",
        "architecture": "x86_64",
        "os": {
            "platform": "centos",
            "version": "7 (Core)",
            "family": "redhat",
            "name": "CentOS Linux",
            "kernel": "4.15.9-1.el7.elrepo.x86_64",
            "codename": "Core"
        }
    },
    "agent": {
        "ephemeral_id": "c97d819c-c456-42d0-xxxx-xxxxxxxxx",
        "hostname": "node00xxx",
        "id": "359f6ddb-de27-42c9-9ce1-1624205d6af0",
        "version": "7.2.0",
        "type": "filebeat"
    },
    "log": {
        "offset": 2585741,
        "file": {
            "path": "/data/mysql_3306/logs/node00xxxx.err"
        }
    },
    "message": "2019-07-21T01:43:21.585528Z 4345026 [Note] Access denied for user 'user'@'192.168.x.xx' (using password: YES)",
    "ecs": {
        "version": "1.0.0"
    }
}

Python脚本
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
# -*- coding: utf8 -*-
# __author__ = 'Fan()'
# Date: 2019-07-18

import time
import json
import pytz
import requests
import datetime
import logging

from utils.conn_db import Fandb
from utils.config import *
from confluent_kafka import Consumer, KafkaError, TopicPartition, OFFSET_END, OFFSET_BEGINNING, Producer


class MyRequest():
    @staticmethod
    def get(url, params=None, timeout=(2, 5)):
        response = requests.get(url=url, params=params, timeout=timeout)
        if response.status_code == requests.codes.ok:
            return response.json()
        else:
            response.raise_for_status()

    @staticmethod
    def post(url, data=None, json=None, timeout=(5)):
        response = requests.post(url=url, data=data, json=json, timeout=timeout)
        if response.status_code == requests.codes.ok:
            return response.json()
        else:
            response.raise_for_status()


def confLog(logfile):
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S',
                        filename=logfile,
                        filemode='a')


def _on_send_response(err, partations):
    pt = partations[0]
    if isinstance(err, KafkaError):
        print('Topic {} 偏移量 {} 提交异常. {}'.format(pt.topic, pt.offset, err))
        logging.error('Topic {} 偏移量 {} 提交异常. {}'.format(pt.topic, pt.offset, err))
       # raise Exception(err)


def getConsumer(topic_name, bootstrap_servers, offset_end=True):
    config = {'bootstrap.servers': bootstrap_servers,
              "group.id": topic_name,
              'enable.auto.commit': True,
              "fetch.wait.max.ms": 3000,
              "max.poll.interval.ms": 60000,
              'session.timeout.ms': 60000,
              "on_commit": _on_send_response,
              "default.topic.config": {"auto.offset.reset": "latest"}}

    consumer = Consumer(config)
    offset = OFFSET_END if offset_end else OFFSET_BEGINNING
    pt = TopicPartition(topic_name, 0, offset)  # 动态获取 一级kafka的 topic
    consumer.assign([pt])
    # consumer.seek(pt)

    try:
        while True:
            ret = consumer.consume(num_messages=10, timeout=0.1)
            if ret is None:
                print("No message Continue!")
                continue
            for msg in ret:
                if msg.error() is None:
                    # print("Received message:{}".format(msg.value().decode("utf-8")))
                    yield msg.value().decode("utf-8")
                elif msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                else:
                    raise Exception(msg.error())
    except Exception as e:
        print(e)
        consumer.close()
    except KeyboardInterrupt:
        consumer.close()


def utc_to_local(utc_time_str, utc_format='%Y-%m-%dT%H:%M:%S.%fZ'):
    local_tz = pytz.timezone('Asia/Chongqing')
    local_format = "%Y-%m-%d %H:%M:%S"
    utc_dt = datetime.datetime.strptime(utc_time_str, utc_format)
    local_dt = utc_dt.replace(tzinfo=pytz.utc).astimezone(local_tz)
    time_str = local_dt.strftime(local_format)
    return time_str


def get_mysql_ip(ips, hostname):
    last_ip = str(int(hostname[-3:]))
    for i in ips:
        x = i.split('.')
        if len(x) == 4:
            # 过滤掉本机vip
            if x[3] == last_ip and x[2] not in ('3', '8', '16') and x[0] == '192':
                return i


def get_mysql_port(error_log_file):
    spliter = 'mysql_'

    mysql_port = error_log_file.split(spliter)[1].split('/')[0]
    if mysql_port == '':
        mysql_port = 3306

    return mysql_port


def sendWechatBot(send_message):
    web_hoot_address = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人key'
    body = {
    "msgtype": "markdown",
    "markdown": {
        "content": '''<font color=\"warning\">捕获告警日志报错信息:</font>\n
>产品线:<font color=\"comment\">{product_name}</font>
>项目名称:<font color=\"comment\">{project_name}</font>
>高可用组:<font color=\"comment\">{ha_group_name}</font>
>IP:<font color=\"comment\">{ip_app}</font>
>PORT:<font color=\"comment\">{port}</font>
>节点角色:<font color=\"comment\">{role_name}</font>
>告警时间:<font color=\"comment\">{error_timestamp}</font>
>捕获时间:<font color=\"comment\">{catch_timestamp}</font>
>下发时间:<font color=\"comment\">{send_timestamp}</font>

告警内容:
<font color=\"comment\">{error_message}</font>\n
'''.format(**send_message)
        }
    }
    data = MyRequest.post(web_hoot_address, json=body)
    return data


def get_mysql_info(mysql_ip, mysql_port):
    conn = Fandb(cmdb_host, cmdb_port, cmdb_user, cmdb_pass, cmdb_schema, dic=True)
    sql = '一个根据ip端口查询数据库实例想关信息的SQL'
    res = conn.dql(sql)
    conn.close()
    return res[0]


if __name__ == '__main__':
    verbose = 1
    logfile = '/tools/mysql_error_log_watchdog.log'
    confLog(logfile)

    topic = 'log_mysql_error_log'
    bootstrap_servers = "192.168.x.xxx:9092,192.168.x.xxx:9092,192.168.x.xxx:9092"

    consumer = getConsumer(topic, bootstrap_servers)

    for message in consumer:
        message_dict = json.loads(message)

        if verbose >= 3: print(message_dict)

        catch_timestamp = utc_to_local(message_dict['@timestamp'])
        error_log_file = message_dict['log']['file']['path']
        hostname = message_dict['host']['hostname']
        ips = message_dict['host']['ip']
        mysql_ip = get_mysql_ip(ips, hostname)
        mysql_port = get_mysql_port(error_log_file)

        error_log_message = message_dict['message']
        if '[ERROR]' in error_log_message:
            try:
                catch_timestamp = utc_to_local(message_dict['@timestamp'])
                error_log_file = message_dict['log']['file']['path']
                hostname = message_dict['host']['hostname']
                ips = message_dict['host']['ip']
                mysql_ip = get_mysql_ip(ips, hostname)
                mysql_port = get_mysql_port(error_log_file)

                error_timestamp = utc_to_local(error_log_message.split(' ')[0])
                error_message = ' '.join(message_dict['message'].split(' ')[2:])
                send_message_dict = get_mysql_info(mysql_ip, mysql_port)
                send_message_dict['catch_timestamp'] = catch_timestamp
                send_message_dict['error_timestamp'] = error_timestamp
                send_message_dict['send_timestamp'] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                send_message_dict['error_message'] = error_message
                logging.info(send_message_dict)
                sendWechatBot(send_message_dict)
                time.sleep(3) #企业微信机器人限制3秒一条
            except Exception as e:
                logging.exception(e)
                logging.exception(error_log_message)

如果觉得上面的太麻烦,其实一个shell脚本也可以搞, 就是得在每个机器部署运行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#!/bin/bash
# Oracle警告日志文件监控脚本
# 2015/4/14 King.

# 发送邮件
sendMail()
{
  echo "$1"
  mailTo=$(echo $mailTo | sed 's/,/ /g')
  echo "$1" | /usr/bin/mutt -s "$(date +"%Y-%m-%d %H:%M:%S") 警告日志错误" -b ${mailTo} -c ${mailCc}
}

# 判断错误信息中是否有未扫描的行
checkHis()
{
  tag=0
  hisLine=$(cat $errLineNumFile)
  for i in $hisLine
  do
    if [ "$1" == "$i" ]; then
      tag=1
    fi    
  done
  return $tag  
}

#
scriptDir=`pwd $0`
scriptName=`basename $0`
logDir=$scriptDir/logs
logfile=$logDir/alert_error.log
errLineNumFile=$logDir/.alert_errLineNum


#设置警告日日志文件路径
alertFilePath="/data/mysql_3306/logs/nodexx.err"

# 设置邮件接收者，多个用逗号分隔
mailTo="xx@163.com"

# 设置邮件抄送者，多个用逗号分隔
mailCc="a@163.com,b@163.com"

[ ! -f $alertFilePath ] && echo "[Error]: $alertFilePath no such file or directory." && exit 1
[ ! -d $logDir ] && mkdir -p  $logDir
touch $errLineNumFile
echo "正在监控 $alertFilePath... "
while true
do
  arrayNum=()
  isError=false
  # 取出警告日志中 ”ORA?“关键字所在的行
  errNum=$(cat $alertFilePath | grep -n -i "[ERROR]]")
  n=0
  if [ "x$errNum" != "x" ]; then
    # 取出错误行号
    errLineNum=$(echo "$errNum" | awk -F: '{print $1}')
    for num in $errLineNum
    do
      #判断该行错误信息是否已扫描      
      if [ "x$errLineNum" != "x" ]; then
        checkHis "$num"
        if [ $? -eq 0 ]; then
          # 如果该行错误未扫描，记录该行信息
          isError=true  
          echo $num >> $errLineNumFile
          arrayNum[$n]=$num
          let n++
        fi        
      else
        #如果没有错误休眠10s后重新扫描    
        sleep 10s
        break
      fi    
   done
fi
  # 如果发现未扫描的错误信息则根据行号取出该行信息记录日志，并且发送邮件
  if [ "$isError"  == "true" ]
  then
    echo "-------------------------------- $(date +"%Y-%m-%d %H:%M:%S") ---------------------------------------"  >> $logfile
    i=0
    errMsg=$(
    while [ $i -lt ${#arrayNum[@]} ]; do
      echo "$errNum" | grep "^${arrayNum[$i]}:"
      let i++
    done)   
    echo "$errMsg" >> $logfile
    sendMail "$errMsg"
  fi
  #每10s，扫描一次警告日志文件
  sleep 10s
done

Powered by Hexo and Hexo-theme-hiker

Copyright © 2013 - 2022 Fan() All Rights Reserved.

访客数 : | 访问量 :