File Coverage

lib/InfluxDB/LineProtocol.pm

Criterion	Covered	Total	%
statement	184	189	97.3
branch	47	60	78.3
condition	10	12	83.3
subroutine	19	19	100.0
pod	2	8	25.0
total	262	288	90.9

package InfluxDB::LineProtocol;
use strict;
use warnings;

# ABSTRACT: Write and read InfluxDB LineProtocol

our $VERSION = '1.013'; # VERSION

use Carp qw(croak);
use Time::HiRes qw(gettimeofday);

my %versions = (
    'v0.9.2' => '_0_9_2',
);

sub import {
    my $class = shift;
    my $caller = caller();

    my @to_export;
    my $version;
    my $precision = 'ns';
    foreach my $param (@_) {
        if ($param eq 'data2line' || $param eq 'line2data') {
            push(@to_export,$param);
        }
        if ($param =~ /^precision=(\w+)$/) {
            $precision = $1;
        }
        if ($param =~ /^v[\d\.]+$/ && $versions{$param}) {
            $version = $versions{$param};
        }
    }

    foreach my $function (@to_export) {
        my $target = $function;
        $function = '_'.$function.$version if $version;

        {
            no strict 'refs';
            *{"$caller\::$target"} = \&$function;
        }
    }

    # set up ts_$precision
    {
            no strict 'refs';
            my $selected = 'ts_'.$precision;
            *{"$caller\::get_ts"} = \&$selected;
    }

}

sub _format_key {
    my $k = shift;

    $k =~ s/([, ])/\\$1/g;

    return $k;
}

