refactor to use role for the worker
This commit is contained in:
@@ -5,14 +5,7 @@ use Dancer::Plugin::DBIC 'schema';
|
|||||||
|
|
||||||
use Daemon::Generic::While1;
|
use Daemon::Generic::While1;
|
||||||
use Parallel::Prefork;
|
use Parallel::Prefork;
|
||||||
use Netdisco::Util::DeviceProperties 'is_discoverable';
|
use Role::Tiny;
|
||||||
use Try::Tiny;
|
|
||||||
use Moo;
|
|
||||||
use MooX::Types::MooseLike::Base qw(InstanceOf);
|
|
||||||
|
|
||||||
# add dispatch methods for each port control action
|
|
||||||
with "Netdisco::Daemon::Actions::$_"
|
|
||||||
for (qw/Device Port/);
|
|
||||||
|
|
||||||
my $pp = Parallel::Prefork->new(
|
my $pp = Parallel::Prefork->new(
|
||||||
max_workers => 2,
|
max_workers => 2,
|
||||||
@@ -58,107 +51,17 @@ sub gd_run_body {
|
|||||||
$pp->{_no_adjust_until} = 0; # BUG in Prefork.pm
|
$pp->{_no_adjust_until} = 0; # BUG in Prefork.pm
|
||||||
}
|
}
|
||||||
|
|
||||||
# check for new jobs
|
if ($pp->num_workers < $pp->max_workers) {
|
||||||
# take one if available
|
|
||||||
|
|
||||||
# don't block waiting for a needed worker
|
|
||||||
if ($pp->num_workers >= $pp->max_workers) {
|
|
||||||
$self->gd_sleep( setting('daemon_sleep_time') || 5 );
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
$pp->start and return;
|
$pp->start and return;
|
||||||
try { $self->worker_body } catch { print "$_\n" };
|
with 'Netdisco::Daemon::Worker::Interactive';
|
||||||
|
$self->worker_body;
|
||||||
$pp->finish;
|
$pp->finish;
|
||||||
}
|
|
||||||
|
|
||||||
sub worker_body {
|
|
||||||
my $self = shift;
|
|
||||||
|
|
||||||
# get all pending jobs
|
|
||||||
my $rs = schema('netdisco')->resultset('Admin')->search({
|
|
||||||
action => [qw/location contact portcontrol portname vlan power/],
|
|
||||||
status => 'queued',
|
|
||||||
});
|
|
||||||
|
|
||||||
while (1) {
|
|
||||||
while (my $job = $rs->next) {
|
|
||||||
my $target = 'set_'. $job->action;
|
|
||||||
next unless $self->can($target);
|
|
||||||
|
|
||||||
# filter for discover_*
|
|
||||||
next unless is_discoverable($job->device);
|
|
||||||
|
|
||||||
# mark job as running
|
|
||||||
next unless $self->lock_job($job);
|
|
||||||
|
|
||||||
# do job
|
|
||||||
my ($status, $log);
|
|
||||||
try {
|
|
||||||
($status, $log) = $self->$target($job);
|
|
||||||
}
|
}
|
||||||
catch { warn "error running job: $_\n" };
|
|
||||||
|
|
||||||
# revert to queued status if we failed to action the job
|
# check for new jobs, take one if available
|
||||||
if (not $status) {
|
$self->manager_body;
|
||||||
$self->revert_job($job->job);
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
# update job state to done/error with log
|
|
||||||
$self->close_job($job->job, $status, $log);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
$rs->reset;
|
|
||||||
$self->gd_sleep( setting('daemon_sleep_time') || 5 );
|
$self->gd_sleep( setting('daemon_sleep_time') || 5 );
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
sub lock_job {
|
|
||||||
my ($self, $job) = @_;
|
|
||||||
|
|
||||||
# lock db table, check job state is still queued, update to running
|
|
||||||
try {
|
|
||||||
my $status_updated = schema('netdisco')->txn_do(sub {
|
|
||||||
my $row = schema('netdisco')->resultset('Admin')->find(
|
|
||||||
{job => $job->job},
|
|
||||||
{for => 'update'}
|
|
||||||
);
|
|
||||||
|
|
||||||
return 0 if $row->status ne 'queued';
|
|
||||||
$row->update({status => 'running', started => \'now()'});
|
|
||||||
return 1;
|
|
||||||
});
|
|
||||||
|
|
||||||
return 0 if not $status_updated;
|
|
||||||
}
|
|
||||||
catch {
|
|
||||||
warn "error locking job: $_\n";
|
|
||||||
return 0;
|
|
||||||
};
|
|
||||||
|
|
||||||
return 1;
|
|
||||||
}
|
|
||||||
|
|
||||||
sub revert_job {
|
|
||||||
my ($self, $id) = @_;
|
|
||||||
|
|
||||||
try {
|
|
||||||
schema('netdisco')->resultset('Admin')
|
|
||||||
->find($id)
|
|
||||||
->update({status => 'queued', started => undef});
|
|
||||||
}
|
|
||||||
catch { warn "error reverting job: $_\n" };
|
|
||||||
}
|
|
||||||
|
|
||||||
sub close_job {
|
|
||||||
my ($self, $id, $status, $log) = @_;
|
|
||||||
|
|
||||||
try {
|
|
||||||
schema('netdisco')->resultset('Admin')
|
|
||||||
->find($id)
|
|
||||||
->update({status => $status, log => $log, finished => \'now()'});
|
|
||||||
}
|
|
||||||
catch { warn "error closing job: $_\n" };
|
|
||||||
}
|
}
|
||||||
|
|
||||||
# do not remove - must be redefined for Daemon::Generic
|
# do not remove - must be redefined for Daemon::Generic
|
||||||
|
|||||||
105
Netdisco/lib/Netdisco/Daemon/Worker/Interactive.pm
Executable file
105
Netdisco/lib/Netdisco/Daemon/Worker/Interactive.pm
Executable file
@@ -0,0 +1,105 @@
|
|||||||
|
package Netdisco::Daemon::Worker::Interactive;
|
||||||
|
|
||||||
|
use Dancer qw/:moose :syntax :script/;
|
||||||
|
use Dancer::Plugin::DBIC 'schema';
|
||||||
|
|
||||||
|
use Netdisco::Util::DeviceProperties 'is_discoverable';
|
||||||
|
use Try::Tiny;
|
||||||
|
|
||||||
|
use Role::Tiny;
|
||||||
|
use namespace::clean;
|
||||||
|
|
||||||
|
# add dispatch methods for interactive actions
|
||||||
|
with 'Netdisco::Daemon::Worker::Interactive::DeviceActions',
|
||||||
|
'Netdisco::Daemon::Worker::Interactive::PortActions';
|
||||||
|
|
||||||
|
sub worker_body {
|
||||||
|
my $self = shift;
|
||||||
|
|
||||||
|
# get all pending jobs
|
||||||
|
my $rs = schema('netdisco')->resultset('Admin')->search({
|
||||||
|
action => [qw/location contact portcontrol portname vlan power/],
|
||||||
|
status => 'queued',
|
||||||
|
});
|
||||||
|
|
||||||
|
while (1) {
|
||||||
|
while (my $job = $rs->next) {
|
||||||
|
my $target = 'set_'. $job->action;
|
||||||
|
next unless $self->can($target);
|
||||||
|
|
||||||
|
# filter for discover_*
|
||||||
|
next unless is_discoverable($job->device);
|
||||||
|
|
||||||
|
# mark job as running
|
||||||
|
next unless $self->lock_job($job);
|
||||||
|
|
||||||
|
# do job
|
||||||
|
my ($status, $log);
|
||||||
|
try {
|
||||||
|
($status, $log) = $self->$target($job);
|
||||||
|
}
|
||||||
|
catch { warn "error running job: $_\n" };
|
||||||
|
|
||||||
|
# revert to queued status if we failed to action the job
|
||||||
|
if (not $status) {
|
||||||
|
$self->revert_job($job->job);
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
# update job state to done/error with log
|
||||||
|
$self->close_job($job->job, $status, $log);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
$rs->reset;
|
||||||
|
$self->gd_sleep( setting('daemon_sleep_time') || 5 );
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
sub lock_job {
|
||||||
|
my ($self, $job) = @_;
|
||||||
|
|
||||||
|
# lock db table, check job state is still queued, update to running
|
||||||
|
try {
|
||||||
|
my $status_updated = schema('netdisco')->txn_do(sub {
|
||||||
|
my $row = schema('netdisco')->resultset('Admin')->find(
|
||||||
|
{job => $job->job},
|
||||||
|
{for => 'update'}
|
||||||
|
);
|
||||||
|
|
||||||
|
return 0 if $row->status ne 'queued';
|
||||||
|
$row->update({status => 'running', started => \'now()'});
|
||||||
|
return 1;
|
||||||
|
});
|
||||||
|
|
||||||
|
return 0 if not $status_updated;
|
||||||
|
}
|
||||||
|
catch {
|
||||||
|
warn "error locking job: $_\n";
|
||||||
|
return 0;
|
||||||
|
};
|
||||||
|
|
||||||
|
return 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
sub revert_job {
|
||||||
|
my ($self, $id) = @_;
|
||||||
|
|
||||||
|
try {
|
||||||
|
schema('netdisco')->resultset('Admin')
|
||||||
|
->find($id)
|
||||||
|
->update({status => 'queued', started => undef});
|
||||||
|
}
|
||||||
|
catch { warn "error reverting job: $_\n" };
|
||||||
|
}
|
||||||
|
|
||||||
|
sub close_job {
|
||||||
|
my ($self, $id, $status, $log) = @_;
|
||||||
|
|
||||||
|
try {
|
||||||
|
schema('netdisco')->resultset('Admin')
|
||||||
|
->find($id)
|
||||||
|
->update({status => $status, log => $log, finished => \'now()'});
|
||||||
|
}
|
||||||
|
catch { warn "error closing job: $_\n" };
|
||||||
|
}
|
||||||
|
|
||||||
|
1;
|
||||||
@@ -1,10 +1,10 @@
|
|||||||
package Netdisco::Daemon::Actions::Device;
|
package Netdisco::Daemon::Worker::Interactive::DeviceActions;
|
||||||
|
|
||||||
use Netdisco::Util::Connect qw/snmp_connect get_device/;
|
use Netdisco::Util::Connect qw/snmp_connect get_device/;
|
||||||
use Netdisco::Daemon::Actions::Util ':all';
|
use Netdisco::Daemon::Worker::Interactive::Util ':all';
|
||||||
|
|
||||||
|
use Role::Tiny;
|
||||||
use namespace::clean;
|
use namespace::clean;
|
||||||
use Moo::Role;
|
|
||||||
|
|
||||||
sub set_location {
|
sub set_location {
|
||||||
my ($self, $job) = @_;
|
my ($self, $job) = @_;
|
||||||
@@ -1,11 +1,11 @@
|
|||||||
package Netdisco::Daemon::Actions::Port;
|
package Netdisco::Daemon::Worker::Interactive::PortActions;
|
||||||
|
|
||||||
use Netdisco::Util::Connect ':all';
|
use Netdisco::Util::Connect ':all';
|
||||||
use Netdisco::Util::Permissions ':all';
|
use Netdisco::Util::Permissions ':all';
|
||||||
use Netdisco::Daemon::Actions::Util ':all';
|
use Netdisco::Daemon::Worker::Interactive::Util ':all';
|
||||||
|
|
||||||
|
use Role::Tiny;
|
||||||
use namespace::clean;
|
use namespace::clean;
|
||||||
use Moo::Role;
|
|
||||||
|
|
||||||
sub set_portname {
|
sub set_portname {
|
||||||
my ($self, $job) = @_;
|
my ($self, $job) = @_;
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
package Netdisco::Daemon::Actions::Util;
|
package Netdisco::Daemon::Worker::Interactive::Util;
|
||||||
|
|
||||||
# support utilities for Daemon Actions
|
# support utilities for Daemon Actions
|
||||||
|
|
||||||
@@ -44,7 +44,7 @@ install Perl dependencies into a custom library path:
|
|||||||
NetAddr::IP \
|
NetAddr::IP \
|
||||||
Net::MAC \
|
Net::MAC \
|
||||||
namespace::clean \
|
namespace::clean \
|
||||||
Moo \
|
Role::Tiny \
|
||||||
List::MoreUtils \
|
List::MoreUtils \
|
||||||
Socket6 \
|
Socket6 \
|
||||||
HTML::Entities \
|
HTML::Entities \
|
||||||
|
|||||||
Reference in New Issue
Block a user