relocate repo files so ND2 is the only code

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

@@ -0,0 +1,70 @@
+package App::Netdisco::Daemon::Worker::Common;
+
+use Dancer qw/:moose :syntax :script/;
+
+use Try::Tiny;
+use App::Netdisco::Util::Daemon;
+
+use Role::Tiny;
+use namespace::clean;
+
+use Time::HiRes 'sleep';
+use App::Netdisco::JobQueue qw/jq_defer jq_complete/;
+
+sub worker_begin { (shift)->{started} = time }
+
+sub worker_body {
+  my $self = shift;
+  my $wid = $self->wid;
+
+  while (1) {
+      prctl sprintf 'netdisco-daemon: worker #%s poller: idle', $wid;
+
+      my $job = $self->{queue}->dequeue(1);
+      next unless defined $job;
+      my $action = $job->action;
+
+      try {
+          $job->started(scalar localtime);
+          prctl sprintf 'netdisco-daemon: worker #%s poller: working on #%s: %s',
+            $wid, $job->job, $job->summary;
+          info sprintf "pol (%s): starting %s job(%s) at %s",
+            $wid, $action, $job->job, $job->started;
+          my ($status, $log) = $self->$action($job);
+          $job->status($status);
+          $job->log($log);
+      }
+      catch {
+          $job->status('error');
+          $job->log("error running job: $_");
+          $self->sendto('stderr', $job->log ."\n");
+      };
+
+      $self->close_job($job);
+      sleep( setting('workers')->{'min_runtime'} || 0 );
+      $self->exit(0); # recycle worker
+  }
+}
+
+sub close_job {
+  my ($self, $job) = @_;
+  my $now  = scalar localtime;
+
+  prctl sprintf 'netdisco-daemon: worker #%s poller: wrapping up %s #%s: %s',
+    $self->wid, $job->action, $job->job, $job->status;
+  info sprintf "pol (%s): wrapping up %s job(%s) - status %s at %s",
+    $self->wid, $job->action, $job->job, $job->status, $now;
+
+  try {
+      if ($job->status eq 'defer') {
+          jq_defer($job);
+      }
+      else {
+          $job->finished($now);
+          jq_complete($job);
+      }
+  }
+  catch { $self->sendto('stderr', "error closing job: $_\n") };
+}
+
+1;

lib/App/Netdisco/Daemon/Worker/Interactive/DeviceActions.pm

@@ -0,0 +1,50 @@
+package App::Netdisco::Daemon::Worker::Interactive::DeviceActions;
+
+use App::Netdisco::Util::SNMP 'snmp_connect_rw';
+use App::Netdisco::Util::Device 'get_device';
+use App::Netdisco::Daemon::Util ':all';
+
+use Role::Tiny;
+use namespace::clean;
+
+sub location {
+  my ($self, $job) = @_;
+  return _set_device_generic($job->device, 'location', $job->subaction);
+}
+
+sub contact {
+  my ($self, $job) = @_;
+  return _set_device_generic($job->device, 'contact', $job->subaction);
+}
+
+sub _set_device_generic {
+  my ($ip, $slot, $data) = @_;
+  $data ||= '';
+
+  # snmp connect using rw community
+  my $info = snmp_connect_rw($ip)
+    or return job_error("Failed to connect to device [$ip] to update $slot");
+
+  my $method = 'set_'. $slot;
+  my $rv = $info->$method($data);
+
+  if (!defined $rv) {
+      return job_error(sprintf 'Failed to set %s on [%s]: %s',
+                    $slot, $ip, ($info->error || ''));
+  }
+
+  # confirm the set happened
+  $info->clear_cache;
+  my $new_data = ($info->$slot || '');
+  if ($new_data ne $data) {
+      return job_error("Verify of $slot update failed on [$ip]: $new_data");
+  }
+
+  # update netdisco DB
+  my $device = get_device($ip);
+  $device->update({$slot => $data});
+
+  return job_done("Updated $slot on [$ip] to [$data]");
+}
+
+1;

lib/App/Netdisco/Daemon/Worker/Interactive/PortActions.pm

