File Coverage

blib/lib/MMM/MirrorTask.pm

Criterion	Covered	Total	%
statement	24	271	8.8
branch	0	126	0.0
condition	0	37	0.0
subroutine	8	33	24.2
pod	15	15	100.0
total	47	482	9.7

package MMM::MirrorTask;

=head1 NAME

MMM::MirrorTask Class to store mirror task function and data

=cut

use strict;
use warnings;
use MMM::Sync;
use MMM::Utils;
use MMM::Config;
use MMM::Mirror;
use Fcntl qw(:flock);
use Digest::MD5;

=head1 METHODS

=head2 new

=cut

sub new {
    my ($class, $mmm, $name, %options) = @_;
    bless(
        {
            mmm => $mmm,
            name => $name,
            options => { %options },
            lockcount => 0,
        },
        $class
    );
}

sub DESTROY {
    my ($self) = @_;
    $self->{lockcount} = 1; # Force unlock, can't happen
    $self->unlock();
}

=head2 name

Return the name of the current task

=cut

sub name {
    $_[0]->{name}
}

=head2 is_disable

Return true if the current task is disable

=cut

sub is_disable { yes_no($_[0]->val('disable')) }

=head2 val( $var, $default)

Return the configuration value for $var. Return $default if parameter
is not set in the config

=cut

sub val {
    my ( $self, $var, $default ) = @_;
    $self->{mmm}->{config}->val( $self->name, $var, $default );
}

=head2 frequency

Return the frequency value from config

=cut

sub frequency {
    my ($self) = @_;
    duration2m($self->val( 'period', PERIOD ));
}

sub _set_status_time {
    my ($config, $section, $var, $val) = @_;
    $config->newval($section, $var, $val);
    $config->SetParameterComment(
        $section, $var, scalar(gmtime($val))
    );
}

=head2 state_info

Return a hashref about job status

=cut

sub state_info {
    my ($self, $status) = @_;
    $status ||= $self->_get_status();
    my %info = ();
    foreach my $section (qw(job success failure)) {
        foreach my $var ($status->Parameters($section)) {
            $info{$section}{$var} = $status->val($section, $var);
        }
    }
    if  (yes_no($self->val('compute_size', 0))) {
        $info{job}{size} = $status->val('job', 'size');
    } else {
        delete($info{job}{size});
    }

    $info{job}{error_log} = [ grep { $_ } ($status->val('job', 'error_log')) ];
    $info{job}{next_run_time} = $self->next_run_time($status);
    $info{job}{is_running} = $self->is_running;
    return %info;
}

sub _compute_config_sum {
    my ($self) = @_;
    my $md5 = Digest::MD5->new();
    foreach (qw(url source path)) {
        $md5->add("$_=");
        $md5->add(join("\n", $self->val('job', $_, '')));
    }
    return $md5->hexdigest
}

sub _set_compute_config_sum {
    my ($self, $status) = @_;
    $status ||= $self->_get_status();
    my $newsum = $self->_compute_config_sum();
    if ($status->val('job', 'config_sum', '') ne $newsum) {
        $status->newval('job', 'config_sum', $newsum);
        return 1;
    } else {
        return 0;
    }
}

=head2 next_run_time

Return the time (in second) when the next run should be performed

=cut

sub next_run_time {
    my ($self, $status) = @_;
    my @alltime = ( scalar( time() ) );
    $status ||= $self->_get_status();
    my $last_start = 0;
    if ($self->_compute_config_sum() ne $status->val('job', 'config_sum', '')) {
        $self->_log('INFO',
	    'Config for has changed, need to be run immediately')
        if($status->val('job', 'start', 0));
    } else {
        $last_start = $status->val( 'job' , 'start', 0 );
    }

    if ( $last_start ) {
        push( @alltime, $last_start + ( $self->frequency * 60 ) );
    }

    if (   $self->val('waitafter', WAITAFTER_MINIMA)
        && $status->val( 'job' , 'end' ) )
    {
        push( @alltime,
            $status->val( 'job', 'end' ) +
              $self->val('waitafter', WAITAFTER_MINIMA) * 60 );
    }

    if (   $self->val('waitaftersuccess')
        && $status->val( 'success', 'end' ) )
    {
        push( @alltime,
            $status->val( 'job', 'success' ) +
              $self->val('waitaftersuccess') * 60 );
    }

    my ($t) = sort { $b <=> $a } @alltime;
    # $self->_log('DEBUG', 'Next run time for %s is %d (in %d), frequency is %d',
    #    $self->{name}, $t, $t - scalar(time()),
    #    $self->frequency,
    #);

    $t;
}

