python怎么实时监控logstash日志

第一步，实时读取logstash日志，有异常错误keywork即触发报警。

# /usr/b/env python3
# -- codg: utf-8 --
# __author__ = caozhi
# create_time 2018-11-12,update_time 2018-11-15
# version = 1.0
# 录像高可用报警
# 1 读取日志 使用游标移动
# 2 线上业务日志文件会切割，切割后，读取上一个切割的日志

import os
import sys
import json
import requests
import time
import re

ci = conf.i'
log_file = logstash.log'

def readconf():
 try:
 with open(ci, 'r+') as f:
  CONF = json.load(f)
 except:
 CONF = {"seek": 0, "ode": 922817, "last_file": logstash.log"}
 writeconf(CONF=CONF)
 prt('conf.i 配置文件缺失，自动创建一个新的配置文件')
 return CONF

def writeconf(CONF):
 with open(ci, 'w+') as e:
 json.dump(CONF, e)

def read_log(log_file, seek):
 try:
 f = open(log_file, 'r')
 except FileNotFoundError:
 f = open(logstash.log', 'r')
 seek = 0
 prt('上一个文件读取失败了，请检查切割的日志文件')
 except:
 prt('日志文件打开错误，退出程序')
 sys.exit()

f.seek(seek)
le = f.readle()
new_seek = f.tell()
if new_seek == seek:
 prt('没有追加日志，退出程序')
 sys.exit()

while le:
 try:
 logstash = json.loads(le)
 except:
 CONF = {"seek": 0, "ode": 922817, "last_file": "/data/logs/lmrs/logstash.log"}
 writeconf(CONF=CONF)
 prt('json数据加载错误，重新创建一个新的配置文件')
 sys.exit()

 #if '''re.search(time.strftime("%Y:%H:%M", time.localtime()), logstash.get('log_time')) and '''logstash.get('rtype') == 6 and logstash.get('uri') == '/publish' and logstash.get('event') == 0:
 if logstash.get('rtype') == 6 and logstash.get('uri') == '/publish' and logstash.get('event') == 0:
 value = 1
 stream = logstash.get('name')
 prt('{} {}'.format(value, stream))
 record(value=value, stream=stream)
 else:
 value = 0
 stream = 0
 le = f.readle()
seek = f.tell()
f.close
return value, stream, seek

def record(value, stream):
 data = []
 record = {}
 record['metric'] = 'recordg_high_availability_monitor'
 record['endpot'] = os.uname()[1]
 record['timestamp'] = t(time.time())
 record['step'] = 60
 record['value'] = value
 record['counterType'] = 'GAUGE'
 record['Tags'] = '{}={}'.format(t(time.time()), stream)
 data.append(record)

if data:
 prt('这是data的json数据')
 prt(data)
 falcon_request = requests.post("http://127.0.0.1:1988/v1/push", data=json.dumps(data))
 #falcon_request = requests.post("http://127.0.0.1:1988/v1/push", json=data)
 prt('json参数请求返回状态码为' + str(falcon_request.status_code))
 prt('json参数请求返回为' + str(falcon_request.text))

if __name__ == '__ma__':
 prt()
 prt('')
 prt('本次执行脚本时间{}'.format(time.strftime("%Y%m%d_%H%M", time.localtime())))
 CONF = readconf()
 prt('first_CONF :{}'.format(CONF))
 prt('NO1.log_file',log_file)
 last_ode = CONF['ode']
 ode = os.stat(log_file).st_o
 prt('last_ode: {} ode: {}'.format(last_ode, ode))

if ode == last_ode:
 seek = CONF['seek']
 next_file = 0
else:
 log_file = CONF['last_file'] + time.strftime("-%Y%m%d_", time.localtime()) + str(time.strftime("%H%M", time.localtime()))[:-1] + '0'
 next_file = 1
 seek = CONF['seek']

prt('NO2.log_file',log_file)
value, stream, seek = read_log(log_file=log_file,seek=seek)

if next_file:
 CONF['seek'] = 0
else:
 CONF['seek'] = seek

CONF['ode'] = os.stat(logstash.log').st_o
writeconf(CONF=CONF)
prt('last_CONF :{}'.format(CONF))