@@ -0,0 +1,159 @@
+package App::Netdisco::Daemon::Worker::Interactive::PortActions;
+
+use App::Netdisco::Util::Port ':all';
+use App::Netdisco::Util::SNMP 'snmp_connect_rw';
+use App::Netdisco::Util::Device 'get_device';
+use App::Netdisco::Daemon::Util ':all';
+
+use Role::Tiny;
+use namespace::clean;
+
+sub portname {
+  my ($self, $job) = @_;
+  return _set_port_generic($job, 'alias', 'name');
+}
+
+sub portcontrol {
+  my ($self, $job) = @_;
+
+  my $port = get_port($job->device, $job->port)
+    or return job_error(sprintf "Unknown port name [%s] on device [%s]",
+      $job->port, $job->device);
+
+  my $reconfig_check = port_reconfig_check($port);
+  return job_error("Cannot alter port: $reconfig_check")
+    if $reconfig_check;
+
+  # need to remove "-other" which appears for power/portcontrol
+  (my $sa = $job->subaction) =~ s/-\w+//;
+  $job->subaction($sa);
+
+  if ($sa eq 'bounce') {
+      $job->subaction('down');
+      my @stat = _set_port_generic($job, 'up_admin');
+      return @stat if $stat[0] ne 'done';
+      $job->subaction('up');
+      return _set_port_generic($job, 'up_admin');
+  }
+  else {
+      return _set_port_generic($job, 'up_admin');
+  }
+}
+
+sub vlan {
+  my ($self, $job) = @_;
+
+  my $port = get_port($job->device, $job->port)
+    or return job_error(sprintf "Unknown port name [%s] on device [%s]",
+      $job->port, $job->device);
+
+  my $port_reconfig_check = port_reconfig_check($port);
+  return job_error("Cannot alter port: $port_reconfig_check")
+    if $port_reconfig_check;
+
+  my $vlan_reconfig_check = vlan_reconfig_check($port);
+  return job_error("Cannot alter vlan: $vlan_reconfig_check")
+    if $vlan_reconfig_check;
+
+  my @stat = _set_port_generic($job, 'pvid'); # for Cisco trunk
+  return @stat if $stat[0] eq 'done';
+  return _set_port_generic($job, 'vlan');
+}
+
+sub _set_port_generic {
+  my ($job, $slot, $column) = @_;
+  $column ||= $slot;
+
+  my $device = get_device($job->device);
+  my $ip = $device->ip;
+  my $pn = $job->port;
+  my $data = $job->subaction;
+
+  my $port = get_port($ip, $pn)
+    or return job_error("Unknown port name [$pn] on device [$ip]");
+
+  if ($device->vendor ne 'netdisco') {
+      # snmp connect using rw community
+      my $info = snmp_connect_rw($ip)
+        or return job_error("Failed to connect to device [$ip] to control port");
+
+      my $iid = get_iid($info, $port)
+        or return job_error("Failed to get port ID for [$pn] from [$ip]");
+
+      my $method = 'set_i_'. $slot;
+      my $rv = $info->$method($data, $iid);
+
+      if (!defined $rv) {
+          return job_error(sprintf 'Failed to set [%s] %s to [%s] on [%s]: %s',
+                        $pn, $slot, $data, $ip, ($info->error || ''));
+      }
+
+      # confirm the set happened
+      $info->clear_cache;
+      my $check_method = 'i_'. $slot;
+      my $state = ($info->$check_method($iid) || '');
+      if (ref {} ne ref $state or $state->{$iid} ne $data) {
+          return job_error("Verify of [$pn] $slot failed on [$ip]");
+      }
+  }
+
+  # update netdisco DB
+  $port->update({$column => $data});
+
+  return job_done("Updated [$pn] $slot status on [$ip] to [$data]");
+}
+
+sub power {
+  my ($self, $job) = @_;
+
+  my $port = get_port($job->device, $job->port)
+    or return job_error(sprintf "Unknown port name [%s] on device [%s]",
+      $job->port, $job->device);
+
+  return job_error("No PoE service on port [%s] on device [%s]")
+    unless $port->power;
+
+  my $reconfig_check = port_reconfig_check($port);
+  return job_error("Cannot alter port: $reconfig_check")
+    if $reconfig_check;
+
+  my $device = get_device($job->device);
+  my $ip = $device->ip;
+  my $pn = $job->port;
+
+  # munge data
+  (my $data = $job->subaction) =~ s/-\w+//; # remove -other
+  $data = 'true'  if $data =~ m/^(on|yes|up)$/;
+  $data = 'false' if $data =~ m/^(off|no|down)$/;
+
+  # snmp connect using rw community
+  my $info = snmp_connect_rw($ip)
+    or return job_error("Failed to connect to device [$ip] to control port");
+
+  my $powerid = get_powerid($info, $port)
+    or return job_error("Failed to get power ID for [$pn] from [$ip]");
+
+  my $rv = $info->set_peth_port_admin($data, $powerid);
+
+  if (!defined $rv) {
+      return job_error(sprintf 'Failed to set [%s] power to [%s] on [%s]: %s',
+                    $pn, $data, $ip, ($info->error || ''));
+  }
+
+  # confirm the set happened
+  $info->clear_cache;
+  my $state = ($info->peth_port_admin($powerid) || '');
+  if (ref {} ne ref $state or $state->{$powerid} ne $data) {
+      return job_error("Verify of [$pn] power failed on [$ip]");
+  }
+
+  # update netdisco DB
+  $port->power->update({
+    admin => $data,
+    status => ($data eq 'false' ? 'disabled' : 'searching'),
+  });
+
+  return job_done("Updated [$pn] power status on [$ip] to [$data]");
+}
+
+1;

