ljr/wcmtools/spud/bin/rrd-storage

#!/usr/bin/perl

use strict;
use RRDs;
use IO::Socket::INET;
use Getopt::Long;
use Danga::Daemon;

# parse out command line options
my ($server, $port, $rrd_path);
my $res = GetOptions(
    "server=s"   => \$server,
    "port=i"     => \$port,
    "rrd-path=s" => \$rrd_path,
);
die "Unable to parse options.\n" unless $res;

# make sure we have a minimum set of options
die "Please specify a path to access the RRD files in --rrd-path.\n"
    unless $rrd_path;
die "RRD path $rrd_path does not exist or is not a directory.\n"
    unless -e $rrd_path && -d $rrd_path;
die "Requires --server (and a --port if it's not 9107).\n"
    unless $server;

# setup more internal variables
$port ||= 9107;
my $spud_server = "$server:$port";

# connect to local spud server
my $sock = IO::Socket::INET->new(PeerAddr => $spud_server)
    or die "Can't connect to server: $!\n";
print $sock "sub *\r\n";

# data storage
my $shutdown = 0;
my $lastpurge = time();
my (%data, %exists);

# now we want to daemonize
Danga::Daemon::daemonize(
    \&worker,
    {
        interval => 1,
    }
);

# process updates as they come in
sub worker {
    # try to reconnect?
    debug("Worker start...");
    unless ($sock) {
        $sock = IO::Socket::INET->new(PeerAddr => $spud_server)
            or return;
        print $sock "sub *\r\n";
    }

    # loop and process what they're giving us
    debug("Beginning main loop...");
    while (<$sock>) { 
        my $line = $_;
        $line =~ s/[\r\n]+$//;
        next unless $line =~ /^set\s+(\d+)\s+(\S+)\s+(\d+)$/i;
        my ($what, $time, $dpoint) = ($2, $1, $3);

        # fix up the colons in $what to be underscores :/
        $what =~ s/:/_/g;
        push @{$data{$what} ||= []}, [ $time, $dpoint ];

        # and if necessary, do a purge
        if ($Danga::Daemon::stop || ($lastpurge + 60 < time())) {
            do_purge();
            $lastpurge = time();
        }

        # done if we were told to shutdown
        return if $Danga::Daemon::stop;
    }

    # got undef from $sock... so undef $sock itself
    debug("Oops, lost connection to SPUD server...");
    $sock = undef;
}

sub do_purge {
    debug("** Beginning purge...");

    # dump to file
    my ($keys, $count);
    foreach my $what (keys %data) {
        # update this particular file
        $keys++;
        my $fn = "$rrd_path/$what";
        unless ($exists{$fn}) {
            unless (-e $fn) {
                RRDs::create($fn,
                    "--start", "-120",            # up to two minutes ago... should be enough
                    "--step", "5",                # data is expected to be 5 seconds apart
                    "DS:val:GAUGE:10:U:U",        # all data in 'val', 10 seconds before UNKNOWN
                    'RRA:AVERAGE:0.5:1:1440',     # past 2 hours of data
                    'RRA:AVERAGE:0.5:60:288',     # 5 minute averages for the day
                    'RRA:AVERAGE:0.5:360:336',    # 30 minute averages for a week
                    'RRA:AVERAGE:0.5:720:720',    # 1 hour averages for 30 days
                    'RRA:AVERAGE:0.5:17280:365',  # 1 day averages for a year
                    'RRA:AVERAGE:0.5:120960:520', # 1 week averages for 10 years
                );
                if (my $err = RRDs::error) {
                    die "Error creating RRD file: $err\n";
                }
            }
            $exists{$fn} = 1;
        }

        # now pipe out the updates
        my @updates;
        foreach my $row (@{$data{$what}}) {
            $count++;
            push @updates, "$row->[0]:$row->[1]";
        }
        RRDs::update($fn, @updates);
        #debug("\t$what: ", scalar(@updates), " written");
        if (my $err = RRDs::error) {
            warn "WARNING: Error updating $fn: $err\n";
        }
    }
    debug("\tupdates processed: $count");
    debug("\tfiles touched: $keys");
    debug("\tratio: %.2f updates per file\n", ($count / $keys));

    # and now that we're done with that old data ...
    %data = ();
}