=head2 source

Return the source associate to the list, if any

=cut

sub source {
    $_[0]->val('source') || '';
}

sub _log {
    my ($self, $level, $message, @args) = @_;
    $self->{mmm}->log(
        $level,
        sprintf('[%s] %s', $self->name, $message),
        @args
    );
}

sub _lockpath {
    my ($self) = @_;
    my $lockfile = $self->name;
    $lockfile =~ s:/:_:g;
    join('/', ($self->{mmm}->statedir, "/$lockfile.lck"));
}

sub _statusfile {
    my ($self) = @_;
    my $lockfile = $self->name;
    $lockfile =~ s:/:_:g;
    join('/', ($self->{mmm}->statedir, $lockfile));
}

=head2 getlock($share)

Try to lock the lockfile for this task, in shared mode if
$share is set.

=cut

sub getlock {
    my ($self, $share) = @_;
    if ($self->{lockcount}) {
        $self->{lockcount}++;
        $self->_log(
            'DEBUG',
            'Lock is already done, counter is now %d', $self->{lockcount}
        );
        return $self->{lockcount};
    }
    $self->_log( 'DEBUG', 'Trying to acquire lock' );
    $self->{lockfile} = $self->_lockpath;
    if ( open( $self->{lockfh}, $share ? '<' : '>', $self->_lockpath) ) {
        if ( !flock(
                $self->{lockfh}, LOCK_NB | ( $share ? LOCK_SH : LOCK_EX ) ) ) {
            if ( ( $! + 0 ) != 11 ) { # E_AGAIN, does this is really need
                $self->_log(
                    'FATAL',
                    "Cannot lock file %s",
                    $self->_lockpath
                );
                unlink( $self->_lockpath );
                close( $self->{lockfh} );
                return;
            }
            $self->_log( 'DEBUG', 'is already lock' );
            return;
        }
        my $fh = $self->{lockfh};
        print $fh "$$\n" unless($share);
    }
    else {
        $self->_log( 'FATAL',
            'Cannot open lock file %s :%s', $self->{lockfile}, $!);
        return;
    }
    ++$self->{lockcount};
}

=head2 unlock

Release the lock for the task

=cut

sub unlock {
    my ($self) = @_;
    $self->{lockfh} or return;
    --$self->{lockcount} and return;
    unlink( $self->{lockfile} );
    close( $self->{lockfh} );
}

=head2 is_running

Return true is the task is running (lock check)

=cut

sub is_running {
    my ($self) = @_;
    my @stat = stat($self->_lockpath);
    if ($self->{mmm}->_task_is_registred($self->name)) {
        return $stat[9] || scalar(time);
    }

    if (!defined($stat[9])) { return }
    else {
        my $res = $self->getlock(1);
        if ($res) {
            $self->unlock;
	        if ($res > 1) { return $stat[9]; }
            else { return; }
        } else { return $stat[9]; }
    }
}

sub _get_status {
    my ($self) = @_;
    Config::IniFiles->new(
        -f $self->_statusfile ? ( -file => $self->_statusfile ) : ()
    ) || Config::IniFiles->new();
}

sub _write_status {
    my ($self, $status) = @_;
    $self->_log( 'DEBUG', 'Write status file: %s', $self->_statusfile );
    $status->WriteConfig( $self->_statusfile );
}

=head2 failure_count

Return three values:
- the count of failure since last success
- the previous failure count
- and if this count has change between the two previous run
(eg if failure count is different of previous failure count)

=cut

sub failure_count {
    my ($self, $status) = @_;
    my $before = defined($self->{successive_failure_before})
        ? $self->{successive_failure_before}
        : ($status ||= $self->_get_status())->val('job', 'old_failure_count', 0);
    my $after = defined($self->{successive_failure_after})
        ? $self->{successive_failure_after}
        : ($status ||= $self->_get_status())->val('job', 'successive_failure_count', 0);
    return(
        $before, $after, defined($after) ? $after != $before : undef
    );
}

=head2 sync

Perform the synchronization

=cut

