File Coverage

blib/lib/InfluxDB/Client.pm

Criterion	Covered	Total	%
statement	26	121	21.4
branch	0	74	0.0
condition	0	39	0.0
subroutine	9	15	60.0
pod	5	5	100.0
total	40	254	15.7

package InfluxDB::Client;
# ABSTRACT: query and manage InfluxDB


use strict;
use warnings;
use v5.10;

our $VERSION = '0.1';

use Carp;
use LWP::UserAgent;
use HTTP::Request::Common;
use URI;
use HTTP::Status ();

use InfluxDB::Client::Response;


sub new {
    my ( $class, %args ) = @_;

    # validate args
    croak "no url given" unless $args{url};
    my $url = $args{url} || 'http://localhost:8086';

    croak "url must be a string or an ARRAY ref"
        if ( ref $url && ref $url ne 'ARRAY' );

    $url = ref $url ? $url : [ $url ];

    foreach ( @$url ) {
        # we check each given url, if it is valid
        my $uri = URI->new($_);
        croak "invalid url given: $_"
            unless ( $uri && $uri->has_recognized_scheme && $uri->scheme =~ m/^https*$/ );
    }

    my $opts = $args{ua_opts} // {};
    croak "ua_opts must be a hash reference"
        unless ( ref $opts && ref $opts eq 'HASH' );

    $opts->{agent}      //= sprintf('perl-InfluxDB-Client/%f',$VERSION);

    # 10 second timeout should be more then enough
    $opts->{timeout}    //= 10;

    # check if a user or password arge given solo
    croak "username without a password supplied"
        if ( $args{username} && !$args{password} );
    croak "password without a username supplied"
        if ( $args{password} && !$args{username} );

    my $self = bless {
        # we don't save URI objects here, es we have construct them later on the fly to implement RR
        urls        => $url,
        username    => $args{username},
        password    => $args{password},
        # index for the round robin mechanism
        _cur_uri    => 0,
        _num_uri    => scalar(@$url),

        # our useragent
        _ua         => LWP::UserAgent->new( %$opts )
    } => $class;

    return $self;
}

sub ping {
    my ( $self ) = @_;
    my $response = $self->_send_request('GET','/ping');

    # our status code must be 204
    if ($response->http_code != 204) {
        # everything else is an error
        $response->status(InfluxDB::Client::Response::Status::SRVFAIL);
    }
    return $response;
}

sub query {
    # shortcut, if we only get an query as parameter
    if (@_ == 2) {
        return $_[0]->_send_request('GET','/query', { q => $_[1] });
    }

    # else we expect a hash
    my ( $self, %args ) = @_;

    # check the query argument
    croak 'no query given'
        unless $args{query};
    croak 'query must not be a reference'
        unless (!ref $args{query});

    # validate the method argument
    my $method = 'GET';
    if ($args{method}) {
        croak 'method must be either "GET" or "POST"'
            unless ($args{method} =~ m/^(GET|POST)$/);
        $method = $args{method};
    }

    # validate the epoch argument
    my $epoch = 'ns';
    if ($args{epoch}) {
        croak 'epoch must be one of "ns","u","ms","s","m","h"'
            unless ($args{epoch} =~ m/^(s|u|ms|s|m|h)$/);
        $epoch = $args{epoch};
    }

    # build our parameter lis
    my $data;
    my $qs = { epoch   => $epoch  };
    $qs->{db} = $args{db} if ($args{db});
    if ( $method eq 'POST' ) {
        # if we post our query, we have to put it in the body
        $data = { q => $args{query} }
    } else {
        $qs->{q} = $args{query};
    }

    # send our request and return the result
    return $self->_send_request($method,'/query',$qs,$data);
}