lib/App/Netdisco/Daemon/Worker/Manager.pm

@@ -0,0 +1,86 @@
+package App::Netdisco::Daemon::Worker::Manager;
+
+use Dancer qw/:moose :syntax :script/;
+
+use List::Util 'sum';
+use App::Netdisco::Util::Daemon;
+
+use Role::Tiny;
+use namespace::clean;
+
+use App::Netdisco::JobQueue qw/jq_locked jq_getsome jq_getsomep jq_lock/;
+
+sub worker_begin {
+  my $self = shift;
+  my $wid = $self->wid;
+
+  return debug "mgr ($wid): no need for manager... skip begin"
+    if setting('workers')->{'no_manager'};
+
+  debug "entering Manager ($wid) worker_begin()";
+
+  # requeue jobs locally
+  debug "mgr ($wid): searching for jobs booked to this processing node";
+  my @jobs = jq_locked;
+
+  if (scalar @jobs) {
+      info sprintf "mgr (%s): found %s jobs booked to this processing node",
+        $wid, scalar @jobs;
+      $self->{queue}->enqueuep(100, @jobs);
+  }
+}
+
+sub worker_body {
+  my $self = shift;
+  my $wid = $self->wid;
+
+  if (setting('workers')->{'no_manager'}) {
+      prctl sprintf 'netdisco-daemon: worker #%s manager: inactive', $wid;
+      return debug "mgr ($wid): no need for manager... quitting"
+  }
+
+  while (1) {
+      prctl sprintf 'netdisco-daemon: worker #%s manager: gathering', $wid;
+      my $num_slots = 0;
+
+      $num_slots = parse_max_workers( setting('workers')->{tasks} )
+                     - $self->{queue}->pending();
+      debug "mgr ($wid): getting potential jobs for $num_slots workers (HP)";
+
+      # get some high priority jobs
+      # TODO also check for stale jobs in Netdisco DB
+      foreach my $job ( jq_getsomep($num_slots) ) {
+
+          # mark job as running
+          next unless jq_lock($job);
+          info sprintf "mgr (%s): job %s booked out for this processing node",
+            $wid, $job->job;
+
+          # copy job to local queue
+          $self->{queue}->enqueuep(100, $job);
+      }
+
+      $num_slots = parse_max_workers( setting('workers')->{tasks} )
+                     - $self->{queue}->pending();
+      debug "mgr ($wid): getting potential jobs for $num_slots workers (NP)";
+
+      # get some normal priority jobs
+      # TODO also check for stale jobs in Netdisco DB
+      foreach my $job ( jq_getsome($num_slots) ) {
+
+          # mark job as running
+          next unless jq_lock($job);
+          info sprintf "mgr (%s): job %s booked out for this processing node",
+            $wid, $job->job;
+
+          # copy job to local queue
+          $self->{queue}->enqueue($job);
+      }
+
+      debug "mgr ($wid): sleeping now...";
+      prctl sprintf 'netdisco-daemon: worker #%s manager: idle', $wid;
+      sleep( setting('workers')->{sleep_time} || 1 );
+  }
+}
+
+1;

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

