octopussy 1.0.6-0ubuntu2 / usr / sbin / octo_pusher

#!/usr/bin/perl -w

=head1 NAME

octo_pusher - Octopussy Pusher program

=head1 SYNOPSIS

octo_pusher

=head1 DESCRIPTION

octo_pusher is the program used by the Octopussy Project to parse 'no-syslog' Logs for 'no-syslog' Devices

=cut

use strict;
use warnings;
use Readonly;

use File::Copy;
use File::Path;

use AAT::Syslog;
use AAT::Utils qw( ARRAY NOT_NULL );
use DateTime::Format::Strptime;
use Octopussy;
use Octopussy::Device;
use Octopussy::FS;
use Octopussy::Storage;

Readonly my $PROG_NAME => 'octo_pusher';
Readonly my $TIMEZONE => Octopussy::Parameter('octo_pusher_timezone');

my %conf;
my %dir_device;

my %datetime_subs = ( 
	'localtime' 				=> new DateTime::Format::Strptime(pattern => '%s'),
	'dw3_m3_dm2_h2_m2_s2_y4' 	=> new DateTime::Format::Strptime(pattern => '%a %b %d %H:%M:%S %Y'), 
	'd2_m3_y4_h2_m2_s2' 		=> new DateTime::Format::Strptime(pattern => '%d-%b-%Y %H:%M:%S'), 
	'd2m2y2hms' 				=> new DateTime::Format::Strptime(pattern => '%d%m%y%H%M%S'), 
	);

=head1 FUNCTIONS

=head2 Init()

Inits Pusher

=cut

sub Init
{
    %dir_device = ();
    AAT::Syslog::Message($PROG_NAME, 'LOAD_DEVICES_CONFIG');
    my $storage = Octopussy::Storage::Default();
    my @devices = Octopussy::Device::Configurations('name');
    foreach my $d (@devices)
    {
        if (   (defined $d->{logtype})
            && ($d->{logtype} eq 'asynchronous')
            && (defined $d->{async}))
        {
            foreach my $as (@{$d->{async}})
            {
                push @{$conf{$d->{name}}},
                    {regexp => qr/$as->{regexp}/, output => $as->{output}};
            }
            $dir_device{$d->{name}} =
                Octopussy::Storage::Directory_Incoming($d->{name});
        }
    }

    return (scalar @devices);
}

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

Handles Directory ($device/$year/$month/$day)

=cut

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

    if (!defined $dir_device{$device})
    {
        if (!-f Octopussy::Device::Filename($device))
        {
            Octopussy::Device::New(
                {
                    name        => $device,
                    address     => $device,
                    description => 'New Device !'
                }
            );
            my $storage = Octopussy::Storage::Default();
            $dir_device{$device} =
                Octopussy::Storage::Directory_Incoming($device);
        }
    }
    $day = sprintf('%02d', $day);
    my $dir_incoming =
        "$dir_device{$device}/$device/Incoming/$year/$month/$day";
    Octopussy::FS::Create_Directory($dir_incoming);

    return ($dir_incoming);
}

=head2 Convert_To_Syslog($line, $device)

Converts line $line to a syslog line

=cut

sub Convert_To_Syslog
{
    my ($line, $device) = @_;
    chomp $line;
    my $syslog = $line;

    foreach my $as (ARRAY($conf{$device}))
    {
        if (NOT_NULL($as->{output}))
        {
            $syslog = $as->{output};
            if (my @var = $line =~ /$as->{regexp}/)
            {
            	foreach my $k (keys %datetime_subs)
            	{
            		if ($syslog =~ /__${k}\((\d+)\)__/)
                	{
                        my $dt = ${datetime_subs}{$k}->parse_datetime($var[$1 - 1]);
                        my $substitution = sprintf("%sT%s.000000%s", $dt->ymd('-'), $dt->hms(':'), $TIMEZONE);
                        $syslog =~ s/__${k}\(\d+\)__/$substitution/g;
                	}	
            	}
                $syslog =~ s/__(\d+)__/$var[$1-1]/g;
                $syslog =~ s/__device__/$device/g;
                last;
            }
        }
    }

    return ($syslog);
}

