octo_dispatcher

#!/usr/bin/perl

=head1 NAME

octo_dispatcher - Octopussy Dispatcher program

=head1 DESCRIPTION

octo_dispatcher is the program used by the Octopussy Project
to receive syslog lines and dispatche into device directories

(syslog --> <device>/Incoming/YYYY/MM/DD/msg_HHhMM_SS.log)

=cut

use strict;
use warnings;

use File::Copy;

use AAT::Syslog;
use AAT::Utils qw( NOT_NULL );
use Octopussy;
use Octopussy::App;
use Octopussy::Cache;
use Octopussy::Device;
use Octopussy::FS;
use Octopussy::RRDTool;
use Octopussy::Storage;

my $PROG_NAME = 'octo_dispatcher';
my $QR_ISO8601_SYSLOG =
qr/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|(?:.\d{2}:\d{2})) (\S+)/i;

exit if (!Octopussy::App::Valid_User($PROG_NAME));

my $exit_request = 0;
my $cache        = Octopussy::Cache::Init('octo_dispatcher');

my %dir_device;
my %device_type;
my %logs;
my %dtype_stats = ();
my %nb_events   = ();

=head1 FUNCTIONS

=head2 Init()

Inits Dispatcher

=cut

sub Init
{
    %dir_device  = ();
    %device_type = ();
    AAT::Syslog::Message($PROG_NAME, 'LOAD_DEVICES_CONFIG');
    my @devices = Octopussy::Device::Configurations('name');
    foreach my $d (@devices)
    {
        my $type = $d->{type};
        $type =~ s/ /_/g;
        my $status = (defined $d->{status} ? $d->{status} : 'Paused');
        $device_type{$d->{name}} = ($status ne 'Stopped' ? $type : '');
        $dir_device{$d->{name}} =
            Octopussy::Storage::Directory_Incoming($d->{name});
    }
    Octopussy::RRDTool::Syslog_By_DeviceType_Init();

    return (scalar @devices);
}

=head2 Stop()

Stops octo_dispatcher

=cut

sub Stop
{
    $exit_request = 1;

    return ($exit_request);
}

=head2 Handle_Dir($device, $year, $month, $day, $hour, $min)

Handles directory

=cut

sub Handle_Dir
{
    my ($device, $year, $month, $day, $hour, $min) = @_;

    if (!defined $dir_device{$device})
    {
        if (!Octopussy::Device::Valid_Name($device))
        {
            AAT::Syslog::Message($PROG_NAME,
                'LOGS_DROPPED_BECAUSE_INVALID_DEVICE_NAME', $device);
            return (undef);
        }

        my $param_auto_create =
            Octopussy::Parameter('automatic_device_creation');
        my $param_device_regexp =
            Octopussy::Parameter('device_filtering_regexp');
        if (   (defined $param_auto_create)
            && ($param_auto_create =~ /^Disabled$/i))
        {
            AAT::Syslog::Message($PROG_NAME,
                'LOGS_DROPPED_BECAUSE_AUTO_DEVICE_CREATION_DISABLED', $device);
            return (undef);
        }
        elsif ((defined $param_device_regexp)
            && ($device !~ /$param_device_regexp/))
        {
            AAT::Syslog::Message($PROG_NAME,
                'LOGS_DROPPED_BECAUSE_DEVICE_DIDNT_MATCH_REGEXP', $device);
            return (undef);
        }
        else
        {
            if (!-f Octopussy::Device::Filename($device))
            {
                Octopussy::Device::New(
                    {
                        name    => $device,
                        address => $device,
                        description =>
                            "New Device ($year/$month/$day $hour:$min) !"
                    }
                );
                $device_type{$device} = Octopussy::Parameter('devicetype');
                $dir_device{$device} =
                    Octopussy::Storage::Directory_Incoming($device);
            }
        }
    }

    return (undef) if (!defined $dir_device{$device});

    my $dir_incoming =
        "$dir_device{$device}/$device/Incoming/$year/$month/$day";
    Octopussy::FS::Create_Directory($dir_incoming);

    return ($dir_incoming);
}

=head2 Write_Logs_10secs($y, $m, $d, $hour, $min, $mod)

Writes Logs by 10 seconds block

=cut

