135/netdisco
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

New backend daemon code, no SQLite. MCE::Flow.

Browse Source 
Squashed commit of the following:

commit 3284b62509
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 21:17:06 2014 +0100

    config defaults tidying

commit ade7bcd880
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 20:00:01 2014 +0100

    high priority jobs are picked first and inserted to prio queue

commit d450dfd2bd
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 19:25:21 2014 +0100

    better status

commit b8a742e5de
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 16:54:03 2014 +0100

    update proctitle when worker not running

commit 0c3675a8f4
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 16:48:58 2014 +0100

    remove all trace of SQLite - new lightweight Job object

commit a13ed25f6a
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 14:45:22 2014 +0100

    rename pollers to tasks

commit 44b50f413f
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 14:13:00 2014 +0100

    update docs

commit 517b1ae4c1
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 13:55:31 2014 +0100

    merge interactive and poller worker types

commit e9043b90e8
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 13:47:41 2014 +0100

    only take one job at a time per worker

commit 2366738d54
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 13:43:31 2014 +0100

    auto job priorities

commit 1fd473fd50
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 13:18:59 2014 +0100

    preload all worker modules into shared memory

commit 9ceb43c0f7
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 13:13:07 2014 +0100

    daemon clean

commit c817a35537
Author: Oliver Gorwits <oliver@cpan.org>
Date:   Sun Aug 10 12:36:24 2014 +0100

    first refactor for MCE::Flow and MCE::Queue
This commit is contained in:
Oliver Gorwits
2014-08-10 21:34:46 +01:00
parent d3992f57fd
commit 9afbe6d7e6
23 changed files with 278 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
Netdisco/lib/App/Netdisco/Daemon/DB.pm
View File

@@ -1,10 +0,0 @@
use utf8;
package App::Netdisco::Daemon::DB;
use strict;
use warnings;
use base 'DBIx::Class::Schema';
__PACKAGE__->load_namespaces;
1;

90
Netdisco/lib/App/Netdisco/Daemon/DB/Result/Admin.pm
View File

@@ -1,90 +0,0 @@
use utf8;
package App::Netdisco::Daemon::DB::Result::Admin;
use strict;
use warnings;
use base 'DBIx::Class::Core';
__PACKAGE__->table("admin");
__PACKAGE__->add_columns(
"job",
{ data_type => "integer", is_nullable => 0 },
"type", # Poller, Interactive, etc
{ data_type => "text", is_nullable => 0 },
"wid", # worker ID, only != 0 once taken
{ data_type => "integer", is_nullable => 0, default_value => 0 },
"entered",
{ data_type => "timestamp", is_nullable => 1 },
"started",
{ data_type => "timestamp", is_nullable => 1 },
"finished",
{ data_type => "timestamp", is_nullable => 1 },
"device",
{ data_type => "inet", is_nullable => 1 },
"port",
{ data_type => "text", is_nullable => 1 },
"action",
{ data_type => "text", is_nullable => 1 },
"subaction",
{ data_type => "text", is_nullable => 1 },
"status",
{ data_type => "text", is_nullable => 1 },
"username",
{ data_type => "text", is_nullable => 1 },
"userip",
{ data_type => "inet", is_nullable => 1 },
"log",
{ data_type => "text", is_nullable => 1 },
"debug",
{ data_type => "boolean", is_nullable => 1 },
);
__PACKAGE__->set_primary_key("job");
=head1 METHODS
=head2 summary
An attempt to make a meaningful statement about the job.
=cut
sub summary {
my $job = shift;
return join ' ',
$job->action,
($job->device || ''),
($job->port || '');
# ($job->subaction ? (q{'}. $job->subaction .q{'}) : '');
}
=head1 ADDITIONAL COLUMNS
=head2 extra
Alias for the C<subaction> column.
=cut
sub extra { (shift)->subaction }
=head2 entererd_stamp
Formatted version of the C<entered> field, accurate to the minute.
The format is somewhat like ISO 8601 or RFC3339 but without the middle C<T>
between the date stamp and time stamp. That is:
2012-02-06 12:49
=cut
sub entered_stamp {
(my $stamp = (shift)->entered) =~ s/\.\d+$//;
return $stamp;
}
1;

54
Netdisco/lib/App/Netdisco/Daemon/Job.pm Normal file
View File