=head2 Write_To_Incoming($dev, $year, $month, $mday, $hour, $min, @lines)

Writes to Incoming Directory

=cut

sub Write_To_Incoming
{
    my ($dev, $year, $month, $mday, $hour, $min, @lines) = @_;

    if (scalar @lines > 0)
    {
        my $dir = Handle_Dir($dev, $mday, $month, $year);
        my $file = "$dir/msg_${hour}h${min}_00.log";
        if (defined open my $INCOMING, '>>', $file)
        {
            foreach my $l (@lines) { print {$INCOMING} "$l\n"; }
            close $INCOMING;
            Octopussy::FS::Chown($file);
        }
        else
        {
            print "Unable to open file '$file'\n";
            AAT::Syslog::Message('octo_pusher', 'UNABLE_OPEN_FILE', $file);
        }
    }

    return (scalar @lines);
}

#
# MAIN
#
exit if (!Octopussy::Valid_User($PROG_NAME));

my $device = undef;

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

$SIG{HUP} = \&Init;
Octopussy::PID_File($PROG_NAME);
AAT::Syslog::Message($PROG_NAME, 'PROGRAM_START');

while (1)
{
    Init();
    my $DIR_PUSHER = Octopussy::FS::Directory('data_logs_pusher');
    if (-d $DIR_PUSHER)
    {
        opendir DIR, $DIR_PUSHER;
        my @dirs = grep { !/^\./ } readdir DIR;
        closedir DIR;
        foreach my $d (@dirs)
        {
            if ((-d "$DIR_PUSHER/$d/") && (defined $conf{$d}))
            {
                opendir DIR2, "$DIR_PUSHER/$d/";
                my @files = grep { !/^\./ } readdir DIR2;
                foreach my $f (@files)
                {
                    if (defined open my $FILE, '<', "$DIR_PUSHER/$d/$f")
                    {
                        my @lines = ();
                        while (<$FILE>)
                        {
                            my $syslog = Convert_To_Syslog($_, $d);
                            if ($syslog =~
                            	/^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):\d{2}(?:\.\d{1,6})?.\d{2}:\d{2} (\S+)/
                               )
                            {
                            	($n_year, $n_month, $n_mday, $n_hour, $n_min, $device) = ($1, $2, $3, $4, $5, $6);
                                if (   ($n_min ne $min)
                                    || ($n_hour  ne $hour)
                                    || ($n_mday  ne $mday)
                                    || ($n_month ne $month))
                                {
                                    Write_To_Incoming(
                                        $device, $year, $month, $mday,
                                        $hour,   $min,  @lines
                                    );
                                    ($year, $month, $mday, $hour, $min) = (
                                        $n_year, $n_month, $n_mday,
                                        $n_hour, $n_min
                                    );
                                    @lines = ();
                                }
                                push @lines, $syslog;
                            }
                        }
                        close $FILE;
                        unlink "$DIR_PUSHER/$d/$f"
                            if ((scalar @lines > 0)
                            || (-z "$DIR_PUSHER/$d/$f"));
                        Write_To_Incoming(
                            $device, $year, $month, $mday,
                            $hour,   $min,  @lines
                        ) if (defined $device);
                    }
                    else
                    {
                        print "Unable to open file '$f'\n";
                        AAT::Syslog::Message('octo_pusher', 'UNABLE_OPEN_FILE',
                            $f);
                    }
                }
                closedir DIR2;
            }
        }
    }
    sleep 300;
}

=head1 AUTHOR

Sebastien Thebert <octo.devel@gmail.com>

=head1 SEE ALSO

octo_extractor, octo_parser, octo_uparser, octo_reporter, octo_rrd, 
octo_scheduler

=cut