File Coverage

blib/lib/Fluent/Logger.pm

Criterion	Covered	Total	%
statement	192	251	76.4
branch	43	78	55.1
condition	9	25	36.0
subroutine	35	40	87.5
pod	4	6	66.6
total	283	400	70.7

# -*- coding: utf-8; -*-
package Fluent::Logger;

use strict;
use warnings;

our $VERSION = '0.25';

use IO::Select;
use IO::Socket::INET;
use IO::Socket::UNIX;
use Data::MessagePack;
use Time::Piece;
use Carp;
use Scalar::Util qw/ refaddr /;
use Time::HiRes qw/ time /;
use UUID::Tiny qw/ create_uuid UUID_V4 /;
use MIME::Base64 qw/ encode_base64 /;

use constant RECONNECT_WAIT           => 0.5;
use constant RECONNECT_WAIT_INCR_RATE => 1.5;
use constant RECONNECT_WAIT_MAX       => 60;
use constant RECONNECT_WAIT_MAX_COUNT => 12;

use constant MP_HEADER_3ELM_ARRAY => "\x93";
use constant MP_HEADER_4ELM_ARRAY => "\x94";
use constant MP_HEADER_EVENT_TIME => "\xd7\x00";

use subs 'prefer_integer';

use Class::Tiny +{
    tag_prefix => sub {},
    host => sub { "127.0.0.1" },
    port => sub { 24224 },
    socket => sub {},
    timeout => sub { 3.0 },
    buffer_limit => sub { 8 * 1024 * 1024 }, # fixme
    buffer_overflow_handler => sub { undef },
    truncate_buffer_at_overflow => sub { 0 },
    max_write_retry => sub { 5 },
    write_length => sub { 8 * 1024 * 1024 },
    socket_io => sub {},
    errors => sub { [] },
    prefer_integer => sub { 1 },
    packer => sub {
        my $self = shift;
        my $mp   = Data::MessagePack->new;
        $mp->prefer_integer( $self->prefer_integer );
        $mp;
    },
    pending => sub { "" },
    connect_error_history => sub { +[] },
    owner_pid => sub {},
    event_time => sub { 0 },
    ack => sub { 0 },
    pending_acks => sub { +[] },
    unpacker => sub {
        require Data::MessagePack::Stream;
        Data::MessagePack::Stream->new;
    },
    selector => sub { },
};

sub BUILD {
    my $self = shift;
    $self->_connect;
}

sub prefer_integer {
    my $self = shift;

    if (@_) {
        $self->{prefer_integer} = shift;
        $self->packer->prefer_integer( $self->prefer_integer );
    } elsif ( exists $self->{prefer_integer} ) {
        return $self->{prefer_integer};
    } else {
        my $defaults = Class::Tiny->get_all_attribute_defaults_for( ref $self );
        return $self->{prefer_integer} = $defaults->{prefer_integer}->();
    }
}

sub _carp {
    my $self = shift;
    my $msg  = shift;
    chomp $msg;
    carp(
        sprintf "%s %s[%s](%s): %s",
        localtime->strftime("%Y-%m-%dT%H:%M:%S%z"),
        ref $self,
        refaddr $self,
        $self->_connect_info,
        $msg,
    );
}

sub _add_error {
    my $self = shift;
    my $msg  = shift;
    $self->_carp($msg);
    push @{ $self->errors }, $msg;
}

sub errstr {
    my $self = shift;
    return join ("\n", @{ $self->errors });
}

sub _connect_info {
    my $self = shift;
    $self->socket || sprintf "%s:%d", $self->host, $self->port;
}

sub _connect {
    my $self  = shift;
    my $force = shift;

    return if $self->socket_io && !$force;

    my $sock = defined $self->socket
             ? IO::Socket::UNIX->new( Peer => $self->socket )
             : IO::Socket::INET->new(
                 PeerAddr  => $self->host,
                 PeerPort  => $self->port,
                 Proto     => 'tcp',
                 Timeout   => $self->timeout,
                 ReuseAddr => 1,
             );
    if (!$sock) {
        $self->_add_error("Can't connect: $!");
        push @{ $self->connect_error_history }, time;
        if (@{ $self->connect_error_history } > RECONNECT_WAIT_MAX_COUNT) {
            shift @{ $self->connect_error_history };
        }
        return;
    }
    $self->connect_error_history([]);
    $self->owner_pid($$);
    $self->selector(IO::Select->new($sock));
    $self->socket_io($sock);
}

sub close {
    my $self = shift;

    if ( length $self->{pending} ) {
        $self->_carp("flushing pending data on close");
        $self->_connect unless $self->socket_io;
        my $written = eval {
            $self->_write( $self->{pending} );
        };
        if ($@ || !$written) {
            my $size = length $self->{pending};
            $self->_carp("Can't send pending data. LOST $size bytes.: $@");
            $self->_call_buffer_overflow_handler();
        } else {
            $self->_carp("pending data was flushed successfully");
        }
    };
    $self->{pending} = "";
    $self->{pending_acks} = [];
    delete $self->{selector};
    my $socket = delete $self->{socket_io};
    $socket->close if $socket;
}

