基于perl的C/S模式的性能监控预警系统

基于perl的C/S模式的性能监控预警系统

因为要做毕业设计的原因，就有了这么一个东西···网页哥不懂，所以就没做成B/S模式的···再说，b/s模式的只能用SNMP吧？好像也不能按照自己的实际情况监控一些系统的资源·额··反正各有各好处啦····

现在这个只是基本的框架，基本的功能已经实现了，不过bug肯定是还有很多的···还请各位指正....

服务端负责收集客户端的数据，将数据写入rrd数据库里边（可以生成图表），然后判断是否超过阀值，超过就发报警的邮件···当然，如果某些监控的项目异常了，例如ftp服务挂掉，服务端也可以发送重启ftp服务的命令（通过客户端执行已存在的脚本实现）来修复挂掉的服务·····额··只是想到那么多了···

ps：rrd（RRDtool）数据库要自己建立，还有图表的生成也要自己手动生成，服务端只负责写数据··

服务端如下：

#BY Weigun http://blog.chinaunix.net/u2/70443/

use warnings;
use strict;
use 5.010;
use lib 'D:/test-area/desgin/Module';
use PublicLib;
use YAML qw(Dump LoadFile);
use Config::IniFiles;
use Data::Dumper;
#use AnyEvent;

#use Coro;

use POE;
use POE::Filter::Reference;
use POE::Component::Server::TCP;
#use POE::Component::Client::SMTP;

use Mail::Sender;
use RRDTool::OO;
use Log::Log;
use DBI;
#use AnyEvent;

$|++;
my $lib = PublicLib->new();
my $cfg_file = 'server config.ini';
my $cfg = $lib->load_config($cfg_file);
#say Dumper $cfg;

my %client_table;
my @alarm_id;
my $timer = 60*5;

POE::Component::Server::TCP->new(
  Alias => $cfg->{General}->{server_name} ||= "server",
  Address => "localhost",
  Port => $cfg->{General}->{port} ||= 12345,
  ClientFilter => ["POE::Filter::Reference", "YAML"],
  #ClientFilter => "POE::Filter::Reference",

  ClientConnected => \&ClientConnected,
  ClientInput => \&ClientInput,
  ClientDisconnected => \&ClientDisconnected,
  InlineStates =>{
      time_to_check => \&time_to_check,
      client_time_out => \&client_time_out,
      send_warning_email => \&send_warning_email,
      update_rrd_file => \&update_rrd_file,
      create_graph => \&create_graph,
      write_db => \&write_db,
#     write_log => \&write_log,

  }
);
$poe_kernel->run();

sub ClientConnected
{
    #一上线就发送检查信息

    my ($kernel,$heap) = @_[KERNEL,HEAP];
    say 'ClientConnected';
    say "id is:",$heap->{client}->ID;
    say 'ip:',$heap->{remote_ip};
    write_log('Log',"[$heap->{remote_ip}]Connected");
    my $client_id = $heap->{client}->ID;
    for my $client_node (keys %$cfg)
    {
        if (exists $cfg->{$client_node}->{host} and $cfg->{$client_node}->{host} eq $heap->{remote_ip})
        {
            $client_table{$client_id}{client} = $heap->{client};
            $client_table{$client_id}{host} = $heap->{remote_ip};
            $client_table{$client_id}{alias} = $cfg->{$client_node}->{alias};
            $client_table{$client_id}{check_item} = $cfg->{$client_node}->{check_item};
            $client_table{$client_id}{rrd_file} = $cfg->{$client_node}->{RRDfile};
            $client_table{$client_id}{status} = 0;#1=>down,0=>up

            my %check_items =map{$_,1} split / /,$cfg->{$client_node}->{check_item};
            $heap->{client}->put(\%check_items); #TODO

#            send_msg(\%check_items);

            $client_table{$client_id}{last_check_time} = time;    #record time

#            say 'check_items are follow:';

            write_log('Log',"[$heap->{remote_ip}]check for ".$client_table{$client_id}{check_item});
#            say Dumper \%check_items;

        }
    }
    my $task_id = $kernel->delay_set('time_to_check',$timer,$client_id);#定时执行

    #may be has problem here

    push @alarm_id,$kernel->alarm_set('client_time_out',time()+$cfg->{General}->{timeout},$client_id,$task_id );
}

