python實時監控logstash日誌代碼

實時讀取logstash日誌，有異常錯誤keywork即觸發報警。

# /usr/bin/env python3 # -*- coding: 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 cini = conf.ini' log_file = logstash.log' def readconf(): try: with open(cini, 'r+') as f: CONF = json.load(f) except: CONF = {"seek": 0, "inode": 922817, "last_file": logstash.log"} writeconf(CONF=CONF) print('conf.ini 配置文件缺失，自動創建一個新的配置文件') return CONF def writeconf(CONF): with open(cini, '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 print('上一個文件讀取失敗了，請檢查切割的日誌文件') except: print('日誌文件打開錯誤，退出程序') sys.exit() f.seek(seek) line = f.readline() new_seek = f.tell() if new_seek == seek: print('沒有追加日誌，退出程序') sys.exit() while line: try: logstash = json.loads(line) except: CONF = {"seek": 0, "inode": 922817, "last_file": "/data/logs/lmrs/logstash.log"} writeconf(CONF=CONF) print('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') print('{} {}'.format(value, stream)) record(value=value, stream=stream) else: value = 0 stream = 0 line = f.readline() seek = f.tell() f.close return value, stream, seek def record(value, stream): data = [] record = {} record['metric'] = 'recording_high_availability_monitor' record['endpoint'] = os.uname()[1] record['timestamp'] = int(time.time()) record['step'] = 60 record['value'] = value record['counterType'] = 'GAUGE' record['Tags'] = '{}={}'.format(int(time.time()), stream) data.append(record) if data: print('這是data的json數據') print(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) print('json參數請求返回狀態碼為：' + str(falcon_request.status_code)) print('json參數請求返回為：' + str(falcon_request.text)) if __name__ == '__main__': print() print('***************************************') print('本次執行腳本時間：{}'.format(time.strftime("%Y%m%d_%H%M", time.localtime()))) CONF = readconf() print('first_CONF :{}'.format(CONF)) print('NO1.log_file',log_file) last_inode = CONF['inode'] inode = os.stat(log_file).st_ino print('last_inode: {} inode: {}'.format(last_inode, inode)) if inode == last_inode: 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'] print('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['inode'] = os.stat(logstash.log').st_ino writeconf(CONF=CONF) print('last_CONF :{}'.format(CONF))

補充知識：logstash 調用exec

我就廢話不多說了，還是直接看代碼吧！

[elk@Vsftp logstash]$ cat t3.conf input { stdin { } } filter { grok { match => [ "message","(?m)s*%{TIMESTAMP_ISO8601:time}s*(?

(S+)).*"] } date { match => ["time", "yyyy-MM-dd HH:mm:ss,SSS"] } mutate { add_field =>["type","tailong"] add_field =>["messager","%{type}-%{message}"] remove_field =>["message"] } } output { if ([Level] == "ERROR" or [messager] =~ "Exception" ) and [messager] !~ "溫金服務未連接" and [messager] !~ "調用溫金代理系統接口錯誤" and [messager] !~ "BusinessException" { exec { command => "/bin/smail.pl "%{messager}" "%{type}" " } } stdout { codec =>rubydebug } } Vsftp:/root# cat /bin/smail.pl #!/usr/bin/perl use Net::SMTP; use HTTP::Date qw(time2iso str2time time2iso time2isoz); use Data::Dumper; use Getopt::Std; use vars qw($opt_d ); getopts('d:'); # mail_user should be your_mail@163.com $message= "@ARGV"; $env="$opt_d"; sub send_mail{ my $CurrTime = time2iso(time()); my $to_address = shift; my $mail_user = 'zhao.yangjian@163.com'; my $mail_pwd = 'xx'; my $mail_server = 'smtp.163.com'; my $from = "From: $mail_user

"; my $subject = "Subject: zjcap info

"; my $info = "$CurrTime--$message"; my $message = <new($mail_server); $smtp->auth($mail_user, $mail_pwd) || die "Auth Error! $!"; $smtp->mail($mail_user); $smtp->to($to_address); $smtp->data(); # begin the data $smtp->datasend($from); # set user $smtp->datasend($subject); # set subject $smtp->datasend("



"); $smtp->datasend("$message

"); # set content $smtp->dataend(); $smtp->quit(); }; send_mail ('zhao.yangjian@163.com'); 2017-01-12 10:19:19,888 jjjjj Exception { "@version" => "1", "@timestamp" => "2017-01-12T02:19:19.888Z", "host" => "Vsftp", "time" => "2017-01-12 10:19:19,888", "Level" => "jjjjj", "type" => "tailong", "messager" => "tailong-2017-01-12 10:19:19,888 jjjjj Exception" }