用Juggernaut实时推送syslog分析结果

#!/usr/bin/perl
use AnyEvent;
use Getopt::Long;
use Redis;
use JSON;
use POSIX qw/ strftime /;
use warnings;
use strict;
use 5.010;
# 后台运行
use App::Daemon qw( daemonize );
$App::Daemon::as_user    = "root";
daemonize();
# 设定调试和间隔
my ($debug, $interval, $help);                                                                                                                                                        
GetOptions(                                                                                                                                                                           
    "debug|d" => \$debug,                                                                                                                                                             
    "interval|i=i" => \$interval,                                                                                                                                                     
    "help|h" => \$help,                                                                                                                                                               
);                                                                                                                                                                                    
if( $help ) {
    say "Usage: $0 [start|stop|-X] -d -i num";
    say "       -X means run frontend;";
    say "       -d means debug for timer and submit;";
    say "       -i means define a special interval seconds for regexp and submit, default set 300s.";
};
$interval = 300 unless $interval;

my @str = ();
# syslog的pri，定义见http://www.ietf.org/rfc/rfc3164.txt。因为本例只收集kernel信息，即Facility = 0,所以PRI = 0 * 8 + Severity。这里为了页面好看直接写成bootstrap里button的class了。
my @pri = qw(
    btn-inverse
    btn-danger
    btn-warning
    btn-success
    btn-info
    btn-primary
    btn
    disabled
);
# syslog格式<IP> <TIME> <PRI> kernel: <MSG>
# 注意最后msg里的\S.+，因为当内存出错等情况时，msg里开头会以空格表示附属关系
my $re = qr/((?:\d{1,3}\.){3}\d{1,3}) \[(\w+ \d+ \d{2}:\d{2}:\d{2})\] <(\d+)> kernel: (\S.+)/;
# 目前是直接收集成时间格式命名了，所以每天crond里要restart脚本，如果是日志名不变，crond切割的，那么脚本可以一直跑
open(my $r, "-|", "tail", "-F", '/data1/syslog/kern.' . strftime("%Y%m%d", localtime) . '.log' ) or die "can't fork: $!";

my $io = AnyEvent->io(
    fh => $r,
    poll => "r",
    cb => sub {
        my $input = scalar <$r>;
        return if $input =~ /repeat|suppress|window/;
        push @str, $input;
    }
);

my $w = AnyEvent->timer(
    after => $interval,
    interval => $interval,
    cb => sub {
        my $data;
        say "######## ",time if $debug;
        for ( @str ) {
            next unless $_ =~ $re;
            my ( $ip, $time, $level, $msg ) = ( $1, $2, $3, $4 );
            next if exists $data->{$ip};
            $data->{$ip} = classify($msg);
            submit( $time, $ip, $pri[$level], $data->{$ip}, $msg );
        };
        @str = ();
    },
);
AnyEvent->condvar->recv;

sub submit {
    my @msg = @_;
    say @msg if $debug;
    my $redis = Redis->new( server => '198.168.0.2:6379' );
    $redis->publish("juggernaut", to_json({
        channels => ['channel1'],
        data => \@msg,
    }));
};

sub classify {
    my $msg = shift;
    if ( $msg =~ /TCP|UDP|SYN|socket/ ) {
        'network';
    } elsif ( $msg =~ /segfault|swap|mem|allocation/ ) {
        'memory';
    } elsif ( $msg =~ /IPMI|EXT|cciss|scsi|mpt|usb|DRAC|sd\w/ ) {
        'disk';
    } elsif ( $msg =~ /CPU|IRQ/ ) {
        'cpu';
    } else {
        'unknown';
    };
};