sub _format_value {
    my $k = shift;
    my $v = shift;

    if ( $v =~ /^(-?\d+)(?:i?)$/ ) {
        $v = $1 . 'i';
    }
    elsif ( $v =~ /^[Ff](?:ALSE|alse)?$/ ) {
        $v = 'FALSE';
    }
    elsif ( $v =~ /^[Tt](?:RUE|rue)?$/ ) {
        $v = 'TRUE';
    }
    elsif ( $v =~ /^-?\d+(?:\.\d+)?(?:e(?:-|\+)?\d+)?$/ ) {
        # pass it on, no mod
    }
    else {
        # string actually, but this should be quoted differently?
        $v =~ s/(["\\])/\\$1/g;
        $v = '"' . $v . '"';
    }

    return $v;
}

sub data2line {
    my ( $measurement, $values, $tags, $timestamp ) = @_;
    my $caller = caller();

    if ( @_ == 1 ) {
        # no $fields, so assume we already got a line
        return $measurement;
    }

    my $key = $measurement;
    $key =~ s/([, ])/\\$1/g;

    # $tags has to be a hashref, if it's not, we don't have tags, so it's the timestamp
    if ( defined $tags ) {
        if ( ref($tags) eq 'HASH' ) {
            my @tags;
            foreach my $k ( sort keys %$tags )
            {    # Influx wants the tags presorted
                # TODO check if sorting algorithm matches
                #      http://golang.org/pkg/bytes/#Compare
                my $v = $tags->{$k};
                next unless defined $v;
                $k =~ s/([, ])/\\$1/g;
                $v =~ s/([, ])/\\$1/g;
                push( @tags, $k . '=' . $v );
            }
            $key .= join( ',', '', @tags ) if @tags;
        }
        elsif ( !ref($tags) ) {
            $timestamp = $tags;
        }
    }

    $timestamp ||= $caller->get_ts();
    croak("$timestamp does not look like an epoch timestamp")
        unless $timestamp =~ /^\d+$/;

    # If values is not a hashref, convert it into one
    $values = { value => $values } if (not ref($values));

    my @fields;
    foreach my $k ( sort keys %$values ) {
        my $v = $values->{$k};

        my $esc_k = _format_key($k);
        my $esc_v = _format_value($k, $v);

        push( @fields, $esc_k . '=' . $esc_v );
    }
    my $fields = join( ',', @fields );

    return sprintf( "%s %s %s", $key, $fields, $timestamp );
}

sub ts_h {
    my $now = time();
    return int $now / 3600;
}

sub ts_m {
    my $now = time();
    return int $now / 60;
}

sub ts_s {
    return scalar time();
}

sub ts_ms {
    my ($s,$us) = gettimeofday();
    return sprintf("%s%03d", $s,substr($us,0,3));
}

sub ts_us {
    return sprintf("%s%06d", gettimeofday());
}

sub ts_ns {
    return sprintf("%s%06d000", gettimeofday());
}

sub line2data {
    my $line = shift;
    chomp($line);

    $line =~ s/\\ /ESCAPEDSPACE/g;
    $line =~ s/\\,/ESCAPEDCOMMA/g;
    $line =~ s/\\"/ESCAPEDDBLQUOTE/g;
    $line =~ s/\\\\/ESCAPEDBACKSLASH/g;

    $line=~/^(.*?) (.*) (.*)$/;
    my ($key, $fields, $timestamp) = ( $1, $2, $3);

    my ( $measurement, @taglist ) = split( /,/, $key );
    $measurement =~ s/ESCAPEDSPACE/ /g;
    $measurement =~ s/ESCAPEDCOMMA/,/g;

    my $tags;
    foreach my $tagset (@taglist) {
        $tagset =~ s/ESCAPEDSPACE/ /g;
        $tagset =~ s/ESCAPEDCOMMA/,/g;
        my ( $k, $v ) = split( /=/, $tagset );
        $tags->{$k} = $v;
    }

    my $values;
    my @strings;
    if ($fields =~ /"/) {
        my $cnt=0;
        $fields=~s/"(.*?)"/push(@strings, $1); 'ESCAPEDSTRING_'.$cnt++;/ge;
    }
    foreach my $valset ( split( /,/, $fields ) ) {
        $valset =~ s/ESCAPEDSPACE/ /g;
        $valset =~ s/ESCAPEDCOMMA/,/g;
        my ( $k, $v ) = split( /=/, $valset );
        $v =~ s/ESCAPEDSTRING_(\d+)/$strings[$1]/ge;
        $v =~ s/ESCAPEDDBLQUOTE/"/g;
        $v =~ s/ESCAPEDBACKSLASH/\\/g;
        $v =~ s/^(-?\d+)i$/$1/;
        $k =~ s/ESCAPEDBACKSLASH/\\\\/g;
        $values->{$k} = $v;
    }

    return ( $measurement, $values, $tags, $timestamp );
}

sub _data2line_0_9_2 {
    my ( $measurement, $values, $tags, $timestamp ) = @_;

    if ( @_ == 1 ) {
        # no $fields, so assume we already got a line
        return $measurement;
    }

    my $key = $measurement;
    $key =~ s/([, ])/\\$1/g;

    # $tags has to be a hashref, if it's not, we don't have tags, so it's the timestamp
    if ( defined $tags ) {
        if ( ref($tags) eq 'HASH' ) {
            my @tags;
            foreach my $k ( sort keys %$tags )
            {    # Influx wants the tags presorted
                # TODO check if sorting algorithm matches
                #      http://golang.org/pkg/bytes/#Compare
                my $v = $tags->{$k};
                next unless defined $v;
                $k =~ s/([, ])/\\$1/g;
                $v =~ s/([, ])/\\$1/g;
                push( @tags, $k . '=' . $v );
            }
            $key .= join( ',', '', @tags ) if @tags;
        }
        elsif ( !ref($tags) ) {
            $timestamp = $tags;
        }
    }

    if ($timestamp) {
        croak("$timestamp does not look like an epoch timestamp")
            unless $timestamp =~ /^\d+$/;
        if ( length($timestamp) < 19 ) {
            my $missing = 19 - length($timestamp);
            my $zeros   = 0 x $missing;
            $timestamp .= $zeros;
        }
    }
    else {
        $timestamp = join( '', gettimeofday(), '000' );
        $timestamp .= '0' if length($timestamp) < 19;
    }

    # If values is not a hashref, convert it into one
    $values = { value => $values } if (not ref($values));

    my @fields;
    foreach my $k ( sort keys %$values ) {
        my $v = $values->{$k};
        my $esc_k = _format_key($k);

        if (
            # positive & negativ ints, exponentials, use Regexp::Common?
            $v !~ /^-?\d+(?:\.\d+)?(?:e-?\d+)?$/
            &&
            # perl 5.12 Regexp::Assemble->new->add(qw(t T true TRUE f F false FALSE))->re;
            $v !~ /^(?:F(?:ALSE)?|f(?:alse)?|T(?:RUE)?|t(?:rue)?)$/
        )
        {
            $v =~ s/"/\\"/g;
            $v = '"' . $v . '"';
        }
        push( @fields, $esc_k . '=' . $v );
    }
    my $fields = join( ',', @fields );

    return sprintf( "%s %s %s", $key, $fields, $timestamp );
}

sub _line2data_0_9_2 {
    my $line = shift;
    chomp($line);

    $line =~ s/\\ /ESCAPEDSPACE/g;
    $line =~ s/\\,/ESCAPEDCOMMA/g;
    $line =~ s/\\"/ESCAPEDDBLQUOTE/g;

    $line=~/^(.*?) (.*) (.*)$/;
    my ($key, $fields, $timestamp) = ( $1, $2, $3);

    my ( $measurement, @taglist ) = split( /,/, $key );
    $measurement =~ s/ESCAPEDSPACE/ /g;
    $measurement =~ s/ESCAPEDCOMMA/,/g;

    my $tags;
    foreach my $tagset (@taglist) {
        $tagset =~ s/ESCAPEDSPACE/ /g;
        $tagset =~ s/ESCAPEDCOMMA/,/g;
        my ( $k, $v ) = split( /=/, $tagset );
        $tags->{$k} = $v;
    }

    my $values;
    my @strings;
    if ($fields =~ /"/) {
        my $cnt=0;
        $fields=~s/"(.*?)"/push(@strings, $1); 'ESCAPEDSTRING_'.$cnt++;/ge;
    }
    foreach my $valset ( split( /,/, $fields ) ) {
        $valset =~ s/ESCAPEDSPACE/ /g;
        $valset =~ s/ESCAPEDCOMMA/,/g;
        my ( $k, $v ) = split( /=/, $valset );
        $v =~ s/ESCAPEDSTRING_(\d+)/$strings[$1]/ge;
        $v =~ s/ESCAPEDDBLQUOTE/"/g;
        $values->{$k} = $v;
    }

    return ( $measurement, $values, $tags, $timestamp );
}

1;

__END__