persistant_worker.pl

   1 #!/usr/bin/perl
   2
   3 package Cog; # persistant Gearman Worker - see the pun?
   4
   5 use Moose;
   6 use JSON::XS qw();
   7 extends 'Gearman::Driver::Worker::Base';
   8
   9 use lib 'lib';
  10 with 'APKPM::Gearman';
  11 with 'APKPM::Gearman::Client';
  12
  13 sub prefix { '' }
  14
  15 package main;
  16
  17 use Gearman::Driver;
  18 use Data::Dump qw(dump);
  19
  20 use Regexp::Common qw(net);
  21 use Redis;
  22
  23 use lib 'lib';
  24 use H1::ZTEDSLAM;
  25 use H1::ZTEMSAN;
  26
  27 my ($variant,$max_processes) = @ARGV;
  28 $variant ||= 'ZTEDSLAM';
  29 $max_processes ||= 1;
  30 warn "variant: $variant max_processes: $max_processes\n";
  31
  32 my $driver = Gearman::Driver->new(
  33         server   => 'localhost:4730',
  34         interval => 60,
  35         loglevel   => 'DEBUG',
  36         logfile    => "log/$variant.log",
  37 ) || die $!;
  38
  39 my $w1 = 'Cog';
  40 my $worker = $w1->new();
  41
  42
  43 my $crm = $worker->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
  44
  45
  46 my $poll;
  47
  48 foreach my $user ( @$crm ) {
  49         next unless $user->{IP_UREDAJA} =~ /$RE{net}{IPv4}/;
  50         $poll->{ $user->{IP_UREDAJA} }->{ $user->{SHELF_SLOT_PORT} }++;
  51 }
  52
  53 warn "# poll = ", dump($poll);
  54
  55 foreach my $ip ( keys %$poll ) {
  56
  57         my $method = $variant . '_' . $ip;
  58
  59         my $module = 'H1::' . $variant;
  60         my $zte = $module->new( ip => $ip );
  61
  62         $driver->add_job({
  63                 max_processes => $max_processes,
  64                 min_processes => 1,
  65                 name            => $method,
  66                 worker          => $worker,
  67                 methods    => [
  68                         {
  69                                 decode => 'd_json',
  70 #                               encode => 'e_json',
  71                                 name   => $method,
  72                                 body   => sub {
  73
  74 # XXX -- worker body
  75
  76                 my ( $self, $job, $crm ) = @_;
  77
  78                 my $redis = Redis->new;
  79
  80                 if ( exists $crm->{logout} ) {
  81                         $zte->logout;
  82                         $redis->sadd("$variant.$ip.logout", $self->datetime_now);
  83                         return "logout ok"; # XXX return scalar!
  84                 }
  85
  86                 my $port     = $crm->{SHELF_SLOT_PORT} || die "no SHELF_SLOT_PORT";
  87                 my $username = $crm->{USERNAME}        || die "no USERNAME";
  88
  89                 my $hash;
  90                 eval { $hash = $zte->hash( $port ) };
  91
  92                 if ( $@ ) {
  93                         $redis->sadd("$variant.$ip.error" => $@);
  94                         return "error: $@";
  95                 } elsif ( ! $hash ) {
  96                         $redis->sadd("$variant.$ip.empty" => $port);
  97                         return "empty";
  98                 } else {
  99                         $redis->sadd("$variant.$ip.ok" => $port);
 100                 }
 101
 102                 $self->do_background_json( 'Store_insert', {
 103                         _table => 'dslam_h',
 104                         ip => $crm->{IP_MANAGEMENT}, # FIXME IP_UREDAJA ?
 105                         username => $username,
 106                         timestamp => $self->datetime_now,
 107                         h => $self->to_hstore($hash),
 108                 });
 109
 110                 return "ok $username $ip $port"; # body end
 111
 112 # XXX -- worker body
 113                                 },
 114                         },
 115                 ]
 116         });
 117
 118         warn "$method added\n";
 119
 120 }
 121
 122 $driver->add_job({
 123         max_processes => 1,
 124         min_processes => 1,
 125         worker          => $worker,
 126         name            => "poll_$variant",
 127         methods    => [ {
 128                 name            => "poll_$variant",
 129                 body   => sub {
 130
 131                 my ( $self, $job, $workload ) = @_;
 132
 133                 my $crm = $self->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
 134
 135                 warn "# crm $crm";
 136
 137                 my $gc = Gearman::Client->new;
 138                 $gc->job_servers( $self->server || 'localhost:4730' );
 139                 my $taskset = $gc->new_task_set;
 140                 my $seen;
 141
 142                 my $redis = Redis->new;
 143                 $redis->del( $_ ) foreach $redis->keys("$variant.*");
 144
 145                 $redis->set( "$variant.poll.start" => $self->datetime_now );
 146
 147                 foreach my $user ( @$crm ) {
 148                         # required CRM fields for worker
 149                         my $ip   = $user->{IP_UREDAJA}      || next;
 150                         my $port = $user->{SHELF_SLOT_PORT} || next;
 151                         my $username = $user->{USERNAME}    || next;
 152
 153                         if ( $ip !~ /$RE{net}{IPv4}/ ) {
 154                                 $redis->sadd("$variant.invalid.IP_UREDAJA" => $ip);
 155                                 next;
 156                         } elsif ( $port !~ m{\d+(/\d+)+} ) {
 157                                 $redis->sadd("$variant.$ip.invalid.SHELF_SLOT_PORT" => $port);
 158                                 next;
 159                         } elsif ( ! $username ) {
 160                                 $redis->sadd("$variant.$ip.invalid.USERNAME" => $port);
 161                                 next;
 162                         } elsif ( $seen->{ $ip }->{ $port }++ ) {
 163                                 $redis->sadd("$variant.$ip.invalid.duplicate" => $port);
 164                                 next;
 165                         }
 166
 167                         $redis->incr("$variant.poll.queued");
 168                         $redis->sadd("$variant.$ip.queued" => $port);
 169
 170                         my $name = $variant . '_' . $ip;
 171                         $taskset->add_task( $name, $self->e_json( $user ), {
 172                                 on_complete => sub { $redis->sadd("$variant.$ip.complete" => $port) },
 173                                 on_fail     => sub { $redis->sadd("$variant.$ip.failed" => $port) },
 174                         } )
 175                 }
 176
 177                 warn "# queue logouts";
 178                 foreach my $ip ( keys %$seen ) {
 179                         $taskset->add_task( $variant . '_' . $ip, $self->e_json( { logout => 1 } ) );
 180                 }
 181
 182                 warn "# wait";
 183                 $taskset->wait;
 184
 185                 $redis->set( "$variant.poll.finish" => $self->datetime_now );
 186                 warn "# seen ", dump($seen);
 187
 188                 my $ips = scalar keys %$seen;
 189                 return "polled $ips IPs"; # body end
 190                 }
 191         } ],
 192 });
 193
 194 $driver->run;
 195