json-logger: A practical script to capture multiline output from e.g. cron jobs into a single logstash JSON event

_json-logger.md

Can be used in crontab like this:

SHELL=/bin/bash
BASH_ENV=~/.bash_profile
# m h  dom mon dow   command
0    * * * * /home/portfolio/portfolio/bin/session.pl                2> >(json-logger -ep pf_session)                | json-logger -p pf_session
0    2 * * * /home/portfolio/portfolio/bin/purge_deleted_messages.pl 2> >(json-logger -ep pf_purge_deleted_messages) | json-logger -p pf_purge_deleted_messages
0,30 * * * * /home/portfolio/portfolio/bin/expire_page_views.pl      2> >(json-logger -ep pf_expire_page_views)      | json-logger -p pf_expire_page_views

I would suggest you put it in /usr/bin to avoid having to setup your PATH variable.

This is what you need of config in Logstash to capture the events generated by json-logger:

input {
    tcp {
        type   => 'json-logger'
        port   => 3517
        format => "json_event"
    }
}

You can of course also use the UDP input, but then your events are limited to 8K because of the UDP packet size limitation.

json-logger

#!/usr/bin/env perl

# Copyright 2012 Robin Smidsrød <[email protected]>
# Licensed under the same terms as Perl itself
# http://dev.perl.org/licenses/

use strict;
use warnings;

use Encode qw(decode FB_QUIET);
use Time::HiRes qw(time);
use IO::Socket::INET;

use Getopt::Long::Descriptive qw(describe_options);
use JSON qw(decode_json encode_json);
use DateTime;
use File::Slurp qw(read_file);

my ($opt, $usage) = describe_options(
    "json-logger %o",
    [ 'config|f=s',   "default: /etc/json-logger/json-logger.json",
                      { default => '/etc/json-logger/json-logger.json' } ],
    [ 'encoding|c=s', "input encoding (input-encoding) default: utf-8" ],
    [ 'mimetype|m=s', "input mimetype (input-mimetype) default: text/plain" ],

    [ 'is-stderr|e',  "event is error output           default: false" ],
    [ 'timezone|z=s', "event timezone (event-timezone) default: UTC" ],
    [ 'type|t=s',     "event type         (event-type) default: json-logger" ],
    [ 'tags|g=s@',    "event tags         (event-tags) default: none" ],
    [ 'program|p=s',  "event program                   default: json-logger",
                      { default => 'json-logger' } ],

    [ 'send-empty|y', "send empty events  (send-empty) default: false" ],

    [ 'server|s=s',   "server to emit to      (server) default: 127.0.0.1" ],
    [ 'port|o=i',     "port to emit to          (port) default: 3517" ],
    [ 'protocol|l=s', "sending protocol     (protocol) default: tcp" ],

    [ 'help|h',       "bring up this help message" ],
);

# Exit immediately if help requested
if ( $opt->help ) {
    print STDERR $usage->text;
    print STDERR "\nThe words in parenthesis are the JSON config file keys.\n";
    exit 1; # indicate error to caller
}

# Fetch the start timestamp
my $start_epoch = time();

# Read in config file
my $conf = -r $opt->config
         ? decode_json( read_file( $opt->config ) )
         : {};

# Generate a timestamp in the requested timezone
my $tz = $opt->timezone || $conf->{'event-timezone'} || 'UTC';
my $start_ts = DateTime->from_epoch( epoch => $start_epoch, time_zone => $tz );

# Figure out what protocol to use
my $protocol = $opt->protocol || $conf->{'protocol'} || 'tcp';

# Figure out what type of event to emit
my $event_type = $opt->type || $conf->{'event-type'} || 'json-logger';

# Figure out what type of tags to attach to event
my $event_tags = $opt->tags || $conf->{'event-tags'} || [];

# Figure out what encoding to use for input data
my $input_encoding = $opt->encoding || $conf->{'input-encoding'} || 'utf-8';

# Figure out what content type our input is
my $mimetype = $opt->mimetype || $conf->{'input-mimetype'} || 'text/plain';

# Figure out hostname to use
my $source_host = qx(hostname);
chomp $source_host;

# Get network settings
my $server = $opt->server || $conf->{'server'} || '127.0.0.1';
my $port   = $opt->port || $conf->{'port'} || 3517;

# Set up some kind of @source variable
my $source = "$protocol://$server:$port/client/$source_host/" . $opt->program;
my $source_path = "/client/$source_host/" . $opt->program;

# Slurp in all lines of output and try to decode it according to input-encoding
# Silently convert broken input
my $input = "";
$input .= $_ while <>;
chomp $input;
$input = decode($input_encoding, $input, FB_QUIET);

# Deal with empty output from program, exit if unwanted
my $is_empty = 0;
if ( length $input == 0 ) {
    if ( $opt->send_empty ) {
        $input = "Program <" . $opt->program . "> produced no output";
        $is_empty = 1;
    }
    else {
        exit;
    }
}

# Create socket for transmission of event
my $sock = IO::Socket::INET->new(
    PeerAddr  => $server,
    PeerPort  => $port,
    Proto     => $protocol,
    Timeout   => 10,
);

# Calculate end timestamp and duration
my $end_epoch = time();
my $end_ts = DateTime->from_epoch( epoch => $end_epoch, time_zone => $tz );
my $duration = $end_epoch - $start_epoch;

# Encode JSON
my $json = encode_json({
    '@timestamp'   => $end_ts->strftime('%FT%T.%9N%z'),
    '@message'     => $input,
    '@type'        => $event_type,
    '@tags'        => $event_tags,
    '@source'      => $source, 
    '@source_path' => $source_path,
    '@source_host' => $source_host,
    '@fields'      => {
        'started_at' => $start_ts->strftime('%FT%T.%6N%z'),
        'duration'   => 0 + $duration,
        'program'    => $opt->program,
        'fd'         => ( $opt->is_stderr ? 'stderr' : 'stdout' ),
        'mimetype'   => $mimetype,
        ( $is_empty ? ( 'empty' => JSON::true ) : () ),
    },
});

# Send to socket if available
if ( $sock ) {
    $sock->send("$json\n");
    exit;
}

# Crash and burn on STDERR if socket failed
print STDERR "Unable to send JSON event to $protocol://$server:$port/: $!\n";
print STDERR "$json\n";
exit 1;