sub Write_Logs_10secs
{
    my ($y, $m, $d, $hour, $min, $mod) = @_;

    foreach my $device (keys %logs)
    {
        my $dir = Handle_Dir($device, $y, $m, $d, $hour, $min);
        if (defined $dir)
        {
            my $file = "$dir/msg_${hour}h${min}_$mod.log";
            my $i    = 0;
            if (defined open my $FILE, '>>', $file)
            {
                foreach my $l (@{$logs{$device}})
                {
                    print {$FILE} "$l\n";
                    $i++;
                }
                close $FILE;
                $nb_events{$device} += $i;
                $dtype_stats{$device_type{$device}} += $i;
                Octopussy::FS::Chown($file) if (-f $file);
            }
            else
            {
                AAT::Syslog::Message($PROG_NAME, 'UNABLE_OPEN_FILE', $file);
            }
            delete $logs{$device};
        }
    }

    return (1);
}

=head2 Write_Stats($y, $m, $d, $hour, $min)

Writes Statistics

=cut

sub Write_Stats
{
    my ($y, $m, $d, $hour, $min) = @_;
    my @syslogs = ();
    my $total   = 0;

    foreach my $k (sort keys %nb_events)
    {
        push @syslogs, "Device: $k - Events: $nb_events{$k}";
        $total += $nb_events{$k};
    }
    AAT::Syslog::Messages($PROG_NAME, \@syslogs);
    $cache->set('dispatcher_stats_datetime',    "$y$m$d$hour$min");
    $cache->set('dispatcher_stats_devices',     \%nb_events);
    $cache->set('dispatcher_stats_devicetypes', \%dtype_stats);

    my $nb = $cache->get("dispatcher_stats_hourly_$y$m$d$hour");
    $nb = (NOT_NULL($nb) ? $nb + $total : $total);
    $cache->set("dispatcher_stats_hourly_$y$m$d$hour", $nb);

    %dtype_stats = ();
    %nb_events   = ();

    return ($total);
}

#
# MAIN
#

my ($device, $line) = (undef, undef);
my ($year,   $month,   $mday,   $hour,   $min,   $sec)   = AAT::Utils::Now();
my ($n_year, $n_month, $n_mday, $n_hour, $n_min, $n_sec) = AAT::Utils::Now();

$SIG{HUP}  = \&Init;
$SIG{USR1} = \&Stop;

Octopussy::PID_File($PROG_NAME);
AAT::Syslog::Message($PROG_NAME, 'PROGRAM_START');
Init();

while (!$exit_request)
{
    my $fifo = Octopussy::FS::File('fifo');
    Octopussy::Create_Fifo($fifo);
    open my $FIFO, '<', $fifo
        or Octopussy::Die($PROG_NAME,
        "[CRITICAL] Can't open named pipe $fifo: $!");
    while (my $line = <$FIFO>)
    {
        chomp $line;
        $line =~ s/^<\d+>//;
        if ($line =~ $QR_ISO8601_SYSLOG)
        {
            $device = $1;

            # Line dropped if devicename is 'not valid'
            if (!Octopussy::Device::Valid_Name($device))
            {
                AAT::Syslog::Message($PROG_NAME,
                    'LOGS_DROPPED_BECAUSE_INVALID_DEVICE_NAME', $device);
                next;
            }

            ($n_year, $n_month, $n_mday, $n_hour, $n_min, $n_sec) =
                AAT::Utils::Now();
            if (   (int($n_sec / 10) != int($sec / 10))
                || ($n_min != $min)
                || ($n_hour != $hour)
                || ($n_mday != $mday)
                || ($n_month != $month))
            {
                my $mod = (int($sec / 10) * 10);

                # every 10 secs, we write log files like msg_HHhMM_XX.log
                Write_Logs_10secs($year, $month, $mday, $hour, $min, $mod);

                if (   ($n_min != $min)
                    || ($n_hour != $hour)
                    || ($n_mday != $mday)
                    || ($n_month != $month)
                   )    # not the same minute than last minute
                {
                    Write_Stats($year, $month, $mday, $hour, $min);
                    ($year, $month, $mday, $hour, $min) =
                        ($n_year, $n_month, $n_mday, $n_hour, $n_min);
                }
                $sec = $n_sec;
            }
            push @{$logs{$device}}, $line
                if ((!defined $device_type{$device})
                || ($device_type{$device} ne ''));
        }
        last if ($exit_request);
    }
    close $FIFO;
    Write_Logs_10secs($year, $month, $mday, $hour, $min, int($sec / 10) * 10);
}

=head1 AUTHOR

Sebastien Thebert <octopussy@onetool.pm>

=head1 SEE ALSO

octo_extractor, octo_parser, octo_uparser, octo_reporter, octo_rrd, 
octo_scheduler, octo_sender

=cut