X-Git-Url: http://git.rot13.org/?a=blobdiff_plain;f=persistant_worker.pl;h=647b2bdf4646df332fad89917b981d3c52e849be;hb=516c0e6f30c2fb9e16eada973a35a46d7000ee3a;hp=e2db7d8781ec724bc4e5a3792121620a6c0ae665;hpb=9b277d2422f992550d8b612f8f9204ba7210a6af;p=APKPM.git

diff --git a/persistant_worker.pl b/persistant_worker.pl
index e2db7d8..647b2bd 100755
--- a/persistant_worker.pl
+++ b/persistant_worker.pl
@@ -1,5 +1,9 @@
 #!/usr/bin/perl
 
+BEGIN {
+	$SIG{'__WARN__'} = sub { warn $_[0] if $_[0] =~ m/^(#+)/ && length($1) <= $ENV{DEBUG} }
+}
+
 package Cog; # persistant Gearman Worker - see the pun?
 
 use Moose;
@@ -18,22 +22,31 @@ use Gearman::Driver;
 use Data::Dump qw(dump);
 
 use Regexp::Common qw(net);
+use Redis;
 
 use lib 'lib';
 use H1::ZTEDSLAM;
+use H1::ZTEMSAN;
+use APKPM::Model;
+
+my ($variant,$max_processes) = @ARGV;
+$variant ||= 'ZTEDSLAM';
+$max_processes ||= 1;
+warn "variant: $variant max_processes: $max_processes\n";
 
 my $driver = Gearman::Driver->new(
 	server   => 'localhost:4730',
-	interval => 60,
+	interval => 15,
 	loglevel   => 'DEBUG',
-	logfile    => 'log/persistant.log',
+	logfile    => "log/$variant.log",
 ) || die $!;
 
 my $w1 = 'Cog';
 my $worker = $w1->new();
 
-warn "CRM_search";
-my $crm = $worker->do( 'CRM_search' => "TIP_UREDJAJA:ZTEDSLAM" );
+
+my $crm = $worker->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
+
 
 my $poll;
 
@@ -46,82 +59,161 @@ warn "# poll = ", dump($poll);
 
 foreach my $ip ( keys %$poll ) {
 
-	my $method = "ZTEDSLAM_$ip";
+	my $method = $variant . '_' . $ip;
+
+	my $module = 'H1::' . $variant;
+	my $zte = $module->new( ip => $ip );
 
-	my $zte = H1::ZTEDSLAM->new( ip => $ip );
+	my $ports = keys %{ $poll->{$ip} };
+	my $max_procs = ( int($ports/50) ) + 1; # one process for 50 ports
+	$max_procs = $max_processes if $max_processes < $max_procs; # hard-limit for maximum ports in parallel
 
 	$driver->add_job({
-		max_processes => 1, # FIXME increase?
-		min_processes => 1,
+		max_processes => $max_procs,
+		min_processes => 0,
 		name		=> $method,
 		worker		=> $worker,
 		methods	   => [
 			{
 				decode => 'd_json',
-				encode => 'e_json',
+#				encode => 'e_json',
 				name   => $method,
 				body   => sub {
 
+# XXX -- worker body
+
 		my ( $self, $job, $crm ) = @_;
-		my $hash;
-		eval { $hash = $zte->hash( $crm->{SHELF_SLOT_PORT} ) };
 
-		return { error => $@ } if $@;
+		my $redis = Redis->new;
+
+		if ( exists $crm->{logout} ) {
+			$zte->logout;
+			$redis->sadd("$variant.$ip.logout", $self->datetime_now);
+			return "logout ok"; # XXX return scalar!
+		}
+
+		my $port     = $crm->{SHELF_SLOT_PORT} || die "no SHELF_SLOT_PORT";
+		my $username = $crm->{USERNAME}        || die "no USERNAME";
+
+		my $hash;
+		eval { $hash = $zte->hash( $port ) };
+
+		if ( $@ ) {
+			$redis->sadd("$variant.$ip.error" => $@);
+			return "error: $@";
+		} elsif ( ! $hash ) {
+			$redis->sadd("$variant.$ip.empty" => $port);
+			return "empty";
+		} else {
+			$redis->sadd("$variant.$ip.ok" => $port);
+		}
 
 		$self->do_background_json( 'Store_insert', {
-			_table => 'dslam_h',
+			_table => 'dslam',
 			ip => $crm->{IP_MANAGEMENT}, # FIXME IP_UREDAJA ?
-			username => $crm->{USERNAME},
+			username => $username,
 			timestamp => $self->datetime_now,
+			variant => $variant,
 			h => $self->to_hstore($hash),
 		});
 
-#		return $hash; # body end
+		return "ok $username $ip $port"; # body end
 
+# XXX -- worker body
 				},
 			},
 		]
 	});
 
-	warn "$method added\n";
+	warn "$method for $ip with $ports ports using $max_procs processes\n";
 
 }
 
 $driver->add_job({
 	max_processes => 1,
-	min_processes => 1,
+	min_processes => 0,
 	worker		=> $worker,
+	name		=> $variant . '_poll',
 	methods	   => [ {
-		name		=> 'poll_ZTEDSLAM',
+		name   => $variant . '_poll',
+		encode => 'e_json',
 		body   => sub {
 
 		my ( $self, $job, $workload ) = @_;
 
-		my $crm = $self->do( 'CRM_search' => "TIP_UREDJAJA:ZTEDSLAM" );
+		my $crm = $self->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
 
 		warn "# crm $crm";
 
-		my $taskset = $worker->gc->new_task_set;
+		my $gc = Gearman::Client->new;
+		$gc->job_servers( $self->server || 'localhost:4730' );
+		my $taskset = $gc->new_task_set;
 		my $seen;
 
+		my $redis = Redis->new;
+		$redis->del( $_ ) foreach $redis->keys("$variant.*");
+
+		$redis->set( "$variant.poll.start" => $self->datetime_now );
+
 		foreach my $user ( @$crm ) {
-			my $ip = $user->{IP_UREDAJA};
-			next unless $ip =~ /$RE{net}{IPv4}/;
-			my $name = 'ZTEDSLAM_' . $ip;
+			# required CRM fields for worker
+			my $ip   = $user->{IP_UREDAJA}      || next;
+			my $port = $user->{SHELF_SLOT_PORT} || next;
+			my $username = $user->{USERNAME}    || next;
+
+			if ( $ip !~ /$RE{net}{IPv4}/ ) {
+				$redis->sadd("$variant.invalid.IP_UREDAJA" => $ip);
+				next;
+			} elsif ( $port !~ m{\d+(/\d+)+} ) {
+				$redis->sadd("$variant.$ip.invalid.SHELF_SLOT_PORT" => $port);
+				next;
+			} elsif ( ! $username ) {
+				$redis->sadd("$variant.$ip.invalid.USERNAME" => $port);
+				next;
+			} elsif ( $seen->{ $ip }->{ $port }++ ) {
+				$redis->sadd("$variant.$ip.invalid.duplicate" => $port);
+				next;
+			}
+
+			$redis->incr("$variant.poll.queued");
+			$redis->sadd("$variant.$ip.queued" => $port);
+
+			my $name = $variant . '_' . $ip;
 			$taskset->add_task( $name, $self->e_json( $user ), {
-				on_complete => sub { "# $name done", $/ }
+				on_complete => sub { $redis->sadd("$variant.$ip.complete" => $port) },
+				on_fail     => sub { $redis->sadd("$variant.$ip.failed" => $port) },
 			} )
-			if ! $seen->{ $user->{IP_UREDAJA} }->{ $user->{SHELF_SLOT_PORT} }++;
 		}
+
+		warn "# queue logouts";
+		foreach my $ip ( keys %$seen ) {
+			$taskset->add_task( $variant . '_' . $ip, $self->e_json( { logout => 1 } ) );
+		}
+
 		warn "# wait";
 		$taskset->wait;
 
-		warn "# seen ", dump($seen);
+		$redis->set( "$variant.poll.finish" => $self->datetime_now );
 
-		return; # body end
+		my $status = APKPM::Model->redis_status;
+		my $row = $status->{poll}->{$variant};
+		$row->{$_} = $status->{"$variant.poll.$_"} foreach ( 'start', 'finish' );
+		$row->{variant} = $variant;
+		$self->do_background_json('Store_insert', { _table => 'dslam_poll', %$row });
+		warn "# dslam_poll ",dump($row);
+
+		my $ips = scalar keys %$seen;
+		my $ports = 0;
+		$ports += scalar keys %{ $seen->{$_} } foreach keys %$seen;
+		warn "# finish $ips ips with $ports ports\n";
+		return { ips => $ips, ports => $ports, dslam_poll => $row }  ; # body end
 		}
 	} ],
 });
 
+open(my $pid, '>', "/tmp/apkpm.$variant.pid");
+print $pid "$$\n";
+close $pid;
+
 $driver->run;