@@ -0,0 +1,54 @@
package App::Netdisco::Daemon::Job;
use Moo;
use namespace::clean;
foreach my $slot (qw/
job
entered
started
finished
device
port
action
subaction
status
username
userip
log
debug
/) {
has $slot => (
is => 'rw',
);
}
=head1 METHODS
=head2 summary
An attempt to make a meaningful statement about the job.
=cut
sub summary {
my $job = shift;
return join ' ',
$job->action,
($job->device || ''),
($job->port || '');
# ($job->subaction ? (q{'}. $job->subaction .q{'}) : '');
}
=head1 ADDITIONAL COLUMNS
=head2 extra
Alias for the C<subaction> column.
=cut
sub extra { (shift)->subaction }
1;

69
Netdisco/lib/App/Netdisco/Daemon/LocalQueue.pm
View File

@@ -1,69 +0,0 @@
package App::Netdisco::Daemon::LocalQueue;
use Dancer qw/:moose :syntax :script/;
use Dancer::Plugin::DBIC 'schema';
use base 'Exporter';
our @EXPORT = ();
our @EXPORT_OK = qw/ add_jobs capacity_for take_jobs reset_jobs release_jobs /;
our %EXPORT_TAGS = ( all => \@EXPORT_OK );
schema('daemon')->deploy;
my $queue = schema('daemon')->resultset('Admin');
sub add_jobs {
my (@jobs) = @_;
info sprintf "adding %s jobs to local queue", scalar @jobs;
schema('daemon')->dclone($_)->insert for @jobs;
}
sub capacity_for {
my ($type) = @_;
debug "checking local capacity for worker type $type";
my $setting = setting('workers')->{ setting('job_type_keys')->{$type} };
my $current = $queue->search({type => $type})->count;
return ($current < $setting);
}
sub take_jobs {
my ($wid, $type, $max) = @_;
return () unless $wid > 1;
$max ||= 1;
debug "deleting completed jobs by worker $wid";
$queue->search({wid => $wid})->delete;
debug "searching for $max new jobs for worker $wid (type $type)";
my $rs = $queue->search(
{type => $type, wid => 0},
{rows => $max},
);
my @rows = $rs->all;
return [] if scalar @rows == 0;
debug sprintf "booking out %s jobs to worker %s", (scalar @rows), $wid;
$queue->search({job => { -in => [map {$_->job} @rows] }})
->update({wid => $wid});
return \@rows;
}
# not used by workers, only the daemon when reinitializing a worker
sub reset_jobs {
my ($wid) = @_;
debug "resetting jobs owned by worker $wid to be available";
return unless $wid > 1;
$queue->search({wid => $wid})
->update({wid => 0});
}
# not used by workers, only the daemon when reinitializing a worker
sub release_jobs {
my ($jid) = @_;
debug "releasing local job ID $jid";
$queue->search({job => $jid})->delete;
}
1;

56
Netdisco/lib/App/Netdisco/Daemon/Worker/Common.pm
View File

