lib/APKPM/CPE.pm

   1 package APKPM::CPE;
   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 use Net::Ping;
   9
  10 use lib 'lib';
  11 use H1::EasyGateway;
  12
  13 with 'APKPM::Gearman::Client';
  14 with 'APKPM::Gearman';
  15
  16 sub prefix { 'CPE_' }
  17
  18 sub poll : Job : Encode(e_json) {
  19         my ( $self, $job, $workload ) = @_;
  20
  21         my $redis = Redis->new;
  22         $redis->del( $_ ) foreach $redis->keys('CPE.*');
  23
  24         my $start = $self->datetime_now;
  25
  26         $redis->set( 'CPE.start' => $start );
  27
  28         my $entries = $self->do( 'LDAP_search' => "(&(cn=$workload*)(dhcpStatements=*))" );
  29
  30         my $taskset = $self->gc->new_task_set;
  31         my $results;
  32
  33         foreach my $entry ( @$entries ) {
  34
  35                 my $username = $entry->{cn} || die "no cn in ", dump($entry);
  36
  37                 if ( $entry->{dhcpStatements} !~ m/fixed-address\s+(\S+)/ ) {
  38                         $redis->sadd('CPE.error.no-ip' => $username);
  39                         next;
  40                 }
  41                 my $ip = $1;
  42
  43                 if ( $entry->{dhcpOption} !~ m/vendor-class-identifier\s\"([^"]+)\"/ ) {
  44                         $redis->sadd('CPE.error.vendor' => $username);
  45                         next;
  46                 }
  47
  48                 my $vendor = $1;
  49
  50                 $taskset->add_task('CPE_ping', "$ip $username $vendor", {
  51                         on_complete => sub { push @$results, ${$_[0]} }
  52                 });
  53                 $redis->sadd('CPE.queued' => $ip);
  54
  55                 $vendor =~ s/\s+/_/g; # sadd dies on spaces in keys
  56                 $redis->sadd("CPE.vendor.$vendor" => $ip); # FIXME
  57         }
  58
  59         warn "# wait";
  60         $taskset->wait;
  61
  62         my $finish = $self->datetime_now;
  63         $redis->set( 'CPE.finish' => $finish );
  64
  65         my $poll;
  66         foreach my $k ( $redis->keys('CPE.*') ) {
  67                 my $n = $k;
  68                 $n =~ s/^CPE\.//;
  69                 $n =~ s/\./_/g;
  70                 $poll->{$n} = eval { $redis->scard($k) } || $redis->get($k);
  71         }
  72
  73         warn "# poll = ",dump $poll;
  74         $self->do_background_json('Store_insert', { _table => 'CPE_poll', %$poll });
  75
  76         warn "# results = ", dump $results;
  77         return $poll;
  78 }
  79
  80 sub ping : Job : Decode(d_array) : Encode(e_json) {
  81         my ( $self, $job, $workload ) = @_;
  82
  83         my ( $ip, $username, $vendor ) = @$workload;
  84
  85         my $redis = Redis->new;
  86
  87         return { error => "invalid workload", expected => "ip username" } unless $ip && $username;
  88
  89         my $p = Net::Ping->new;
  90         $p->hires;
  91
  92         my ( $ok, $rtt, $ping_ip ) = $p->ping( $ip );
  93
  94         if ( $ok ) {
  95                 $redis->sadd( 'CPE.ping.ok' => $ip );
  96         } else {
  97                 $redis->sadd( 'CPE.ping.error' => $ip );
  98                 return { error => "ping $ip" };
  99         }
 100
 101         $self->do_background_json( 'Store_insert', {
 102                  _table => 'ping',
 103                 username => $username,
 104                 timestamp => $self->datetime_now,
 105                 ip => $ping_ip,
 106                 rtt => $rtt,
 107         });
 108
 109         if ( $vendor =~ m/SAMSUNG/ ) {
 110                 $redis->sadd( 'CPE.Davolink.queued' => $username );
 111                 $self->do( 'Davolink_info', "$ip $username adsl" );
 112         } elsif ( $vendor =~ m/zte/ ) {
 113                 $redis->sadd( 'CPE.EasyGateway.queued' => $ip );
 114
 115                 my $ret = eval { H1::EasyGateway->info( $ip ) };
 116
 117                 if ( $ret ) {
 118
 119                         $self->do_background_json( 'Store_insert', {
 120                                 _table => 'easygateway',
 121                                 ip => $ip,
 122                                 username => delete $ret->{customerName},
 123                                 timestamp => $self->datetime_now,
 124                                 h => $self->to_hstore( $ret ),
 125                         });
 126
 127                         $redis->sadd( 'CPE.EasyGateway.ok' => $ip );
 128
 129                 } else {
 130                         $redis->sadd( 'CPE.EasyGateway.error' => $ip );
 131                 }
 132
 133         } else {
 134                 $redis->sadd( 'CPE.skipped' => $username );
 135         }
 136
 137         return { ip => $ip, rtt => $rtt };
 138 }
 139
 140 1;