sub write {
    my ( $self, %args ) = @_;

    # check the database argument
    croak 'no database given'
        unless $args{database};
    croak 'database must not be a reference'
        if (ref $args{database});

    # check the data argument
    croak 'no data given'
        unless $args{data};
    croak 'database must be an array reference'
        unless (ref $args{data} && ref $args{data} eq 'ARRAY');

    # precision argument
    unless ($args{precision}) {
        # check the data array first
        if ( ref $args{data}[0] && ref $args{data}[0] eq 'InfluxDB::Client::DataSet' ) {
            $args{precision} = $args{data}[0]->precision;
        } else {
            $args{precision} = 'ns';
        }
    } else {
        croak 'precision must be one of "ns","u","ms","s","m","h"'
            unless ($args{precision} =~ m/^(s|u|ms|s|m|h)$/);
    }

    # the consistency argument
    if ($args{consistency}) {
        croak 'consistency must not be a reference'
            if (ref $args{consistency});

        croak 'consistency must be one of "any","one","quorum","all'
            unless ($args{consistency} =~ m/^(any|one|quorum|all)$/);
    } else {
        $args{consistency} = "one";
    }

    # all required paramters are checked, we can build our data to write
    my $qs = {
        db          => $args{database},
        precision   => $args{precision},
        consistency => $args{consistency},
    };
    # we only set the retention policy if we have one
    $qs->{rp} = $args{rp} if $args{rp};

    # we need to prepare our data
    my @data;
    foreach my $ds ( @{$args{data}} ) {
        if (ref $ds) {
            croak sprintf('invalid type "%s" for a dataset object',ref $ds)
                unless (ref $ds eq 'InfluxDB::Client::DataSet');
            push @data,$ds->to_string;
        } else {
            push @data,$ds;
        }
    }

    # write the data and return the result
    return $self->send_request('POST','/write',$qs,join("\n",@data));
}
sub show_databases {
    my ( $self ) = @_;
    my $response = $self->query('SHOW DATABASES');

    if ( $response->status == InfluxDB::Client::Response::Status::OK ) {
        $response->content([ map { $_->[0] } @{$response->content->[0]{series}[0]{values}} ]);
    }
    return $response;
}

# helper to send a request to the target database
sub _send_request {
    my ( $self, $method, $path, $qs, $data ) = @_;

    # we build a new URI object to generate the string first
    my $uri = URI->new($path);
    $uri->query_form($qs) if $qs;

    my $resp;
    # iterate over our hosts and try to get an response
    for(my $i = 0; $i < $self->{_num_uri}; ++$i) {
        # make sure we are within the valid range of uri's
        my $idx = $self->{_cur_uri} + $i;
        $idx -= $idx >= $self->{_num_uri} if ( $idx >= $self->{_num_uri});

        my $url = $uri->abs($self->{urls}[$idx])->as_string();
        my $req;
        if ( $method eq 'POST' ) {
            # with POST, we need to add the body content
            $req = POST(
                $url,
                Content => $data
            );
        } else {
            # all other request types could you use there constructor from HTTP::Request::Common
            $req = &{\&{$method}}($url);
        }

        if ( $self->{username} ) {
            # if authentication is enabled, we send the login data with the request
            # LWP::UserAgent would need a REALM, but most user wouldn't know what to enter there,
            # and it would make the code on user-side more complex, so we HTTP::Request::Common for that
            $req->authorization_basic($self->{username}, $self->{password});
        }

        # send our request
        $resp = $self->{_ua}->request($req);

        # check our response code, if we got a timeout or 5xx, we should try the next server
        if ( $resp->code == HTTP::Status::HTTP_REQUEST_TIMEOUT || ( $resp->code >= 500 && $resp->code < 600 ) ) {
            next;
        }

        # increment the current index, make sure we don't overflow
        ++$self->{_cur_uri};
        $self->{_cur_uri} = 0 if $self->{_cur_uri} == $self->{_num_uri};

        # simply break the loop here, we always return the last response we got
        last;
    }

    # we encapsulate the response and return it
    return InfluxDB::Client::Response->new( http_response =>  $resp );
}

1;

__END__