File Coverage

blib/lib/Dancer2/Logger/Fluent.pm

Criterion	Covered	Total	%
statement	37	43	86.0
branch	4	18	22.2
condition	2	17	11.7
subroutine	14	15	93.3
pod			n/a
total	57	93	61.2

package Dancer2::Logger::Fluent;

use strict;
use 5.008_005;
our $VERSION = '0.06';

use Moo;
use Fluent::Logger;
use IO::Socket::INET;
use IO::Socket::UNIX;

use File::Basename 'basename';
use Time::Moment;
use Sys::Hostname;

use Dancer2::Core::Types;
with 'Dancer2::Core::Role::Logger';

has tag_prefix => (
    is        => 'lazy',
    isa       => Str,
);

has host => (
    is        => 'ro',
    isa       => Str,
);

has port => (
    is        => 'ro',
    isa       => Str,
);

has timeout => (
    is        => 'ro',
    isa       => Num,
    predicate => '_has_timeout',
);

has socket => (
    is        => 'ro',
    isa       => Str,
);

has prefer_integer => (
    is        => 'ro',
    isa       => Str,
    predicate => '_has_prefer_integer',
);

has event_time => (
    is        => 'ro',
    isa       => Bool,
);

has buffer_limit => (
    is        => 'ro',
    isa       => Int,
    predicate => '_has_buffer_limit',
);

has buffer_overflow_handler => (
    is        => 'ro',
    isa       => CodeRef,
    predicate => '_has_buffer_overflow_handler',
);

has truncate_buffer_at_overflow => (
    is        => 'ro',
    isa       => Bool,
);

sub _build_tag_prefix {
    my $self = shift;
    return $self->{tag_prefix} || $self->app_name || $ENV{DANCER_APPDIR} || basename($0);
}

sub _connect {
    my $self = shift;
    return defined $self->socket
        ? IO::Socket::UNIX->new( Peer => $self->socket )
        : IO::Socket::INET->new(
            PeerAddr  => $self->host || '127.0.0.1',
            PeerPort  => $self->port || 24224,
            Proto     => 'tcp',
            Timeout   => $self->_has_timeout ? $self->timeout : 3.0,
            ReuseAddr => 1,
    );
}

sub _fluent {
    my $self = shift;

    return unless $self->_connect;

    unless ( exists $self->{_fluent} ) {
        $self->{_fluent} = Fluent::Logger->new(
            host                        => $self->host || '127.0.0.1',
            port                        => $self->port || 24224,
            timeout                     => $self->_has_timeout ? $self->timeout : 3.0,
            socket                      => $self->socket,
            prefer_integer              => $self->_has_prefer_integer ? $self->prefer_integer : 1,
            event_time                  => $self->event_time || 0,
            buffer_limit                => $self->_has_buffer_limit ? $self->buffer_limit : 8388608,
            buffer_overflow_handler     => $self->_has_buffer_overflow_handler ? $self->buffer_overflow_handler : sub { undef },
            truncate_buffer_at_overflow => $self->truncate_buffer_at_overflow || 0,
        );
    }

    return $self->{_fluent};
}

sub DESTROY {
    my $self = shift;
    return unless $self->_fluent;
    $self->_fluent->{pending} ||= '';  # Fluent::Logger->close performs length checks without checking if value is defined first
    $self->_fluent->close;
}

sub log {
    my ($self, $level, $message) = @_;

    my $fluent_message = {
        env       => $ENV{DANCER_ENVIRONMENT} || $ENV{PLACK_ENV} || 'development',
        timestamp => Time::Moment->now_utc->strftime("%Y-%m-%dT%H:%M:%S.%6N%Z"),
        host      => hostname(),
        level     => $level,
        message   => $message,
        pid       => $$
    };

    # Queue pending messages until connectivity is restored
    unless ( $self->_fluent ) {
        push @{ $self->{pending} }, $fluent_message;
        return;
    }

    if ( exists $self->{pending} and @{ $self->{pending} } ) {
        while ( my $pending_message = shift @{ $self->{pending} } ) {
            $self->_fluent->post( $self->tag_prefix, $pending_message );
        }
    }
    $self->_fluent->post( $self->tag_prefix, $fluent_message );
}

1;
__END__