用 Redis 做分布式 DNS/HTTP 检测汇总系统

#!/usr/bin/perl
use Modern::Perl;
use AnyEvent;
use AnyEvent::Redis::RipeRedis;
use Storable qw/freeze thaw/;
use YAML::Syck;
use utf8;
my $area = $ARGV[0];
my $domain = 'fmn.xnimg.cn';
my $master = '10.4.1.21';
my $cv     = AnyEvent->condvar;
my $redis  = AnyEvent::Redis::RipeRedis->new(
    host     => $master,
    port     => 6379,
    encoding => 'utf8',
);
my $dnslist = LoadFile("DNS.yml");
for my $isp ( sort keys %$dnslist ) {
    if ( defined $area ) {
        next unless defined $dnslist->{$isp}->{$area};
        say $area, $isp, join ", ", @{ $dnslist->{$isp}->{$area} };
        my $data = freeze({ domain => $domain, dnslist => $dnslist->{$isp}->{$area} });
        $redis->publish( 'task', $data );
    } else {
        for my $list ( sort keys %{ $dnslist->{$isp} } ) {
            my $data = freeze({ domain => $domain, dnslist => $dnslist->{$isp}->{$list} });
            $cv->begin;
            $redis->publish( 'task', $data );
        }
    }
}
$redis->subscribe(
    qw( report ),
    {
        on_done => sub {
            my $ch_name  = shift;
            my $subs_num = shift;
            print "Subscribed: $ch_name. Active: $subs_numn";
        },
        on_message => sub {
            my $ch_name = shift;
            my $msg     = thaw( shift );
            printf "%s A %s @%s in %s got %s length %sn", $domain, $msg->{ip}, $msg->{dns}, $msg->{local}, $msg->{status}, $msg->{len};
            $cv->end;
        },
        on_error => sub {
            print @_;
        },
    }
);
$cv->recv;

分布在各地的客户端脚本如下：

#!/usr/bin/perl
use Modern::Perl;
use AnyEvent;
use AnyEvent::Socket;
use AnyEvent::DNS;
use AnyEvent::Redis::RipeRedis;
use AnyEvent::HTTP;
use Storable qw/freeze thaw/;
use Digest::MD5 qw/md5_hex/;
use utf8;
my $master = '10.4.1.21';
my $local  = '192.168.0.2';
my $cv     = AnyEvent->condvar;
my $redisr = AnyEvent::Redis::RipeRedis->new(
    host          => $master,
    port          => 6379,
    encoding      => 'utf8',
);
my $redisp = AnyEvent::Redis::RipeRedis->new(
    host          => $master,
    port          => 6379,
    encoding      => 'utf8',
);
$redisr->subscribe(
    'task',
    {
        on_done => sub {
            my $ch_name  = shift;
            my $subs_num = shift;
            print "Subscribed: $ch_name. Active: $subs_numn";
        },
        on_message => sub {
            my $ch_name = shift;
            my $msg     = thaw(shift);
            for my $dns ( @{ $msg->{dnslist} } ) {
                resolv( $dns, $msg->{domain} );
            }
        },
        on_error => sub {
            my $err_msg  = shift;
            my $err_code = shift;
            print "Error: ($err_code) $err_msgn";
        },
    }
);
$cv->recv;
sub resolv {
    my ( $dns, $domain ) = @_;
    return unless $dns =~ m/^d+/;
    my $resolver =
      AnyEvent::DNS->new( server => [ AnyEvent::Socket::parse_address $dns ], );
    $resolver->resolve(
        "$domain" => 'a',
        sub {
            httptest($dns, $domain, $_->[-1]) for @_;
        }
    );
}
sub httptest {
    my ($dns, $domain, $ip) = @_;
    my $url = "http://$domain/10k.html";
    my $begin = time;
    http_get $url, proxy => [$ip, 80], want_body_handle => 1, sub {
        my ($hdl, $hdr) = @_;
        my ($port, $peer) = AnyEvent::Socket::unpack_sockaddr getpeername $hdl->{'fh'};
        my $data = freeze( { dns => $dns, status => $hdr->{Status}, local => $local, ip => $peer, len => $hdr->{'content-length'} } );
        $redisp->publish('report', $data);
    };
}