File Coverage

blib/lib/File/Write/Rotate.pm

Criterion	Covered	Total	%
statement	214	232	92.2
branch	96	130	73.8
condition	33	43	76.7
subroutine	21	22	95.4
pod	7	7	100.0
total	371	434	85.4

package File::Write::Rotate;

our $DATE = '2016-10-07'; # DATE
our $VERSION = '0.31'; # VERSION

use 5.010001;
use strict;
use warnings;

# we must not use Log::Any, looping if we are used as log output
#use Log::Any '$log';

use File::Spec;
use IO::Compress::Gzip qw(gzip $GzipError);
use Scalar::Util qw(weaken);
#use Taint::Runtime qw(untaint is_tainted);
use Time::HiRes 'time';

our $Debug;

sub new {
    my $class = shift;
    my %args0 = @_;

    my %args;

    defined($args{dir} = delete $args0{dir})
        or die "Please specify dir";
    defined($args{prefix} = delete $args0{prefix})
        or die "Please specify prefix";
    $args{suffix} = delete($args0{suffix}) // "";

    $args{size} = delete($args0{size}) // 0;

    $args{period} = delete($args0{period});
    if ($args{period}) {
        $args{period} =~ /\A(daily|day|month|monthly|year|yearly)\z/
            or die "Invalid period, please use daily/monthly/yearly";
    }

    for (map {"hook_$_"} qw(before_rotate after_rotate after_create
                            before_write a)) {
        next unless $args0{$_};
        $args{$_} = delete($args0{$_});
        die "Invalid $_, please supply a coderef"
            unless ref($args{$_}) eq 'CODE';
    }

    if (!$args{period} && !$args{size}) {
        $args{size} = 10 * 1024 * 1024;
    }

    $args{histories} = delete($args0{histories}) // 10;

    $args{binmode} = delete($args0{binmode});

    $args{buffer_size} = delete($args0{buffer_size});

    $args{lock_mode} = delete($args0{lock_mode}) // 'write';
    $args{lock_mode} =~ /\A(none|write|exclusive)\z/
        or die "Invalid lock_mode, please use none/write/exclusive";

    $args{rotate_probability} = delete($args0{rotate_probability});
    if (defined $args{rotate_probability}) {
        $args{rotate_probability} > 0 && $args{rotate_probability} < 1.0
            or die "Invalid rotate_probability, must be 0 < x < 1";
    }

    if (keys %args0) {
        die "Unknown arguments to new(): ".join(", ", sort keys %args0);
    }

    $args{_buffer} = [];

    my $self = bless \%args, $class;

    $self->{_exclusive_lock} = $self->_get_lock
        if $self->{lock_mode} eq 'exclusive';

    $self;
}

sub buffer_size {
    my $self = shift;
    if (@_) {
        my $old = $self->{buffer_size};
        $self->{buffer_size} = $_[0];
        return $old;
    } else {
        return $self->{buffer_size};
    }
}

sub handle {
    my $self = shift;
    $self->{_fh};
}

sub path {
    my $self = shift;
    $self->{_fp};
}

# file path, without the rotate suffix
sub _file_path {
    my ($self) = @_;

    # _now is calculated every time we access this method
    $self->{_now} = time();

    my @lt = localtime($self->{_now});
    $lt[5] += 1900;
    $lt[4]++;

    my $period;
    if ($self->{period}) {
        if ($self->{period} =~ /year/i) {
            $period = sprintf("%04d", $lt[5]);
        } elsif ($self->{period} =~ /month/) {
            $period = sprintf("%04d-%02d", $lt[5], $lt[4]);
        } elsif ($self->{period} =~ /day|daily/) {
            $period = sprintf("%04d-%02d-%02d", $lt[5], $lt[4], $lt[3]);
        }
    } else {
        $period = "";
    }

    my $path = join(
        '',
        $self->{dir}, '/',
        $self->{prefix},
        length($period) ? ".$period" : "",
        $self->{suffix},
    );
    if (wantarray) {
        return ($path, $period);
    } else {
        return $path;
    }
}

sub lock_file_path {
    my ($self) = @_;
    return File::Spec->catfile($self->{dir}, $self->{prefix} . '.lck');
}

sub _get_lock {
    my ($self) = @_;
    return undef if $self->{lock_mode} eq 'none';
    return $self->{_weak_lock} if defined($self->{_weak_lock});

    require File::Flock::Retry;
    my $lock = File::Flock::Retry->lock($self->lock_file_path);
    $self->{_weak_lock} = $lock;
    weaken $self->{_weak_lock};
    return $lock;
}

