refactor to use MCE::Loop and per-device nbtstat

Netdisco/lib/App/Netdisco/Core/Nbtstat.pm

@@ -5,7 +5,6 @@ use Dancer::Plugin::DBIC 'schema';
 
 use App::Netdisco::Util::Node 'check_mac';
 use NetAddr::IP::Lite ':lower';
-use Time::HiRes 'gettimeofday';
 use Net::NBName;
 
 use base 'Exporter';
@@ -36,8 +35,7 @@ Returns whether a node is answering netbios calls or not.
 =cut
 
 sub do_nbtstat {
-    my $host = shift;
+    my ($host, $now) = @_;
-
     my $ip = NetAddr::IP::Lite->new($host) or return;
 
     unless ( $ip->version() == 4 ) {
@@ -46,7 +44,6 @@ sub do_nbtstat {
     }
 
     my $nb = Net::NBName->new;
-
     my $ns = $nb->node_status( $ip->addr );
 
     # Check for NetBIOS Info
@@ -55,7 +52,7 @@ sub do_nbtstat {
     my $nbname = _filter_nbname( $ip->addr, $ns );
 
     if ($nbname) {
-        store_nbt($nbname);
+        store_nbt($nbname, $now);
     }
 
     return 1;
@@ -101,7 +98,7 @@ sub _filter_nbname {
 
         # Just assume it's the last MAC we saw this IP at.
         my $node_ip = schema('netdisco')->resultset('NodeIp')
-            ->single( { ip => $ip, active => \'true' } );
+            ->single( { ip => $ip, -bool => 'active' } );
 
         if ( !defined $node_ip ) {
             debug ' no MAC for %s returned by nbtstat or in DB', $ip;

Netdisco/lib/App/Netdisco/DB/ResultSet/NodeIp.pm

@@ -212,7 +212,7 @@ sub ip_version {
     die "ip_version input must be either 4 or 6\n"
         unless $version && ( $version == 4 || $version == 6 );
 
-    return $rs->search_rs( \[ 'family(ip) = ?', $version ] );
+    return $rs->search_rs( \[ 'family(me.ip) = ?', $version ] );
 }
 
 1;

Netdisco/lib/App/Netdisco/Daemon/Worker/Poller/Common.pm

@@ -88,80 +88,23 @@ sub _single_body {
   return job_done("Ended $job_type for ". $host->addr);
 }
 
-# _walk_nodes_body
-# Queue a job for all active nodes that have been seen in the last
-# configured days.
-# 
-sub _walk_nodes_body {
-  my ($self, $job_type) = @_;
-
-  my $action_method = $job_type .'_action';
-  my $job_action = $self->$action_method;
-
-  my $jobqueue = schema('netdisco')->resultset('Admin');
-  my $rs = schema('netdisco')->resultset('NodeIp')
-    ->search({ip => { -not_in =>
-        $jobqueue->search({
-          device => { '!=' => undef},
-          action => $job_type,
-          status => { -like => 'queued%' },
-        })->get_column('device')->as_query
-    }, -bool => 'active'});
-
-  my $ip_version = $job_type .'_ip_version';
-  my $job_ip_ver = $self->$ip_version;
-
-  if ($job_ip_ver) {
-    $rs = $rs->ip_version($job_ip_ver)
-  }
-
-  my $config_max_age = $job_type . '_max_age';
-  my $max_age        = setting($config_max_age);
-
-  if ($max_age) {
-      my $interval = "$max_age day";
-      $rs = $rs->search(
-          { time_last => \[ '>= now() - ?::interval', $interval ] } );
-  }
-
-  my @nodes = $rs->get_column('ip')->all;
-
-  my $filter_method = $job_type .'_filter';
-  my $job_filter = $self->$filter_method;
-
-  my @filtered_nodes = grep {$job_filter->($_)} @nodes;
-
-  schema('netdisco')->resultset('Admin')->txn_do_locked(sub {
-    $jobqueue->populate([
-      map {{
-          device => $_,
-          action => $job_type,
-          status => 'queued',
-      }} (@filtered_nodes)
-    ]);
-  });
-
-  return job_done("Queued $job_type job for all nodes");
-}
-
 sub _single_node_body {
-  my ($self, $job_type, $job) = @_;
+  my ($self, $job_type, $node, $now) = @_;
 
   my $action_method = $job_type .'_action';
   my $job_action = $self->$action_method;
 
-  my $host = NetAddr::IP::Lite->new($job->device);
-
   my $filter_method = $job_type .'_filter';
   my $job_filter = $self->$filter_method;
 
-  unless ($job_filter->($host->addr)) {
+  unless ($job_filter->($node)) {
-      return job_defer("$job_type deferred: $host is not ${job_type}able");
+      return job_defer("$job_type deferred: $node is not ${job_type}able");
   }
 
-  $job_action->($host->addr);
+  $job_action->($node, $now);
 
-  return job_done("Ended $job_type for ". $host->addr);
+  # would be ignored if wrapped in a loop
+  return job_done("Ended $job_type for $node");
 }
 
 1;

Netdisco/lib/App/Netdisco/Daemon/Worker/Poller/Nbtstat.pm

@@ -1,7 +1,18 @@
 package App::Netdisco::Daemon::Worker::Poller::Nbtstat;
 
+use Dancer qw/:moose :syntax :script/;
+use Dancer::Plugin::DBIC 'schema';
+
 use App::Netdisco::Core::Nbtstat 'do_nbtstat';
 use App::Netdisco::Util::Node 'is_nbtstatable';
+use App::Netdisco::Util::Device qw/get_device is_discoverable/;
+use App::Netdisco::Daemon::Util ':all';
+
+use MCE::Loop;
+MCE::Loop::init {}; # auto chunk size
+
+use NetAddr::IP::Lite ':lower';
+use Time::HiRes 'gettimeofday';
 
 use Role::Tiny;
 use namespace::clean;
@@ -10,9 +21,41 @@ with 'App::Netdisco::Daemon::Worker::Poller::Common';
 
 sub nbtstat_action { \&do_nbtstat }
 sub nbtstat_filter { \&is_nbtstatable }
-sub nbtstat_ip_version { 4 }
 
-sub nbtwalk { (shift)->_walk_nodes_body('nbtstat', @_) }
+sub nbtwalk { (shift)->_walk_body('nbtstat', @_) }
-sub nbtstat  { (shift)->_single_node_body('nbtstat', @_) }
+
+sub nbtstat  {
+  my ($self, $job) = @_;
+
+  my $host = NetAddr::IP::Lite->new($job->device);
+  my $device = get_device($host->addr);
+
+  unless (is_discoverable($device->ip)) {
+      return job_defer("nbtstat deferred: $host is not discoverable");
+  }
+
+  # get list of nodes on device
+  my $interval = (setting('nbt_max_age') || 7) . ' day';
+  my $rs = schema('netdisco')->resultset('NodeIp')->search({
+    -bool => 'me.active',
+    -bool => 'nodes.active',
+    'nodes.switch' => $device->ip,
+    'me.time_last' => \[ '>= now() - ?::interval', $interval ],
+  },{
+    join => 'nodes',
+    columns => 'ip',
+    distinct => 1,
+  })->ip_version(4);
+
+  my $nodes = [$rs->get_column('ip')->all];
+  my $now = 'to_timestamp('. (join '.', gettimeofday) .')';
+
+  mce_loop {
+    $self->_single_node_body('nbtstat', $_, $now)
+      for @$_;
+  } $nodes;
+
+  return job_done("Ended nbtstat for ". $host->addr);
+}
 
 1;