@@ -8,53 +8,47 @@ use App::Netdisco::Util::Daemon;
use Role::Tiny;
use namespace::clean;
use App::Netdisco::JobQueue qw/jq_take jq_defer jq_complete/;
use App::Netdisco::JobQueue qw/jq_defer jq_complete/;
sub worker_body {
my $self = shift;
my $wid = $self->wid;
my $tag = $self->worker_tag;
my $type = $self->worker_type;
while (1) {
prctl sprintf 'netdisco-daemon: worker #%s %s: idle', $wid, lc($type);
my $jobs = jq_take($self->wid, $type);
prctl sprintf 'netdisco-daemon: worker #%s poller: idle', $wid;
foreach my $job (@$jobs) {
my $target = $self->munge_action($job->action);
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 %s: working on #%s: %s',
$wid, lc($type), $job->id, $job->summary;
info sprintf "$tag (%s): starting %s job(%s) at %s",
$wid, $target, $job->id, $job->started;
my ($status, $log) = $self->$target($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);
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);
}
}
sub close_job {
my ($self, $job) = @_;
my $tag = $self->worker_tag;
my $type = $self->worker_type;
my $now = scalar localtime;
prctl sprintf 'netdisco-daemon: worker #%s %s: wrapping up %s #%s: %s',
$self->wid, lc($type), $job->action, $job->id, $job->status;
info sprintf "$tag (%s): wrapping up %s job(%s) - status %s at %s",
$self->wid, $job->action, $job->id, $job->status, $now;
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') {

17
Netdisco/lib/App/Netdisco/Daemon/Worker/Interactive.pm
View File

@@ -1,17 +0,0 @@
package App::Netdisco::Daemon::Worker::Interactive;
use Role::Tiny;
use namespace::clean;
# main worker body
with 'App::Netdisco::Daemon::Worker::Common';
# add dispatch methods for interactive actions
with 'App::Netdisco::Daemon::Worker::Interactive::DeviceActions',
'App::Netdisco::Daemon::Worker::Interactive::PortActions';
sub worker_tag { 'int' }
sub worker_type { 'Interactive' }
sub munge_action { 'set_' . $_[1] }
1;

4
Netdisco/lib/App/Netdisco/Daemon/Worker/Interactive/DeviceActions.pm
View File

@@ -7,12 +7,12 @@ use App::Netdisco::Daemon::Util ':all';
use Role::Tiny;
use namespace::clean;
sub set_location {
sub location {
my ($self, $job) = @_;
return _set_device_generic($job->device, 'location', $job->subaction);
}
sub set_contact {
sub contact {
my ($self, $job) = @_;
return _set_device_generic($job->device, 'contact', $job->subaction);
}

8
Netdisco/lib/App/Netdisco/Daemon/Worker/Interactive/PortActions.pm
View File

@@ -7,12 +7,12 @@ use App::Netdisco::Daemon::Util ':all';
use Role::Tiny;
use namespace::clean;
sub set_portname {
sub portname {
my ($self, $job) = @_;
return _set_port_generic($job, 'alias', 'name');
}
sub set_portcontrol {
sub portcontrol {
my ($self, $job) = @_;
my $port = get_port($job->device, $job->port)
@@ -39,7 +39,7 @@ sub set_portcontrol {
}
}
sub set_vlan {
sub vlan {
my ($self, $job) = @_;
my $port = get_port($job->device, $job->port)
@@ -97,7 +97,7 @@ sub _set_port_generic {
return job_done("Updated [$pn] $slot status on [$ip] to [$data]");
}
sub set_power {
sub power {
my ($self, $job) = @_;
my $port = get_port($job->device, $job->port)

56
Netdisco/lib/App/Netdisco/Daemon/Worker/Manager.pm
View File

@@ -8,7 +8,8 @@ use App::Netdisco::Util::Daemon;
use Role::Tiny;
use namespace::clean;
use App::Netdisco::JobQueue qw/jq_locked jq_getsome jq_lock/;
use App::Netdisco::JobQueue qw/jq_locked jq_getsome jq_getsomep jq_lock/;
use MCE::Util ();
sub worker_begin {
my $self = shift;
@@ -26,7 +27,7 @@ sub worker_begin {
if (scalar @jobs) {
info sprintf "mgr (%s): found %s jobs booked to this processing node",
$wid, scalar @jobs;
$self->do('add_jobs', @jobs);
$self->{queue}->enqueuep(100, @jobs);
}
}
@@ -34,39 +35,54 @@ sub worker_body {
my $self = shift;
my $wid = $self->wid;
return debug "mgr ($wid): no need for manager... quitting"
if setting('workers')->{'no_manager'};
my $num_slots = sum( 0, map { setting('workers')->{$_} }
values %{setting('job_type_keys')} );
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) {
debug "mgr ($wid): getting potential jobs for $num_slots workers";
prctl sprintf 'netdisco-daemon: worker #%s manager: gathering', $wid;
my $num_slots = 0;
# get some pending jobs
$num_slots =
MCE::Util::_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_getsome($num_slots) ) {
# check for available local capacity
my $job_type = setting('job_types')->{$job->action};
next unless $job_type and $self->do('capacity_for', $job_type);
debug sprintf "mgr (%s): processing node has capacity for job %s (%s)",
$wid, $job->id, $job->action;
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->id;
$wid, $job->job;
# copy job to local queue
$self->do('add_jobs', $job);
$self->{queue}->enqueuep(100, $job);
}
$num_slots =
MCE::Util::_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} || 2 );
sleep( setting('workers')->{sleep_time} || 1 );
}
}

8
Netdisco/lib/App/Netdisco/Daemon/Worker/Poller.pm
View File

@@ -11,10 +11,8 @@ 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';
sub worker_tag { 'pol' }
sub worker_type { 'Poller' }
sub munge_action { $_[1] }
'App::Netdisco::Daemon::Worker::Poller::Expiry',
'App::Netdisco::Daemon::Worker::Interactive::DeviceActions',
'App::Netdisco::Daemon::Worker::Interactive::PortActions';
1;

9
Netdisco/lib/App/Netdisco/Daemon/Worker/Scheduler.pm
View File

@@ -13,6 +13,10 @@ 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') }) {
@@ -34,6 +38,11 @@ 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));