# will return \@files. each entry is [filename without compress suffix,
# rotate_suffix (for sorting), period (for sorting), compress suffix (for
# renaming back)]
sub _get_files {
    my ($self) = @_;

    opendir my ($dh), $self->{dir} or do {
        warn "Can't opendir '$self->{dir}': $!";
        return;
    };

    my @files;
    while (my $e = readdir($dh)) {
        my $cs = $1 if $e =~ s/(\.gz)\z//; # compress suffix
        next unless $e =~ /\A\Q$self->{prefix}\E
                           (?:\. (?\d{4}(?:-\d\d(?:-\d\d)?)?) )?
                           \Q$self->{suffix}\E
                           (?:\. (?\d+) )?
                           \z
                          /x;
        push @files,
          [ $e, $+{rotate_suffix} // 0, $+{period} // "", $cs // "" ];
    }
    closedir($dh);

    [ sort { $a->[2] cmp $b->[2] || $b->[1] <=> $a->[1] } @files ];
}

# rename (increase rotation suffix) and keep only n histories. note: failure in
# rotating should not be fatal, we just warn and return.
sub _rotate_and_delete {
    my ($self, %opts) = @_;

    my $delete_only = $opts{delete_only};
    my $lock = $self->_get_lock;
  CASE:
    {
        my $files = $self->_get_files or last CASE;

        # is there a compression process in progress? this is marked by the
        # existence of -compress.pid PID file.
        #
        # XXX check validity of PID file, otherwise a stale PID file will always
        # prevent rotation to be done
        if (-f "$self->{dir}/$self->{prefix}-compress.pid") {
            warn "Compression is in progress, rotation is postponed";
            last CASE;
        }

        $self->{hook_before_rotate}->($self, [map {$_->[0]} @$files])
            if $self->{hook_before_rotate};

        my @deleted;
        my @renamed;

        my $i;
        my $dir = $self->{dir};
        my $rotating_period = @$files ? $files->[-1][2] : undef;
        for my $f (@$files) {
            my ($orig, $rs, $period, $cs) = @$f;
            $i++;

            #say "DEBUG: is_tainted \$dir? ".is_tainted($dir);
            #say "DEBUG: is_tainted \$orig? ".is_tainted($orig);
            #say "DEBUG: is_tainted \$cs? ".is_tainted($cs);

            # TODO actually, it's more proper to taint near the source (in this
            # case, _get_files)
            #untaint \$orig;
            ($orig) = $orig =~ /(.*)/s; # we use this instead, no module needed

            if ($i <= @$files - $self->{histories}) {
                say "DEBUG: Deleting old rotated file $dir/$orig$cs ..."
                    if $Debug;
                if (unlink "$dir/$orig$cs") {
                    push @deleted, "$orig$cs";
                } else {
                    warn "Can't delete $dir/$orig$cs: $!";
                }
                next;
            }
            if (!$delete_only && defined($rotating_period) && $period eq $rotating_period) {
                my $new = $orig;
                if ($rs) {
                    $new =~ s/\.(\d+)\z/"." . ($1+1)/e;
                } else {
                    $new .= ".1";
                }
                if ($new ne $orig) {
                    say "DEBUG: Renaming rotated file $dir/$orig$cs -> ".
                        "$dir/$new$cs ..." if $Debug;
                    if (rename "$dir/$orig$cs", "$dir/$new$cs") {
                        push @renamed, "$new$cs";
                    } else {
                        warn "Can't rename '$dir/$orig$cs' -> '$dir/$new$cs': $!";
                    }
                }
            }
        }

        $self->{hook_after_rotate}->($self, \@renamed, \@deleted)
            if $self->{hook_after_rotate};
    } # CASE
}

sub _open {
    my $self = shift;

    my ($fp, $period) = $self->_file_path;
    open $self->{_fh}, ">>", $fp or die "Can't open '$fp': $!";
    if (defined $self->{binmode}) {
        binmode $self->{_fh}, $self->{binmode}
          or die "Can't set PerlIO layer on '$fp': $!";
    }
    my $oldfh = select $self->{_fh};
    $| = 1;
    select $oldfh;    # set autoflush
    $self->{_fp} = $fp;
    $self->{hook_after_create}->($self) if $self->{hook_after_create};
}

# (re)open file and optionally rotate if necessary
sub _rotate_and_open {

    my $self = shift;
    my ($do_open, $do_rotate) = @_;
    my $fp;
    my %rotate_params;

  CASE:
    {
        # if instructed, only do rotate some of the time to shave overhead
        if ($self->{rotate_probability} && $self->{_fh}) {
            last CASE if rand() > $self->{rotate_probability};
        }

        $fp = $self->_file_path;
        unless (-e $fp) {
            $do_open++;
            $do_rotate++;
            $rotate_params{delete_only} = 1;
            last CASE;
        }

        # file is not opened yet, open
        unless ($self->{_fh}) {
            $self->_open;
        }

        # period has changed, rotate
        if ($self->{_fp} ne $fp) {
            $do_rotate++;
            $rotate_params{delete_only} = 1;
            last CASE;
        }

        # check whether size has been exceeded
        my $inode;

        if ($self->{size} > 0) {

            my @st   = stat($self->{_fh});
            my $size = $st[7];
            $inode = $st[1];

            if ($size >= $self->{size}) {
                say "DEBUG: Size of $self->{_fp} is $size, exceeds $self->{size}, rotating ..."
                    if $Debug;
                $do_rotate++;
                last CASE;
            } else {
                # stat the current file (not our handle _fp)
                my @st = stat($fp);
                die "Can't stat '$fp': $!" unless @st;
                my $finode = $st[1];

                # check whether other process has rename/rotate under us (for
                # example, 'prefix' has been moved to 'prefix.1'), in which case
                # we need to reopen
                if (defined($inode) && $finode != $inode) {
                    $do_open++;
                }
            }

        }
    } # CASE

    $self->_rotate_and_delete(%rotate_params) if $do_rotate;
    $self->_open if $do_rotate || $do_open;    # (re)open
}

sub write {
    my $self = shift;

    # the buffering implementation is currently pretty naive. it assume any
    # die() as a write failure and store the message to buffer.

    # FYI: if privilege is dropped from superuser, the failure is usually at
    # locking the lock file (permission denied).

    my @msg = (map( {@$_} @{ $self->{_buffer} } ), @_);

    eval {
        my $lock = $self->_get_lock;

        $self->_rotate_and_open;

        $self->{hook_before_write}->($self, \@msg, $self->{_fh})
            if $self->{hook_before_write};

        print { $self->{_fh} } @msg;
        $self->{_buffer} = [];

    };
    my $err = $@;

    if ($err) {
        if (($self->{buffer_size} // 0) > @{ $self->{_buffer} }) {
            # put message to buffer temporarily
            push @{ $self->{_buffer} }, [@_];
        } else {
            # buffer is already full, let's dump the buffered + current message
            # to the die message anyway.
            die join(
                "",
                "Can't write",
                (
                    @{ $self->{_buffer} }
                    ? " (buffer is full, "
                      . scalar(@{ $self->{_buffer} })
                      . " message(s))"
                    : ""
                ),
                ": $err, message(s)=",
                @msg
            );
        }
    }
}

sub compress {
    my ($self) = shift;

    my $lock           = $self->_get_lock;
    my $files_ref        = $self->_get_files;
    my $done_compression = 0;

    if (@{$files_ref}) {
        require Proc::PID::File;

        my $pid = Proc::PID::File->new(
            dir    => $self->{dir},
            name   => "$self->{prefix}-compress",
            verify => 1,
        );
        my $latest_period = $files_ref->[-1][2];

        if ($pid->alive) {
            warn "Another compression is in progress";
        } else {
            my @tocompress;
            #use DD; dd $self;
            for my $file_ref (@{$files_ref}) {
                my ($orig, $rs, $period, $cs) = @{ $file_ref };
                #say "D:compress: orig=<$orig> rs=<$rs> period=<$period> cs=<$cs>";
                next if $cs; # already compressed
                next if !$self->{period} && !$rs; # not old file
                next if  $self->{period} && $period eq $latest_period; # not old file
                push @tocompress, File::Spec->catfile($self->{dir}, $orig);
            }

            if (@tocompress) {
                for my $file (@tocompress) {
                    gzip($file => "$file.gz")
                        or do { warn "gzip failed: $GzipError\n"; next };
                    unlink $file;
                }
                $done_compression = 1;
            }
        }
    }

    return $done_compression;

}

sub DESTROY {
    my ($self) = @_;

    # Proc::PID::File's DESTROY seem to create an empty PID file, remove it.
    unlink "$self->{dir}/$self->{prefix}-compress.pid";
}

1;

# ABSTRACT: Write to files that archive/rotate themselves

__END__