@@ -0,0 +1,18 @@
+package App::Netdisco::Daemon::Worker::Poller;
+
+use Role::Tiny;
+use namespace::clean;
+
+# main worker body
+with 'App::Netdisco::Daemon::Worker::Common';
+
+# add dispatch methods for poller tasks
+with 'App::Netdisco::Daemon::Worker::Poller::Device',
+     'App::Netdisco::Daemon::Worker::Poller::Arpnip',
+     'App::Netdisco::Daemon::Worker::Poller::Macsuck',
+     'App::Netdisco::Daemon::Worker::Poller::Nbtstat',
+     'App::Netdisco::Daemon::Worker::Poller::Expiry',
+     'App::Netdisco::Daemon::Worker::Interactive::DeviceActions',
+     'App::Netdisco::Daemon::Worker::Interactive::PortActions';
+
+1;

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

@@ -0,0 +1,18 @@
+package App::Netdisco::Daemon::Worker::Poller::Arpnip;
+
+use App::Netdisco::Core::Arpnip 'do_arpnip';
+use App::Netdisco::Util::Device 'is_arpnipable';
+
+use Role::Tiny;
+use namespace::clean;
+
+with 'App::Netdisco::Daemon::Worker::Poller::Common';
+
+sub arpnip_action { \&do_arpnip }
+sub arpnip_filter { \&is_arpnipable }
+sub arpnip_layer { 3 }
+
+sub arpwalk { (shift)->_walk_body('arpnip', @_) }
+sub arpnip  { (shift)->_single_body('arpnip', @_) }
+
+1;

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

@@ -0,0 +1,98 @@
+package App::Netdisco::Daemon::Worker::Poller::Common;
+
+use Dancer qw/:moose :syntax :script/;
+
+use App::Netdisco::Util::SNMP 'snmp_connect';
+use App::Netdisco::Util::Device 'get_device';
+use App::Netdisco::Daemon::Util ':all';
+use App::Netdisco::JobQueue qw/jq_queued jq_insert/;
+
+use Dancer::Plugin::DBIC 'schema';
+use NetAddr::IP::Lite ':lower';
+
+use Role::Tiny;
+use namespace::clean;
+
+# queue a job for all devices known to Netdisco
+sub _walk_body {
+  my ($self, $job_type, $job) = @_;
+
+  my $layer_method = $job_type .'_layer';
+  my $job_layer = $self->$layer_method;
+
+  my %queued = map {$_ => 1} jq_queued($job_type);
+  my @devices = schema('netdisco')->resultset('Device')
+    ->has_layer($job_layer)->get_column('ip')->all;
+  my @filtered_devices = grep {!exists $queued{$_}} @devices;
+
+  jq_insert([
+      map {{
+          device => $_,
+          action => $job_type,
+          username => $job->username,
+          userip => $job->userip,
+      }} (@filtered_devices)
+  ]);
+
+  return job_done("Queued $job_type job for all devices");
+}
+
+sub _single_body {
+  my ($self, $job_type, $job) = @_;
+
+  my $action_method = $job_type .'_action';
+  my $job_action = $self->$action_method;
+
+  my $layer_method = $job_type .'_layer';
+  my $job_layer = $self->$layer_method;
+
+  my $device = get_device($job->device)
+    or job_error("$job_type failed: unable to interpret device parameter");
+  my $host = $device->ip;
+
+  if ($device->in_storage
+      and $device->vendor and $device->vendor eq 'netdisco') {
+      return job_done("$job_type skipped: $host is pseudo-device");
+  }
+
+  my $filter_method = $job_type .'_filter';
+  my $job_filter = $self->$filter_method;
+
+  unless ($job_filter->($device->ip)) {
+      return job_defer("$job_type deferred: $host is not ${job_type}able");
+  }
+
+  my $snmp = snmp_connect($device);
+  if (!defined $snmp) {
+      return job_error("$job_type failed: could not SNMP connect to $host");
+  }
+
+  unless ($snmp->has_layer( $job_layer )) {
+      return job_done("Skipped $job_type for device $host without OSI layer $job_layer capability");
+  }
+
+  $job_action->($device, $snmp);
+
+  return job_done("Ended $job_type for $host");
+}
+
+sub _single_node_body {
+  my ($self, $job_type, $node, $now) = @_;
+
+  my $action_method = $job_type .'_action';
+  my $job_action = $self->$action_method;
+
+  my $filter_method = $job_type .'_filter';
+  my $job_filter = $self->$filter_method;
+
+  unless ($job_filter->($node)) {
+      return job_defer("$job_type deferred: $node is not ${job_type}able");
+  }
+
+  $job_action->($node, $now);
+
+  # would be ignored if wrapped in a loop
+  return job_done("Ended $job_type for $node");
+}
+
+1;

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