sub ClientInput
{
    my ($kernel,$heap,$yaml) = @_[KERNEL,HEAP,ARG0];
    $kernel->alarm_remove(shift @alarm_id);
#    print Dump $yaml;

    my $client_alias = delete $yaml->{alias};
    unless (defined $client_alias)
    {
        write_log('Fatal',"Client Alias is undef");
    }
    say 'client alias is:',$client_alias;
    for my $item (sort keys %$yaml)
    {
        next unless exists $cfg->{$client_alias}->{$item};
        say "$item check?";
        my ($operator,$threshold_value) = ($cfg->{$client_alias}->{$item})=~/^([<>=]+)(\d+)|unlimit$/;
        my %operator_table =(
            '>=' => sub{$_[0] >= $_[1];},
            '<=' => sub{$_[0] <= $_[1];},
            unlimit => sub{0},
            );
        my $is_failed = $operator_table{$operator}->($yaml->{$item},$threshold_value);
        if ($is_failed)
        {
            say "$item over:",$yaml->{$item};
            write_log('Log',"[$heap->{remote_ip}]$item overload:$yaml->{$item}");
            set_client_down_flag($heap->{client}->ID);
            if (exists $cfg->{$client_alias}->{repair})
            {
                for (split / /,$cfg->{ $client_alias }->{repair})
                {
                    next unless $_ eq $item;
                    my %check_item = map{$_ => 1} qw($_ repair);
                    $heap->{client}->put(\%check_item);
                    write_log('Log',"[$heap->{remote_ip}]system try to repair");
                }
            }
            $kernel->yield('send_warning_email',$item,$yaml->{$item});
            write_log('Log',"[$heap->{remote_ip}]warning mail send:$item");#不应该在这里写日志

            #TODO:设置报警原因，填写邮件模板，发邮件，DC

        }
        else
        {
            say "$item normal:",$yaml->{$item};
            #TODO:设计数据结构，为写数据库准备。DC

        }
    }    
    $kernel->yield('update_rrd_file',$client_alias,$yaml);
    #$kernel->yield('write_db',$client_alias,$yaml);

}

sub time_to_check
{
    my ($kernel,$heap,$client_id) = @_[KERNEL,HEAP,ARG0];
    say 'time_to_check';
#    $client_table{$client_id}{client}->put($client_table{$client_id}{check_item});

    if (exists $client_table{$client_id})
    {
        say "id is:$client_id";
        my %check_items =map{$_,1} split / /,$client_table{$client_id}{check_item};
        $client_table{$client_id}{client}->put(\%check_items) if exists $client_table{$client_id}{client}; #error here

        write_log('Log',"[$client_table{$client_id}{host}]check for ".$client_table{$client_id}{check_item});
        $client_table{$client_id}{last_check_time} = time;
        my $task_id = $kernel->delay_set('time_to_check',$timer,$client_id);#定时执行

        $client_table{$client_id}{task_id} = $task_id;
        push @alarm_id,$kernel->alarm_set('client_time_out',time()+$cfg->{General}->{timeout},$client_id,$task_id);
    }
    
}

sub client_time_out
{
    my ($kernel,$heap,$client_id,$task_id) = @_[KERNEL,HEAP,ARG0,ARG1];
    say 'client_time_out';
    $kernel->alarm_remove($task_id);
    if (exists $client_table{$client_id})
    {
        set_client_down_flag($client_id);
        write_log('Log',"[$client_table{$client_id}{host}]client time out");
    }
    $kernel->yield('send_warning_email','ClientTimeOut','down');
    write_log('Log',"[$client_table{$client_id}{host}]warning mail send:ClientTimeOut"); #不应该在这里写日志

    #TODO:set warning etc    

}

sub update_rrd_file
{
    my ($kernel,$heap,$client_alias,$yaml) = @_[KERNEL,HEAP,ARG0,ARG1];
    say "update rrd datafile";
    my $rrd = RRDTool::OO->new(file => $cfg->{$client_alias}->{RRDfile} ) || return;
    #尝试所有数据绘在同一张图上

    #构造一个hash

    my %value_structure = map{$_,$yaml->{$_}} get_ds($rrd); #这里要加判断

    eval{
        $rrd->update(
            time => time()+60*5,
        values => \%value_structure,
    );
  };
  if ($@)
  {
      write_log('Log',"[$client_alias]rrd update failed");
      return;
  }
  write_log('Log',"[$client_alias]rrd update success");
}

