File Coverage

lib/Beekeeper/Service/LogTail/Worker.pm

Criterion	Covered	Total	%
statement	21	106	19.8
branch	0	34	0.0
condition	0	41	0.0
subroutine	7	19	36.8
pod	3	4	75.0
total	31	204	15.2

package Beekeeper::Service::LogTail::Worker;

use strict;
use warnings;

our $VERSION = '0.10';

use Beekeeper::Worker ':log';
use base 'Beekeeper::Worker';

use Beekeeper::Logger ':log_levels';
use Scalar::Util 'weaken';
use JSON::XS;

my @Log_buffer;


sub authorize_request {
    my ($self, $req) = @_;

    return unless $self->__has_authorization_token('BKPR_ADMIN');

    return BKPR_REQUEST_AUTHORIZED;
}

sub on_startup {
    my $self = shift;
    weaken $self;

    $self->{max_entries} = $self->{config}->{buffer_entries} || 20000;
    $self->{log_level}   = $self->{config}->{log_level}      || LOG_DEBUG;

    $self->_connect_to_all_brokers;

    $self->accept_remote_calls(
        '_bkpr.logtail.tail' => 'tail',
    );

    # Ping backend brokers to avoid disconnections due to inactivity
    $self->{ping_timer} = AnyEvent->timer(
        after    => 60 * rand(),
        interval => 60,
        cb       => sub { $self->_ping_backend_brokers },
    );

    log_info "Ready";
}

sub _connect_to_all_brokers {
    my $self = shift;
    weaken $self;

    my $own_bus = $self->{_BUS};
    my $group_config = Beekeeper::Config->get_bus_group_config( bus_id => $own_bus->bus_id );

    $self->{_BUS_GROUP} = [];

    foreach my $config (@$group_config) {

        my $bus_id = $config->{'bus_id'};

        if ($bus_id eq $own_bus->bus_id) {
            # Already connected to our own bus
            $self->_collect_log($own_bus);
            next;
        }

        my $bus; $bus = Beekeeper::MQTT->new(
            %$config,
            bus_id   => $bus_id,
            timeout  => 300,
            on_error => sub {
                # Reconnect
                my $errmsg = $_[0] || ""; $errmsg =~ s/\s+/ /sg;
                log_error "Connection to $bus_id failed: $errmsg";
                my $delay = $self->{connect_err}->{$bus_id}++;
                $self->{reconnect_tmr}->{$bus_id} = AnyEvent->timer(
                    after => ($delay < 10 ? $delay * 3 : 30),
                    cb => sub {
                        $bus->connect(
                            on_connack => sub {
                                # Setup subscriptions
                                log_warn "Reconnected to $bus_id";
                                $self->_collect_log($bus);
                            }
                        );
                    },
                );
            },
        );

        push @{$self->{_BUS_GROUP}}, $bus;

        $bus->connect(
            on_connack => sub {
                # Setup subscriptions
                $self->_collect_log($bus);
            }
        );
    }
}

sub _collect_log {
    my ($self, $bus) = @_;

    # Default logger logs to topic log/$level/$service

    my $max_entries = $self->{max_entries};
    my $log_level   = $self->{log_level};
    my $worker      = $self->{_WORKER};

    foreach my $level (1..$log_level) {

        my $topic = "log/$level/#";
        my $req;

        $bus->subscribe(
            topic      => $topic,
            on_publish => sub {
              # my ($payload_ref, $mqtt_properties) = @_;

                $req = decode_json( ${$_[0]} );

                push @Log_buffer, $req->{params};

                shift @Log_buffer if (@Log_buffer > $max_entries);

                # Track number of collected log entries
                $worker->{notif_count}++;
            },
            on_suback => sub {
                my ($success, $prop) = @_;
                die "Could not subscribe to log topic '$topic'" unless $success;
            },
        );
    }
}

sub _ping_backend_brokers {
    my $self = shift;

    foreach my $bus (@{$self->{_BUS_GROUP}}) {

        next unless $bus->{is_connected};
        $bus->pingreq;
    }
}

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

    foreach my $bus (@{$self->{_BUS_GROUP}}) {

        next unless $bus->{is_connected};
        $bus->disconnect;
    }

    log_info "Stopped";
}

sub tail {
    my ($self, $params) = @_;

    foreach ('count','level','after') {
        next unless defined $params->{$_};
        unless ($params->{$_} =~ m/^\d+(\.\d+)?$/) {
            die "Invalid parameter $_";
        }
    }

    foreach ('host','pool','service','message') {
        next unless defined $params->{$_};
        # Allow simple regexes
        unless ($params->{$_} =~ m/^[\w .*+?:,()\-\[\]\\]+$/) {
            die "Invalid parameter $_";
        }
    }

    my $count = $params->{count} || 10;
    my $after = $params->{after};
    my $level = $params->{level};

    # This will die when an invalid regex is provided, but that's fine
    my $host_re = defined $params->{host}    ? qr/$params->{host}/i    : undef;
    my $pool_re = defined $params->{pool}    ? qr/$params->{pool}/i    : undef;
    my $svc_re  = defined $params->{service} ? qr/$params->{service}/i : undef;
    my $msg_re  = defined $params->{message} ? qr/$params->{message}/i : undef;

    my ($entry, @filtered);

    for (my $i = @Log_buffer - 1; $i >= 0; $i--) {

        $entry = $Log_buffer[$i];

        next if (defined $level   && $entry->{level}    > $level   ) ||
                (defined $after   && $entry->{tstamp}  <= $after   ) ||
                (defined $host_re && $entry->{host}    !~ $host_re ) ||
                (defined $pool_re && $entry->{pool}    !~ $pool_re ) ||
                (defined $svc_re  && $entry->{service} !~ $svc_re  ) ||
                (defined $msg_re  && $entry->{message} !~ $msg_re  );

        unshift @filtered, $entry;

        last if (@filtered >= $count);
    }

    return \@filtered;
}

1;

__END__