File Coverage

blib/lib/InfluxDB/Client/Simple.pm

Criterion	Covered	Total	%
statement	23	145	15.8
branch	0	66	0.0
condition	0	8	0.0
subroutine	8	16	50.0
pod	5	5	100.0
total	36	240	15.0

package InfluxDB::Client::Simple;

use 5.006;
use strict;
use warnings;

use Carp;
use IO::Socket::INET;
use JSON;
use LWP::UserAgent;
use URI;

=head1 NAME

InfluxDB::Client::Simple - The lightweight InfluxDB client

=head1 VERSION

Version 1.00

=cut

our $VERSION = '1.00';

=head1 SYNOPSIS

InfluxDB::Client::Simple provides an easy way to interact with an InfluxDB server.

    use InfluxDB::Client::Simple;

    ########################## TCP ##########################
    my $client = InfluxDB::Client::Simple->new( host => 'server.address.com', port => 8086, protocol => 'tcp' ) or die "Can't instantiate client";

    # Check server connectivity
    my $result = $client->ping();
    die "No pong" unless $result;

    # You can also get the server version
    print $result->{version};

    # Read
    $result = $client->query('SELECT "severity_code" FROM "syslog" WHERE ("severity" = \'err\' AND "hostname" =~ /^(srv01|srv02)$/) AND time >= 1558878013531ms and time <= 1609886964827ms', database => 'grafana');

    # Write
    $result = $client->write("testing,host=containment,repo=cadi-libs,file=testfile statement=42,pod=85", database => 'dbname');

    ########################## UDP ##########################
    $client = InfluxDB::Client::Simple->new( host => 'server.address.com', port => 8089, protocol => 'udp', database => 'grafana' ) or die "Can't instantiate client";

    # UDP allows only write()
    $result = $client->write("testing,host=containment,repo=cadi-libs,file=testfile statement=47,pod=89");
    # or with a timestamp
    $result = $client->write("testing,host=containment,repo=cadi-libs,file=testfile statement=47,pod=89 1610462906000");


=head1 WHY