sub send_warning_email
{
    my ($kernel,$heap,$item,$value) = @_[KERNEL,HEAP,ARG0,ARG1];
    say "send mail ok";
    my @to = split / /,$cfg->{MailServer}->{to};
    my $subject = '监控项目'.$item.'异常';
    my $msg = '当前值:'.$value;
    print "$subject;$msg\n";
#    my $sender = new Mail::Sender();

#    my @protocols = $sender->QueryAuthProtocols(); #查询服务器支持的认证方式

#    say "QueryAuthProtocols:@protocols";

#    if ($sender->MailMsg({

#     smtp => $cfg->{MailServer}->{smtp},

#     from => $cfg->{MailServer}->{from},

#     to => \@to,

#     subject => $subject, #主题

#     msg => $msg, #内容

#     auth => 'LOGIN', #smtp的验证方式    

#     authid =>$cfg->{MailServer}->{user}, #user

#     authpwd => $cfg->{MailServer}->{passwd}, #pwd

#     }) < 0) {

#     warn "$Mail::Sender::Error\n";

#     }

#    else

#    {

#     print "send mail OK.\n";

#    }

#    POE::Component::Client::SMTP->send(

# # Email related parameters

# From => $cfg->{MailServer}->{from} || 'Admin',

# To => \@to,

# Body => \$email_body, # here's where your message is stored use Email::MIME

# Server => $cfg->{MailServer}->{smtp},

# # POE related parameters

# Alias => 'smtp',

# SMTP_Success => 'send_success',

# SMTP_Failure => 'send_failure',

# );

}

sub write_db
{
    my ($kernel,$heap,$client_alias,$yaml) = @_[KERNEL,HEAP,ARG0,ARG1];
    
}

sub write_log
{
#    my ($kernel,$heap,$type,$msg) = @_[KERNEL,HEAP,ARG0,ARG1];
    my ($type,$msg) = @_;
    my $logger = Log::Log->new(file => $cfg->{General}->{log},size => $cfg->{General}->{log_size});    
    my %method =(
        Log => sub{$logger->Log(shift);},
        Fatal => sub{$logger->Fatal(shift);}
    );
    $method{$type}->($msg);
    return 1;
}

sub ClientDisconnected
{
    my ($kernel,$heap) = @_[KERNEL,HEAP];
    $kernel->alarm_remove(shift @alarm_id); #warn!!

    $kernel->alarm_remove($client_table{$heap->{client}->ID}->{task_id});
    say "ClientDisconnected:delete client ok";
    write_log('Log',"[$heap->{remote_ip}]Disconnected");
    delete $client_table{$heap->{client}->ID};
    delete $heap->{client};
}

sub set_client_down_flag
{
    #超时或者over时执行

    my $client_id =shift;
    $client_table{$client_id}{status} = 1 ;#1=>down,0=>up

    my $host = $client_table{$client_id}{host};
    my $alias = $client_table{$client_id}{alias};    
}

sub get_ds
{
    return keys %{ shift->info()->{ds}};
}

客户端如下：

#BY Weigun http://blog.chinaunix.net/u2/70443/

use strict;
use warnings;
use 5.010;
use lib 'D:/test-area/desgin/Module';
use PublicLib;
use YAML qw(LoadFile Dump);
use Config::IniFiles;
use AnyEvent;
use POE;
use POE::Filter::Reference;
use POE::Wheel::Run::Win32;
use POE::Component::RunScript;
use POE::Component::Client::TCP;
use Data::Dumper;
use Log::Log;
$|++;
my $pid_file = 'client.pid';
my $lib = PublicLib->new();
exit if $lib->is_running($pid_file);
my $cfg = init();
POE::Component::Client::TCP->new(
     Alias => $cfg->{General}->{'alias'},
     RemoteAddress => "localhost",
     RemotePort => $cfg->{General}->{port},
     ConnectTimeout => $cfg->{General}->{'timeout'},
     Filter => ["POE::Filter::Reference", "YAML"],
#     Filter => "POE::Filter::Reference",

     Connected         => sub{write_log('Log',"Connect to server success");},
     ConnectError     => \&connect_err,
     ServerInput     => \&collect_data,
     ServerError        => \&server_err,
     InlineStates =>{
         run_scripts => \&run_scripts,
            are_scripts_finish => \&are_scripts_finish,
            send_data => \&send_data,
            repair_service => \&repair_service,
        }
);
POE::Kernel->run();
exit(0);

sub init
{
    my $cfg_file = 'config.ini';
    my $cfg = $lib->load_config($cfg_file);
    say "init";
#    $lib->create_pid_file($pid_file);

    return $cfg;
}