sub sync {
    my ($self) = @_;
    $self->_log('INFO', 'Start to process' );
    $self->_log('DEBUG', 'goes into %s%s',
        $self->dest,
        $self->{options}{dryrun} ? ' (dryrun mode)' : '',
    );

    my $oldname = $0;
    $0 = 'mmm [' . $self->name . ']';

    $self->getlock() or return;

    my $status = $self->_get_status();
    $self->_set_compute_config_sum($status);
    my ($ouid, $ogid) = MMM::Utils::setid( $self->val('user'), $self->val('group') );

    $self->{successive_failure_before} =
        $status->val('job', 'successive_failure_count', 0);
    $status->newval('job', 'old_failure_count', $self->{successive_failure_before});
    $status->delval( 'job', 'command');
    $status->newval('job', 'processed_count',
        $status->val('job', 'processed_count', 0) + 1
    );

    if (!defined($status->val('job', 'first_sync'))) {
        _set_status_time($status, 'job', 'first_sync', scalar( time() ));
    }

    my $res = 0;

    if (! -d $self->dest) {
        push(@{ $self->{output} }, sprintf('Directory %s does not exists (%s)',
            $self->dest,
            $self->name,
        ));
        foreach (qw(start end)) {
            _set_status_time($status, 'job', $_, scalar( time() ));
        }
        return $res;
    }

    if ($self->val('pre_exec')) {
        my @cmd = ($self->val('pre_exec'), $self->name, $self->dest);
        $self->_log('INFO', 'Executing PRE: %s',
            join(' ', map { qq{"$_"} } (@cmd)));
        if (system(@cmd) != 0) {
            if ($? == -1) {
                $self->_log('ERROR', 'failed to execute pre_exec: %s', $!);
            } else {
                $self->_log('ERROR',
                    'Pre_exec exited with value %d, abborting sync', $? >> 8);
            }
            return $res;
        }
    }

    if (my $url = $self->val('url')) {
        $res = $self->_sync_url(
            $status, $url,
            password => $self->val('password') || undef,
            use_ssh => yes_no($self->val('rsync_use_ssh')),
        );
    } else {
        $self->_log('ERROR', 'No source or url' );
        return $res
    }

    $status->newval('job', 'success', $res ? 1 : 0);
    if ($res) {
        $self->_log('NOTICE', 'Sync done%s from %s',
            $self->{options}{dryrun} ? ' (dryrun mode)' : '',
            $status->val('success', 'sync_from'),
        );
        $status->newval('job', 'successive_failure_count', 0);
    } else {
        $self->_log('WARNING', 'Unable to sync');
        $status->newval('job', 'successive_failure_count',
            $status->val('job', 'successive_failure_count', 0) + 1
        );
        foreach (@{ $self->{output} ? $self->{output} : [ "No output from process" ]}) {
            $self->_log('ERROR', $_);
        }
    }
    $self->{successive_failure_after} =
        $status->val('job', 'successive_failure_count', 0);

    if ($self->val('post_exec')) {
        $ENV{MMM_RESULT} = $res;
        if ($res) {
            $ENV{MMM_FROM} = $status->val('success', 'sync_from');
            $ENV{MMM_MIRROR} = $status->val('job', 'try_from');
        }
        my @cmd = ($self->val('post_exec'), $self->name, $self->dest);
        $self->_log('INFO', 'Executing POST: %s',
            join(' ', map { qq{"$_"} } (@cmd)));
        if (system(@cmd) != 0) {
            if ($? == -1) {
                $self->_log('WARNING', 'failed to execute post_exec: %s', $!);
            } else {
                $self->_log('WARNING',
                    'Post_exec exited with value %d, abborting sync', $? >> 8);
            }
        }
    }

    if (yes_no($self->val('compute_size', 0)) &&
        scalar(time) > $status->val('job', 'size_time', 0) +
        duration2m($self->val('size_delay', SIZE_DELAY) * 60)) {
        $self->du_dest($status);
    }

    MMM::Utils::setid($ouid, $ogid);

    $self->_write_status($status) unless($self->{options}{dryrun});

    if ($self->{mmm}->can('send_mail')) {
        if ($status->val('job', 'old_failure_count', 0) !=
            $status->val('job', 'successive_failure_count', 0) &&
            grep { $status->val('job', 'successive_failure_count', 0) == $_ }
            (0, $self->val('errors_mail', 3))
        ) {

            $self->{mmm}->body_queue($self, $self, $self->state_info($status));
            $self->{mmm}->send_mail();
        }
    }

    $self->unlock();
    $0 = $oldname;
    $res
}