@@ -0,0 +1,99 @@
+package App::Netdisco::Daemon::Worker::Poller::Device;
+
+use Dancer qw/:moose :syntax :script/;
+
+use App::Netdisco::Util::SNMP 'snmp_connect';
+use App::Netdisco::Util::Device qw/get_device is_discoverable/;
+use App::Netdisco::Core::Discover ':all';
+use App::Netdisco::Daemon::Util ':all';
+use App::Netdisco::JobQueue qw/jq_queued jq_insert/;
+
+use Dancer::Plugin::DBIC 'schema';
+use NetAddr::IP::Lite ':lower';
+
+use Role::Tiny;
+use namespace::clean;
+
+# queue a discover job for all devices known to Netdisco
+sub discoverall {
+  my ($self, $job) = @_;
+
+  my %queued = map {$_ => 1} jq_queued('discover');
+  my @devices = schema('netdisco')->resultset('Device')
+    ->get_column('ip')->all;
+  my @filtered_devices = grep {!exists $queued{$_}} @devices;
+
+  jq_insert([
+      map {{
+          device => $_,
+          action => 'discover',
+          username => $job->username,
+          userip => $job->userip,
+      }} (@filtered_devices)
+  ]);
+
+  return job_done("Queued discover job for all devices");
+}
+
+# run a discover job for one device, and its *new* neighbors
+sub discover {
+  my ($self, $job) = @_;
+
+  my $device = get_device($job->device)
+    or return job_error(
+      "discover failed: unable to interpret device parameter: "
+      . ($job->device || "''"));
+  my $host = $device->ip;
+
+  if ($device->ip eq '0.0.0.0') {
+      return job_error("discover failed: no device param (need -d ?)");
+  }
+
+  if ($device->in_storage
+      and $device->vendor and $device->vendor eq 'netdisco') {
+      return job_done("discover skipped: $host is pseudo-device");
+  }
+
+  unless (is_discoverable($device->ip)) {
+      return job_defer("discover deferred: $host is not discoverable");
+  }
+
+  my $snmp = snmp_connect($device);
+  if (!defined $snmp) {
+      return job_error("discover failed: could not SNMP connect to $host");
+  }
+
+  store_device($device, $snmp);
+  set_canonical_ip($device, $snmp); # must come after store_device
+  store_interfaces($device, $snmp);
+  store_wireless($device, $snmp);
+  store_vlans($device, $snmp);
+  store_power($device, $snmp);
+  store_modules($device, $snmp) if setting('store_modules');
+  discover_new_neighbors($device, $snmp);
+
+  # if requested, and the device has not yet been arpniped/macsucked, queue now
+  if ($device->in_storage and $job->subaction and $job->subaction eq 'with-nodes') {
+      if (!defined $device->last_macsuck) {
+          jq_insert({
+              device => $device->ip,
+              action => 'macsuck',
+              username => $job->username,
+              userip => $job->userip,
+          });
+      }
+
+      if (!defined $device->last_arpnip) {
+          jq_insert({
+              device => $device->ip,
+              action => 'arpnip',
+              username => $job->username,
+              userip => $job->userip,
+          });
+      }
+  }
+
+  return job_done("Ended discover for $host");
+}
+
+1;

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

