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*(?<Level>(\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\n";
 my $subject = "Subject: zjcap info\n";
 my $info = "$CurrTime--$message";
 my $message = <<CONTENT;
 $info
CONTENT
 my $smtp = Net::SMTP->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("\n\n");
 $smtp->datasend("$message\n"); # 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"
}

以上这篇python实时监控logstash日志代码就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持我们。

(0)

相关推荐

  • Python如何把Spark数据写入ElasticSearch

    这里以将Apache的日志写入到ElasticSearch为例,来演示一下如何使用Python将Spark数据导入到ES中. 实际工作中,由于数据与使用框架或技术的复杂性,数据的写入变得比较复杂,在这里我们简单演示一下. 如果使用Scala或Java的话,Spark提供自带了支持写入ES的支持库,但Python不支持.所以首先你需要去这里下载依赖的ES官方开发的依赖包包. 下载完成后,放在本地目录,以下面命令方式启动pyspark: pyspark --jars elasticsearch-ha

  • python logging通过json文件配置的步骤

    logconfig.json { "version":1, "disable_existing_loggers":false, "formatters":{ "simple":{ "format":"[%(asctime)s - %(levelname)s - line(%(lineno)d) - %(filename)s]: %(message)s", "datefmt&qu

  • 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 impor

  • python脚本监控logstash进程并邮件告警实例

    supervisor虽然也能拉起来logstash进程,但是有时候supervisor也会挂,也有时会拉不起,就算拉起来了也没有邮件告警功能 ,所以编写一个python脚本监控所有服务器,以下代码只列举了一台服务器,需要更多服务器在列表里面添加就行! (日志采集过程中连续几天数据异常,由于服务器太多,当时不太想一一去将近40台服务器查看logstash进程,但又一直查不出原因,后来就每台服务器查看logstash进程,果然发现3台采集搜索日志的logstash进程没了,然后就编写了此脚本监控,效

  • python实时监控cpu小工具

    本文实例为大家分享了python实时监控cpu的工具,供大家参考,具体内容如下 虽然写的很不完善,但是当练手吧,对于实时监控cpu还是有点用处的,虽然android studio已经提供了一样的功能,但是能够记录并生成成本地csv文件还是比as多一个功能的. 结果图: 直接放源码: # !_*_coding:utf-8_*_ import re import os import wx import thread import pychartdir import csv class GetCpuS

  • Python实时监控网站浏览记录实现过程详解

    需求: (1) 获取你对象chrome前一天的浏览记录中的所有网址(url)和访问时间,并存在一个txt文件中 (2)将这个txt文件发送给指定的邮箱地址(你的邮箱) (3)建立例行任务,每天定时自动完成这些操作,你就可以通过邮件查看你对象每天看啥了 准备 macOS Sierra Python3.6 Chrome 发送邮件的qq邮箱地址 qq邮箱授权码 SMTP服务器地址 : smtp.qq.com 接受邮件的邮箱地址 执行: (1) 首先我们用DB Browser for SQLite来看下

  • Linux下用Python脚本监控目录变化代码分享

    #!/usr/bin/env python #coding=utf-8 import os from pyinotify import WatchManager, Notifier, ProcessEvent, IN_DELETE, IN_CREATE,IN_MODIFY wm = WatchManager() mask = IN_DELETE | IN_CREATE |IN_MODIFY # watched events class PFilePath(ProcessEvent): def p

  • 基于Python实现实时监控CPU使用率

    目录 导语 一.模块简介 二.准备中 三.开始敲代码 3.1 代码内容 3.2 附完整代码 四.效果展示 导语 哈喽!大家好,我是木木子!准时上线给大家更新啦!​ Ps:小故事 前两天有位粉丝朋友在群里问我说,有没有不装一些我们熟知的电脑管理软件但是依然可以查看CPU的使用率方法,或者能够介绍一款比较给力的小插件也好.确实,在生活中有不少人是不喜欢装那些管理软件的. 今天,木木子就给大家介绍一款手写编程代码的小脚本,能够轻松在界面上展示:利用Python实时监控CPU使用率,随时展现.也无需下载

  • python实现实时监控文件的方法

    在业务稳定性要求比较高的情况下,运维为能及时发现问题,有时需要对应用程序的日志进行实时分析,当符合某个条件时就立刻报警,而不是被动等待出问题后去解决,比如要监控nginx的$request_time和$upstream_response_time时间,分析出最耗时的请求,然后去改进代码,这时就要对日志进行实时分析了,发现时间长的语句就要报警出来,提醒开发人员要关注,当然这是其中一个应用场景,通过这种监控方式还可以应用到任何需要判断或分析文件的地方,所以今天我们就来看看如何用python实现实时监

  • python动态监控日志内容的示例

    日志文件一般是按天产生,则通过在程序中判断文件的产生日期与当前时间,更换监控的日志文件程序只是简单的示例一下,监控test1.log 10秒,转向监控test2.log 程序监控使用是linux的命令tail -f来动态监控新追加的日志 复制代码 代码如下: #!/usr/bin/python# encoding=utf-8# Filename: monitorLog.pyimport osimport signalimport subprocessimport time logFile1 =

  • python实现tail实时查看服务器日志示例

    我就废话不多说了,直接上代码吧! import paramiko from paramiko_expect import SSHClientInteraction host = your host port = your port username = your un # 自行修改输出函数 json_list = [] def output_func(msg): sys.stdout.write(msg) json_list.append(msg) sys.stdout.flush() def

  • Python 实现网课实时监控自动签到、打卡功能

    响应国家停课不停学的号召,学生们都开始了网上授课,但由于课程繁多,消息繁杂,经常错过课堂签到,针对这一难题,博客主作为Python爱好者,完全使用Python语言写了本篇博客,希望能够帮助小伙伴们完成上课签到问题(注:只是帮你签到,而不是叫你代签,我的出发点是帮助记性不是很好的同学签到,防止漏签被老师点名),所以希望大家能理解博主的苦心. 话不多说,献上效果图两张 进入正式教程 ①Python环境 Python3.6及以上版本,需要配备的库requests,json,time 代码运行软件:Py

随机推荐