In its current state this module offers few additional features over InfluxDB::HTTP (from which it's derived)

The only reasons why you would use this module are:

=over

=item *

Minimal dependencies (no Object::Result and its dependencies)

=item *
You want to use UDP protocol for writing

=back

=head1 SUBROUTINES/METHODS

=head2 new ( [%options] )

Constructor.
%otions is a hash with the following keys:

=over

=item *
database - Database name (default: 'grafana')

=item *
host - Server hostname (default: 'localhost')

=item *
port - Server port (default: 8086)

=item *
protocol - Transport protocol 'udp' or 'tcp' (default: 'tcp')
Note that when using the udp protocol, the default behaviour is to avoid dying on errors.
(You can change that with the 'strict_udp' option)

=item *
strict_udp - Boolean value to die on UDP error (false by default)

=item *
timeout - Timeout value in seconds (default: 180)

=back

=cut

sub new {
    my $class = shift;
    my %args = ( database => 'grafana',
                 host     => 'localhost',
                 port     => 8086,
                 protocol => 'tcp',
                 timeout  => 180,
                 @_,
    );
    my ( $host, $port, $protocol, $strict_udp, $timeout ) = map { defined($_)?lc($_):'' } @args{ 'host', 'port', 'protocol', 'strict_udp', 'timeout' };

    my $self = { host => $host,
                 port => $port,
                 protocol => $protocol,
                 options => { database => $args{database} }
    };

    if ( $protocol eq 'tcp' ) {
        my $ua = LWP::UserAgent->new();
        $ua->agent("InfluxDB-Client-Simple/$VERSION");
        $ua->timeout($timeout);
        $self->{lwp_user_agent} = $ua;
    } else {
        die "Unknown protocol: $protocol" unless $protocol eq "udp";

        my $socket = IO::Socket::INET->new( PeerAddr => "$host:$port",
                                            Proto    => $protocol,
                                            Blocking => 0
        );

        if ($strict_udp) {
          die("Can't open socket: $@") unless $socket;
        }

        $self->{udp} = $socket;
    }

    bless $self, $class;

    return $self;
}

=head2 ping()

Check the server connectivity.

Returns a hashref which evaluates to true if the connection is ok and to false otherwise.
The hashref has the following keys:

=over

=item *
raw - The raw response from the server

=item *
error - The error message returned by the server (empty on success)

=item *
version - The InfluxDB verstion returned by the server through the 'X-Influxdb-Version' header

=back

=cut

sub ping {
    my ($self)   = @_;
    my $uri      = $self->_get_influxdb_http_api_uri('ping');
    my $response = $self->{lwp_user_agent}->head( $uri->canonical() );

    if ( !$response->is_success() ) {
        my $error = $response->message();
        return { raw     => $response,
                 error   => $error,
                 version => undef,
        };
    }

    my $version = $response->header('X-Influxdb-Version');
    return { raw     => $response,
             error   => undef,
             version => $version,
    };
}

=head2 query( $query [, %options] )

Query the InfluxDB database using the $query passed as first parameter.
Optionally %options can be passed as a hash
Allowed keys for options are:

=over

=item *
database - The database to be queried on the InfluxDB server

=item *
chunksize - The size of the chunks used for the returned data

=item *
epoch - The precision format (h, m, s, ms, u, ns) for epoch timestamps (default is 'ns' for nanosecond)

=back

Returns a hashref whose keys are:

=over

=item *
raw - The raw response from the server

=item *
error - The error message returned by the server (empty on success)

=item *
data - The InfluxDB data returned by the server

=back

=cut

sub query {
    my $self  = shift;
    my $query = shift;
    my %args  = ( epoch => 'ns', @_ );
    my ( $database, $chunk_size, $epoch ) = @args{ 'database', 'chunk_size', 'epoch' };

    die "Missing argument 'query'" if !$query;
    die "Argument epoch '$epoch' is not one of (h,m,s,ms,u,ns)" if $epoch !~ /^(h|m|s|ms|u|ns)$/;

    if ( ref($query) eq 'ARRAY' ) {
        $query = join( ';', @$query );
    }

    my $uri = $self->_get_influxdb_http_api_uri('query');

    $uri->query_form( q => $query,
                      ( $database   ? ( db         => $database )   : () ),
                      ( $chunk_size ? ( chunk_size => $chunk_size ) : () ),
                      ( $epoch      ? ( epoch      => $epoch )      : () )
    );

    my $response = $self->{lwp_user_agent}->post( $uri->canonical() );

    chomp( my $content = $response->content() );

    my $error;
    if ( $response->is_success() ) {
        local $@;
        my $data = eval { decode_json($content) };
        $error = $@;

        if ($data) {
            $error = $data->{error};
        }

        if ( !$error ) {
            $data->{request_id} = $response->header('Request-Id');
            return { raw   => $response,
                     data  => $data,
                     error => undef,
            };
        }
    } else {
        $error = $content;
    }

    return { raw   => $response,
             data  => undef,
             error => $error,
    };
}

=head2 write ($measurement | \@measurements, [%options])

$measurement is the data to be send encoded according to the LineProtocol.

%options can have the following keys:

=over

=item *
database - The database to be queried on the InfluxDB server

=item *
retention_policy - The retention policy to be used (if different from the default one)

=item *
precision - The precision used in the data (if diffectent from the default 'ns')

=back

Returns a hashref whose keys are:

=over

=item *
raw - The raw response from the server (obviously empty when using UDP)

=item *
error - The error message returned by the server (empty on success)

=back

=cut

sub write {
    my $self        = shift;
    my $measurement = shift;
    my %args        = (%{$self->{options}},  @_);
    my ( $database, $precision, $retention_policy ) = @args{ 'database', 'precision', 'retention_policy' };

    die "Missing argument 'measurement'"                                        if !$measurement;
    die "Missing argument 'database'"                                           if !$database;
    die "Argument precision '$precision' is set and not one of (h,m,s,ms,u,ns)" if $precision && $precision !~ /^(h|m|s|ms|u|ns)$/;

    if ( ref($measurement) eq 'ARRAY' ) {
        $measurement = join( "\n", @$measurement );
    }

  if ($self->{protocol} eq 'tcp') {
    my $uri = $self->_get_influxdb_http_api_uri('write');

    $uri->query_form( db => $database,
                      ( $precision        ? ( precision => $precision )        : () ),
                      ( $retention_policy ? ( rp        => $retention_policy ) : () )
    );

    my $response = $self->{lwp_user_agent}->post( $uri->canonical(), Content => $measurement );

    chomp( my $content = $response->content() );

    if ( $response->code() != 204 ) {
        local $@;
        my $data = eval { decode_json($content) };
        my $error = $@;
        $error = $data->{error} if ( !$error && $data );

        return { raw   => $response,
                 error => $error,
        };
    }

    return { raw   => $response,
             error => undef,
    };

  } else {

    # Udp send
    my $bytes = $self->{udp}?$self->{udp}->send($measurement):0;

    # should be more picky here : compare $bytes with length of $measurement ?
    return { raw   => undef,
             error => $bytes?undef:"Undefinded error while sending data (udp)",
    };
  }
}


=head2 send_data ($measurement, \%tags, \%fields, [$timestamp, [%options]])

Write data to the influxDB after converting them into LineProtocol format.
(call write() underneath)

$measurement is the name to be used for measurement

\%tags is the tag set associated to this datapoint

\%fields are the field set associated to this datapoint

$timestamp is an optional timestamp value expected to be in the database precision format (default is nanosecond)

\%options are also optional

%options can have the following keys:

=over

=item *
database - The database to be queried on the InfluxDB server

=item *
retention_policy - The retention policy to be used (if different from the default one)

=item *
precision - The precision used in the data (if diffectent from the default 'ns')

=back

Returns a hashref whose keys are:

=over

=item *
raw - The raw response from the server (obviously empty when using UDP)

=item *
error - The error message returned by the server (empty on success)

=back

=cut

sub send_data {
    my $self        = shift;
    my $measurement = shift;
    my $tags        = shift;
    my $fields      = shift;
    my $timestamp   = shift;
    my %options     = @_;

    return $self->write( _line_protocol( $measurement, $tags, $fields, $timestamp ), %options );

}

sub _get_influxdb_http_api_uri {
    my ( $self, $endpoint ) = @_;

    die "Missing argument 'endpoint'" if !$endpoint;

    my $uri = URI->new();

    $uri->scheme('http');
    $uri->host( $self->{host} );
    $uri->port( $self->{port} );
    $uri->path($endpoint);

    return $uri;
}

# Blatantly stolen from InfluxDB::LineProtocol
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 _line_protocol {
    my $measurement = shift;
    my $tags        = shift;
    my $fields      = shift;
    my $timestamp   = shift;

    # sort and encode (LineProtocol) tags
    my @tags;
    foreach my $k ( sort keys %$tags ) {
        my $v = $tags->{$k};
        next unless defined($v);
        $k =~ s/([,\s])/\\$1/g;
        $v =~ s/([,\s])/\\$1/g;

        push( @tags, $k . '=' . $v );
    }
    my $tag_string = join( ',', @tags );

    # sort and encode (LineProtocol) fields
    my @fields;
    foreach my $k ( sort keys %$fields ) {
        my $v = $fields->{$k} || '';
        my $esc_k = $k;
        $esc_k =~ s/([,\s])/\\$1/g;
        my $esc_v = _format_value( $k, $v );

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

    if ($timestamp) {
        return sprintf( "%s,%s %s %s", $measurement, $tag_string, $field_string, $timestamp );
    } else {
        return sprintf( "%s,%s %s", $measurement, $tag_string, $field_string );
    }
}

1;

=head1 AUTHOR

Arnaud (Arhuman) ASSAD, C<<  >>

=head1 BUGS

Please report any bugs or feature requests to C, or through
the web interface at L.  I will be notified, and then you'll
automatically be notified of progress on your bug as I make changes.

=head1 SEE ALSO

This module is derived from InfluxDB::HTTP.
This module borowed code from InfluxDB::LineProtocol

=head1 SUPPORT

You can find documentation for this module with the perldoc command.

    perldoc InfluxDB::Client::Simple


You can also look for information at:

=over 4

=item * RT: CPAN's request tracker (report bugs here)

L

=item * AnnoCPAN: Annotated CPAN documentation

L

=item * CPAN Ratings

L

=item * Search CPAN

L

=back


=head1 ACKNOWLEDGEMENTS


=head1 LICENSE AND COPYRIGHT

This software is copyright (c) 2020 by Arnaud (Arhuman) ASSAD.

This is free software; you can redistribute it and/or modify it under
the same terms as the Perl 5 programming language system itself.


=cut

1;    # End of InfluxDB::Client::Simple