lib/APKPM/Poll.pm

   1 package APKPM::Poll;
   2
   3 use base qw(Gearman::Driver::Worker);
   4 use Moose;
   5 use Time::HiRes qw(time);
   6 use Data::Dump qw(dump);
   7 use Redis;
   8
   9 with 'APKPM::Gearman::Client';
  10 with 'APKPM::Gearman';
  11
  12 sub prefix { '' }
  13
  14 sub poll_prefix : Job : Encode(e_json) {
  15         my ( $self, $job, $workload ) = @_;
  16
  17         my $redis = Redis->new;
  18         $redis->del( $_ ) foreach $redis->keys('poll.*');
  19
  20         my $start = $self->datetime_now;
  21
  22         $redis->set( 'poll.start' => $start );
  23
  24         my $entries = $self->do( 'LDAP_search' => "(&(cn=$workload*)(dhcpStatements=*))" );
  25
  26         my $ip_username;
  27         foreach my $entry ( @$entries ) {
  28                 $ip_username->{ $entry->{cn} } = $1 if $entry->{dhcpStatements} =~ m/fixed-address\s+(\S+)/;
  29         }
  30
  31         my $taskset = $self->gc->new_task_set;
  32         my $results;
  33
  34         $taskset->add_task('poll_' . $_, "start", {
  35                 on_complete => sub { push @$results, ${$_[0]} }
  36         }) foreach ( 'ZTEDSLAM', 'ZTEMSAN' );
  37
  38         while (my ($username,$ip) = each %$ip_username) {
  39                 $taskset->add_task('poll_ip_username', "$ip $username", {
  40                         on_complete => sub { push @$results, ${$_[0]} }
  41                 });
  42                 $redis->sadd('poll.queued' => $ip);
  43         }
  44
  45         warn "# wait";
  46         $taskset->wait;
  47
  48         my $finish = $self->datetime_now;
  49         $redis->set( 'poll.finish' => $finish );
  50
  51         my $poll;
  52         foreach my $k ( $redis->keys('poll.*') ) {
  53                 my $n = $k;
  54                 $n =~ s/^poll\.//;
  55                 $n =~ s/\./_/g;
  56                 $poll->{$n} = eval { $redis->scard($k) } || $redis->get($k);
  57         }
  58
  59         warn dump $poll;
  60         $self->do_background_json('Store_insert', { _table => 'poll', %$poll });
  61
  62         return { ldap => $ip_username, results => $results, poll => $poll };
  63 }
  64
  65 sub poll_ip_username : Job : Decode(d_array) : Encode(e_json) {
  66         my ( $self, $job, $workload ) = @_;
  67
  68         my ( $ip, $username ) = @$workload;
  69
  70         return { error => "invalid workload", expected => "ip username" } unless $ip && $username;
  71         my $ping = $self->do('ping', $ip);
  72
  73         my $redis = Redis->new;
  74
  75         if ( exists $ping->{error} ) {
  76 warn "XXX error: $ip";
  77                 $redis->sadd( 'poll.ping.error' => $ip );
  78                 return $ping;
  79         } else {
  80                 $redis->sadd( 'poll.ping.ok' => $ip );
  81         }
  82
  83         $ping->{username} = $username;
  84         $ping->{timestamp} = $self->datetime_now;
  85         $self->do_background_json( 'Store_insert', {
  86                  _table => 'ping',
  87                 username => $username,
  88                 timestamp => $self->datetime_now,
  89                 %$ping
  90         });
  91
  92         $self->do_background( 'Davolink_info', "$ip $username adsl" );
  93
  94         return $ping;
  95 }
  96
  97 1;