QudoでSCPによるログ回収してみるテスト

config.yml

---
default:
  path:
    access_log: /var/log/data-mining
    action_log: /var/log/data-mining
server:
  - host: 192.168.20.11
  - host: 192.168.20.12
  - host: 192.168.20.13

Enqueue.pl

#!/usr/bin/env perl

use strict;
use warnings;

use Qudo;
use FindBin;
use YAML::Tiny;

my %argv;
use App::Options(
    values => \%argv,
    option => {
        access_log => {
            type => 'boolean',
        },
        action_log => {
            type => 'boolean',
        },
    },
);

my $conf = YAML::Tiny::LoadFile($FindBin::Bin.'/config.yml');

my $qudo = Qudo->new(
    driver_class => 'Skinny',
    databases => [+{
        dsn      => 'dbi:mysql:qudo',
        username => 'root',
        password => '',
    }],
);
$qudo->manager->register_hooks('Qudo::Hook::Serialize::JSON');

for my $server (@{$conf->{server}}) {
    my $arg = {
        host => $server->{host},
        path => {
            access_log => $server->{access_log} || $conf->{default}->{path}->{access_log},
            action_log => $server->{action_log} || $conf->{default}->{path}->{action_log},
        },
    };
    $qudo->enqueue("RetrieveAccessLog", { arg => $arg }) if($argv{'access_log'});
    $qudo->enqueue("RetrieveActionLog", { arg => $arg }) if($argv{'action_log'});
}

Manager.pl

#!/usr/bin/env perl

use strict;
use warnings;
use Qudo::Parallel::Manager;

my $manager = Qudo::Parallel::Manager->new(
    databases => [+{
        dsn      => 'dbi:mysql:qudo',
        username => 'root',
        password => '',
    }],
    manager_abilities      => [qw/RetrieveAccessLog RetrieveActionLog/],
    work_delay             => 5,
    max_workers            => 5,
    min_spare_workers      => 3,
    max_spare_workers      => 5,
    max_request_par_child  => 30,
    auto_load_worker       => 1,
    admin                  => 1,
    debug                  => 1,
);
$manager->{qudo}->manager->register_hooks(qw/Qudo::Hook::Serialize::JSON/);
$manager->run;

RetrieveAccessLog.pm

package RetrieveAccessLog;

use strict;
use warnings;
use base qw/Qudo::Worker/;

use DateTime;
use Net::SCP;
use FindBin;

sub max_retries { 3 }

sub work {
    my ($self , $job ) = @_;
    my $arg = $job->arg;

    my $today = DateTime->today->strftime("%Y%m%d");
    my $local_dir = $FindBin::Bin.'/'.$arg->{host};
    mkdir $local_dir || die $!;

    my $log = 'access_log';
    my $local_path = $local_dir.'/'.$log.'.'.$today;
    my $remote_path = $arg->{path}->{$log}.'/'.$log.'.'.$today;

    print '['.$job->id."]start\n";

    my $scp = Net::SCP->new($arg->{host}) || die;
    $scp->login('data-mining') || die;

    print '['.$job->id.']REMOTE: '.$remote_path.' -> LOCAL: '.$local_path."\n";
    $scp->get($remote_path, $local_path) || die;

    print '['.$job->id."]finish\n";

    $job->completed();
}

1;

RetrieveActionLog.pm

package RetrieveActionLog;

use strict;
use warnings;
use base qw/Qudo::Worker/;

use DateTime;
use Net::SCP;
use FindBin;

sub max_retries { 3 }

sub work {
    my ($self , $job ) = @_;
    my $arg = $job->arg;

    my $today = DateTime->today->strftime("%Y%m%d");
    my $local_dir = $FindBin::Bin.'/'.$arg->{host};
    mkdir $local_dir || die $!;

    my $log = 'action_log';
    my $local_path = $local_dir.'/'.$log.'.'.$today;
    my $remote_path = $arg->{path}->{$log}.'/'.$log.'.'.$today;

    print '['.$job->id."]start\n";

    my $scp = Net::SCP->new($arg->{host}) || die;
    $scp->login('data-mining') || die;

    print '['.$job->id.']REMOTE: '.$remote_path.' -> LOCAL: '.$local_path."\n";
    $scp->get($remote_path, $local_path) || die;

    print '['.$job->id."]finish\n";

    $job->completed();
}

1;