refactor nbtstat to group probes by device

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)) {
-      return job_defer("$job_type deferred: $host is not ${job_type}able");
+  unless ($job_filter->($node)) {
+      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,15 @@
 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 NetAddr::IP::Lite ':lower';
+use Time::HiRes 'gettimeofday';
 
 use Role::Tiny;
 use namespace::clean;
@@ -10,9 +18,40 @@ with 'App::Netdisco::Daemon::Worker::Poller::Common';
 
 sub nbtstat_action { \&do_nbtstat }
 sub nbtstat_filter { \&is_nbtstatable }
-sub nbtstat_ip_version { 4 }
+sub nbtstat_layer  { 2 }
 
-sub nbtwalk { (shift)->_walk_nodes_body('nbtstat', @_) }
-sub nbtstat  { (shift)->_single_node_body('nbtstat', @_) }
+sub nbtwalk { (shift)->_walk_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) .')';
+
+  $self->_single_node_body('nbtstat', $_, $now)
+    for @nodes;
+
+  return job_done("Ended nbtstat for ". $host->addr);
+}
 
 1;