sub collect_data
{
    my ($kernel, $collect_opt) = @_[KERNEL, ARG0];
#    say Dumper $collect_opt;

    #TODO:区别收集信号与修复信号

    my @items = keys %$collect_opt;
    if (defined delete $collect_opt->{repair})
    {
        say "repair_service";    
        write_log('Log',"repair:@items");
        $kernel->yield('repair_service',$collect_opt);
    }
    else
    {
        say "collect_data";
        write_log('Log',"collect for:@items");
        $kernel->yield('run_scripts',$collect_opt);
    }
#    say "collect_data end";

}

sub run_scripts
{
    my ($kernel, $heap,$h_opt) = @_[KERNEL, HEAP,ARG0];
    $heap->{result} = [];
    my %scripts = map{$_ => $cfg->{Item}->{$_} if exists $h_opt->{$_}} keys %{$cfg->{Item}};
#    say Dumper \%scripts;

    #删除yaml？

    eval{POE::Component::RunScript->spawn(script_dir => $cfg->{General}->{'script_dir'} || 'D:\test-area\desgin\scripts', #无视

                                     parent_session => $cfg->{General}->{'alias'},
                                     scripts => \%scripts,
        );
    };
#    write_file($@) if $@;

    write_log('Log',"run scripts error:$@") if $@;
     $kernel->yield('are_scripts_finish',scalar keys %scripts,10);
#     $kernel->delay_set('are_scripts_finish',10,scalar keys %scripts,10);

     say "run_scripts end";
}

sub are_scripts_finish
{
    #TODO:limit try times

    my ($kernel, $heap,$item_num,$max_try) = @_[KERNEL, HEAP,ARG0,ARG1];
    if ($max_try == 0 or scalar @{$heap->{result}} == $item_num)
    {
         write_file($heap->{result});
         $kernel->yield ('send_data');
         write_log('Log',"run scripts end");
    }
    else
    {
        $kernel->delay_set ('are_scripts_finish',1,$item_num,--$max_try);
        say "scripts not finish:total_scripts is $item_num,real is ",scalar @{$heap->{result}};
        say $_ for (@{$heap->{result}})
    }
}

sub send_data
{
    my ($kernel, $heap) = @_[KERNEL, HEAP];
    my $r_struct = {};
    for (@{$heap->{result}})
    {
        chop;
#        next if /None/;

        if (/None/)
        {
            say "some script can't run,return None";
            write_log('Log',"some script can't run,return None");
            next;
        }
        my $yaml = LoadFile($_);
        $r_struct->{$_} = $yaml->{$_} for (keys %$yaml);
    }
    $heap->{server}->put($r_struct);
    undef @{$heap->{result}};
    my @items = sort keys %$r_struct;
    write_log('Log',"send:@items");
    say "all data send";
}

sub repair_service
{
    my ($kernel, $heap,$r_repair) = @_[KERNEL, HEAP,ARG0];
    my %scripts =map{$_ => $cfg->{Repair}->{$_}} keys %$r_repair; #需要检查?

    #删除yaml？

    eval{POE::Component::RunScript->spawn(script_dir => $cfg->{General}->{'script_dir'} || 'D:\test-area\desgin\scripts', #无视

                                     #parent_session => $cfg->{General}->{'alias'},

                                     scripts => \%scripts,
        );
    write_log('Log',"some repair scripts can't run") if $@;
    write_log('Log',"repair completed");
    };
}

sub connect_err
{
    my ($operation, $error_number, $error_string) = @_[ARG0..ARG2];
    say "ConnectError:tmeOut?";
  warn "$operation error $error_number occurred: $error_string";
  say "try to reconnect";
  write_log('Log',"$operation error $error_number occurred: $error_string");
  $_[KERNEL]->delay('reconnect',$cfg->{General}->{'reconnect_interval'});
}

sub server_err
{
    #TODO

    my ($kernel, $heap) = @_[KERNEL, HEAP];
    say "ServerError,try to reconnect";
    $kernel->delay('reconnect',$cfg->{General}->{'reconnect_interval'});
    write_log('Log',"ServerError");
    #unlink $pid_file || die "can't del $pid_file";

}

sub write_file
{
    if (ref $_[0] eq 'ARRAY')
    {
        say 'result is:';
        say $_ for (@{$_[0]});
        return;
    }
    say $_[0];
}

sub write_log
{
#    my ($kernel,$heap,$type,$msg) = @_[KERNEL,HEAP,ARG0,ARG1];
    my ($type,$msg) = @_;
    my $logger = Log::Log->new(file => $cfg->{General}->{log},size => $cfg->{General}->{log_size});    
    my %method =(
        Log => sub{$logger->Log(shift);},
        Fatal => sub{$logger->Fatal(shift);}
    );
    $method{$type}->($msg);
    return 1;
}