@@ -0,0 +1,73 @@
+package App::Netdisco::Daemon::Worker::Poller::Expiry;
+
+use Dancer qw/:moose :syntax :script/;
+use Dancer::Plugin::DBIC 'schema';
+
+use App::Netdisco::Daemon::Util ':all';
+
+use Role::Tiny;
+use namespace::clean;
+
+# expire devices and nodes according to config
+sub expire {
+  my ($self, $job) = @_;
+
+  if (setting('expire_devices') and setting('expire_devices') > 0) {
+      schema('netdisco')->txn_do(sub {
+        schema('netdisco')->resultset('Device')->search({
+          last_discover => \[q/< (now() - ?::interval)/,
+              (setting('expire_devices') * 86400)],
+        })->delete();
+      });
+  }
+
+  if (setting('expire_nodes') and setting('expire_nodes') > 0) {
+      schema('netdisco')->txn_do(sub {
+        schema('netdisco')->resultset('Node')->search({
+          time_last => \[q/< (now() - ?::interval)/,
+              (setting('expire_nodes') * 86400)],
+        })->delete();
+      });
+  }
+
+  if (setting('expire_nodes_archive') and setting('expire_nodes_archive') > 0) {
+      schema('netdisco')->txn_do(sub {
+        schema('netdisco')->resultset('Node')->search({
+          -not_bool => 'active',
+          time_last => \[q/< (now() - ?::interval)/,
+              (setting('expire_nodes_archive') * 86400)],
+        })->delete();
+      });
+  }
+
+  if (setting('expire_jobs') and setting('expire_jobs') > 0) {
+      schema('netdisco')->txn_do(sub {
+        schema('netdisco')->resultset('Admin')->search({
+          entered => \[q/< (now() - ?::interval)/,
+              (setting('expire_jobs') * 86400)],
+        })->delete();
+      });
+  }
+
+  return job_done("Checked expiry for all Devices and Nodes");
+}
+
+# expire nodes for a specific device
+sub expirenodes {
+  my ($self, $job) = @_;
+
+  return job_error('Missing device') unless $job->device;
+
+  schema('netdisco')->txn_do(sub {
+    schema('netdisco')->resultset('Node')->search({
+      switch => $job->device->ip,
+      ($job->port ? (port => $job->port) : ()),
+    })->delete(
+      ($job->extra ? () : ({ archive_nodes => 1 }))
+    );
+  });
+
+  return job_done("Expired nodes for ". $job->device->ip);
+}
+
+1;

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

@@ -0,0 +1,18 @@
+package App::Netdisco::Daemon::Worker::Poller::Macsuck;
+
+use App::Netdisco::Core::Macsuck 'do_macsuck';
+use App::Netdisco::Util::Device 'is_macsuckable';
+
+use Role::Tiny;
+use namespace::clean;
+
+with 'App::Netdisco::Daemon::Worker::Poller::Common';
+
+sub macsuck_action { \&do_macsuck }
+sub macsuck_filter { \&is_macsuckable }
+sub macsuck_layer { 2 }
+
+sub macwalk { (shift)->_walk_body('macsuck', @_) }
+sub macsuck { (shift)->_single_body('macsuck', @_) }
+
+1;

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