sub post {
    my($self, $tag, $msg) = @_;

    $self->_post( $tag || "", $msg, time() );
}

sub post_with_time {
    my ($self, $tag, $msg, $time) = @_;

    $self->_post( $tag || "", $msg, $time );
}

sub _pack_time {
    my ($self, $time) = @_;

    if ($self->event_time) {
        my $time_i  = int $time;
        my $nanosec = int(($time - $time_i) * 10 ** 9);
        return MP_HEADER_EVENT_TIME . pack("NN", $time_i, $nanosec);
    } else {
        return $self->packer->pack(int $time);
    }
}

sub _post {
    my ($self, $tag, $msg, $time) = @_;

    if (ref $msg ne "HASH") {
        $self->_add_error("message '$msg' must be a HashRef");
        return;
    }

    $tag = join('.', $self->tag_prefix, $tag) if $self->tag_prefix;
    my $p = $self->packer;
    $self->_send(
        $p->pack($tag),
        $self->_pack_time($time),
        $p->pack($msg),
    );
}

sub _send {
    my ($self, @args) = @_;

    my ($data, $unique_key);
    if ( $self->ack ) {
        $unique_key = encode_base64(create_uuid(UUID_V4));
        $data = join('', MP_HEADER_4ELM_ARRAY, @args, $self->{packer}->pack({ chunk => $unique_key }));
        push @{$self->{pending_acks}}, $unique_key;
    } else {
        $data = join('', MP_HEADER_3ELM_ARRAY, @args);
    }

    my $prev_size = length($self->{pending});
    my $current_size = length($data);
    $self->{pending} .= $data;

    my $errors = @{ $self->connect_error_history };
    if ( $errors && length $self->pending <= $self->buffer_limit )
    {
        my $suppress_sec;
        if ( $errors < RECONNECT_WAIT_MAX_COUNT ) {
            $suppress_sec = RECONNECT_WAIT * (RECONNECT_WAIT_INCR_RATE ** ($errors - 1));
        } else {
            $suppress_sec = RECONNECT_WAIT_MAX;
        }
        if ( time - $self->connect_error_history->[-1] < $suppress_sec ) {
            return;
        }
    }

    # fork safe
    if (!$self->socket_io || $self->owner_pid != $$) {
        $self->_connect(1);
    }

    my $written;
    eval {
        $written = $self->_write( $self->{pending} );
        my $acked = $self->ack
            ? $self->_wait_ack(@{ $self->{pending_acks} })
            : 1;
        if ($written && $acked) {
            $self->{pending} = "";
            $self->{pending_acks} = [];
        }
    };
    if ($@) {
        my $error = $@;
        $self->_add_error("Cannot send data: $error");
        my $sock = delete $self->{socket_io};
        $sock->close if $sock;
        delete $self->{selector};
        if ( length($self->{pending}) > $self->buffer_limit ) {
            if ( defined $self->buffer_overflow_handler ) {
                $self->_call_buffer_overflow_handler();
                $self->{pending} = "";
                $self->{pending_acks} = [] if $self->ack;
            } elsif ( $self->truncate_buffer_at_overflow ) {
                substr($self->{pending}, $prev_size, $current_size, "");
                pop @{$self->{pending_acks}} if $self->ack;
            }
        }
        return;
    }

    return $written;
}

sub _wait_ack {
    my $self = shift;
    my @acks = @_;

    my $up = $self->unpacker;
    local $SIG{"PIPE"} = sub { die $! };
READ:
    while (1) {
        my ($s) = $self->selector->can_read($self->timeout);
        if (!$s) {
            die "ack read timed out";
        }
        $s->sysread(my $buf, 1024);
        return if @acks > 0 && length($buf) == 0;
        $up->feed($buf);
        while ($up->next) {
            my $ack = $up->data;
            my $unique_key = shift @acks;
            if ($unique_key && ref $ack eq "HASH") {
                if ($ack->{ack} ne $unique_key) {
                    die "ack is not expected: " . $ack->{ack};
                }
            } else {
                unshift @{ $self->{pending_acks} }, $unique_key;
                die "Can't send data. ack is not expected. $@";
            }
            last READ if @acks == 0;
        }
    }
    return 1;
}

sub _call_buffer_overflow_handler {
    my $self = shift;
    if (my $handler = $self->buffer_overflow_handler) {
        eval {
            $handler->($self->{pending});
        };
        if (my $error = $@) {
            $self->_add_error("Can't call buffer overflow handler: $error");
        }
    }
}

sub _write {
    my $self = shift;
    my $data = shift;
    my $length = length($data);
    my $retry  = my $written = 0;
    die "Connection is not available" unless $self->socket_io;

    local $SIG{"PIPE"} = sub { die $! };

    while ($written < $length) {
        my ($s) = $self->selector->can_write($self->timeout);
        die "send write timed out" unless $s;
        my $nwrite
            = $s->syswrite($data, $self->write_length, $written);

        if (!$nwrite) {
            if ($retry > $self->max_write_retry) {
                die "failed write retry; max write retry count. $!";
            }
            $retry++;
        } else {
            $written += $nwrite;
        }
    }
    $written;
}

sub DEMOLISH {
    my $self = shift;
    $self->close;
}


1;
__END__