sub _sync_url {
    my ($self, $status, $based_url, %options) = @_;

    my $url = $based_url;
    $url =~ m:/$: or $url .= '/';
    $url .= '/' . $self->val('subdir') if ($self->val('subdir'));
    $url =~ m:/$: or $url .= '/';

    $self->_log('DEBUG', 'Try from mirror %s', $url);
    _set_status_time($status, 'job', 'start', scalar( time() ));

    foreach my $val (
        'bwlimit',              # bandwidth limit in k
        'timeout',              # timeout
        'rsync_opts',           # specifics rsync options
        'rsync_defaults',       # defaults rsync options
        'exclude',              # excluded files/dir
        'tempdir',
        'partialdir',
        ) {
        if (my $v = $self->val($val)) {
            $options{$val} = $v;
        }
    }
    foreach my $val (
        'delete-after',         # deleting after ?
        'delete',               # delete removed files ?
        'delete-excluded',      # deleting excluded files ?
        ) {
        $options{$val} = yes_no($self->val($val));
    }
    my $sync = MMM::Sync->new(
        $url,
        $self->dest,
        %options,
    );

    if (my $m = MMM::Mirror->new(url => $url)) {
        $status->newval('job', 'try_from', $m->host);
    }

    my $sync_res;
    my $max_try = $self->val( 'max_try', MAX_TRY );
    $self->_log('DEBUG', 'running %s', join(' ', $sync->buildcmd()));
    foreach my $trycount (1 .. $max_try) {
        $sync->reset_output;
        if ( $self->{options}{dryrun} ) {
            $sync_res = 0;
            sleep(10);
        } else {
            $sync_res = $sync->sync();
        }
        $self->_log($sync_res ? 'WARNING' : 'DEBUG',
            'Try %d/%d, res: %d from mirror %s',
            $trycount, $max_try, $sync_res,
            $sync->{source}, # TODO: kill intrusive var access
        );
        $self->{output} = $sync->get_output;
        if ($sync_res != 1) {
            last
        }
    }

    $status->newval( 'job', 'command', join(' ', $sync->buildcmd) );

    _set_status_time($status, 'job', 'end', scalar( time() ) );

    my $concerned_section = $sync_res ? 'failure' : 'success';

    foreach (qw(start end)) {
        _set_status_time($status, $concerned_section, $_,
            $status->val('job', $_)
        );
    }
    $status->newval($concerned_section, 'url', $url);
    $status->newval($concerned_section, 'try_from', $status->val('job', 'try_from'));

    if ($sync_res == 0) {
        $status->delval( 'job', 'error_log' );
        $status->newval('success', 'sync_from', $status->val('job', 'try_from'));
    } else {
        if (@{ $self->{output} || []}) {
            $status->newval('job', 'error_log',
                (@{ $self->{output}} > 10)
                ? (@{$self->{output}}[-9 .. -1], '...')
                : (@{$self->{output}})
            );
        }
        return 0;
    }
}

=head2 dest

The destination directory for this mirror

=cut

sub dest {
    my ($self) = @_;
    return $self->val('path', $self->name);
}

=head2 du_dest

Perform du over the destination directory and store the result into status file

=cut

sub du_dest {
    my ($self, $status) = @_;
    $status ||= $self->_get_status();
    $self->_log('DEBUG', 'Calculating size of %s', $self->dest);
    $self->getlock() or return;
    if (! -d $self->dest) { return }
    open(my $handle, sprintf('\\du -s %s |', $self->dest)) or return;
    my $line = <$handle>;
    if ($line && $line =~ /^(\d+)/) {
        $status->newval('job', 'size', $1);
        $status->newval('job', 'size_time', scalar(time));
    }
    close($handle);
    $self->_write_status($status);
    $self->unlock;
    return 1;
}

=head1 STATUS FILE STRUCTURE

=head2 job SECTION

=over 4

=item successed 1 if last run was ok

=item size The size of the tree

=item size_time Last time the size were checked

=item first_sync

The first time job is take into account

=back

=head2 success SECTION

=item start

=item end

=head2 failure SECTION

=cut

1;