@@ -0,0 +1,73 @@
+package App::Netdisco::Daemon::Worker::Poller::Nbtstat;
+
+use Dancer qw/:moose :syntax :script/;
+use Dancer::Plugin::DBIC 'schema';
+
+use App::Netdisco::Core::Nbtstat qw/nbtstat_resolve_async store_nbt/;
+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;
+
+with 'App::Netdisco::Daemon::Worker::Poller::Common';
+
+sub nbtstat_action { \&do_nbtstat }
+sub nbtstat_filter { \&is_nbtstatable }
+sub nbtstat_layer  { 2 }
+
+sub nbtwalk { (shift)->_walk_body('nbtstat', @_) }
+
+sub nbtstat  {
+  my ($self, $job) = @_;
+
+  my $device = get_device($job->device)
+    or job_error("nbtstat failed: unable to interpret device parameter");
+  my $host = $device->ip;
+
+  unless (is_discoverable($device->ip)) {
+      return job_defer("nbtstat deferred: $host is not discoverable");
+  }
+
+  # get list of nodes on device
+  my $interval = (setting('nbtstat_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;
+
+  # Unless we have IP's don't bother
+  if (scalar @nodes) {
+    # filter exclusions from config
+    @nodes = grep { is_nbtstatable( $_ ) } @nodes;
+
+    # setup the hash nbtstat_resolve_async expects
+    my @ips = map {+{'ip' => $_}} @nodes;
+    my $now = 'to_timestamp('. (join '.', gettimeofday) .')';
+
+    my $resolved_nodes = nbtstat_resolve_async(\@ips);
+
+    # update node_nbt with status entries
+    foreach my $result (@$resolved_nodes) {
+      if (defined $result->{'nbname'}) {
+        store_nbt($result, $now);
+      }
+    }
+  }
+
+  return job_done("Ended nbtstat for $host");
+}
+
+1;

lib/App/Netdisco/Daemon/Worker/Scheduler.pm

@@ -0,0 +1,80 @@
+package App::Netdisco::Daemon::Worker::Scheduler;
+
+use Dancer qw/:moose :syntax :script/;
+
+use Algorithm::Cron;
+use App::Netdisco::Util::Daemon;
+
+use Role::Tiny;
+use namespace::clean;
+
+use App::Netdisco::JobQueue qw/jq_insert/;
+
+sub worker_begin {
+  my $self = shift;
+  my $wid = $self->wid;
+
+  return debug "sch ($wid): no need for scheduler... skip begin"
+    unless setting('schedule');
+
+  debug "entering Scheduler ($wid) worker_begin()";
+
+  foreach my $action (keys %{ setting('schedule') }) {
+      my $config = setting('schedule')->{$action};
+
+      # accept either single crontab format, or individual time fields
+      $config->{when} = Algorithm::Cron->new(
+        base => 'local',
+        %{
+          (ref {} eq ref $config->{when})
+            ? $config->{when}
+            : {crontab => $config->{when}}
+        }
+      );
+  }
+}
+
+sub worker_body {
+  my $self = shift;
+  my $wid = $self->wid;
+
+  unless (setting('schedule')) {
+      prctl sprintf 'netdisco-daemon: worker #%s scheduler: inactive', $wid;
+      return debug "sch ($wid): no need for scheduler... quitting"
+  }
+
+  while (1) {
+      # sleep until some point in the next minute
+      my $naptime = 60 - (time % 60) + int(rand(45));
+
+      prctl sprintf 'netdisco-daemon: worker #%s scheduler: idle', $wid;
+      debug "sched ($wid): sleeping for $naptime seconds";
+
+      sleep $naptime;
+      prctl sprintf 'netdisco-daemon: worker #%s scheduler: queueing', $wid;
+
+      # NB next_time() returns the next *after* win_start
+      my $win_start = time - (time % 60) - 1;
+      my $win_end   = $win_start + 60;
+
+      # if any job is due, add it to the queue
+      foreach my $action (keys %{ setting('schedule') }) {
+          my $sched = setting('schedule')->{$action};
+
+          # next occurence of job must be in this minute's window
+          debug sprintf "sched ($wid): $action: win_start: %s, win_end: %s, next: %s",
+            $win_start, $win_end, $sched->{when}->next_time($win_start);
+          next unless $sched->{when}->next_time($win_start) <= $win_end;
+
+          # queue it!
+          info "sched ($wid): queueing $action job";
+          jq_insert({
+            action => $action,
+            device => $sched->{device},
+            extra  => $sched->{extra},
+          });
+      }
+  }
+}
+
+1;