view release on metacpan or  search on metacpan

lib/ETL/Yertl/Adapter/influxdb.pm  view on Meta::CPAN

#pod
#pod     my @points = $db->read_ts( $query );
#pod
#pod Read a time series from the database. C<$query> is a hash reference
#pod with the following keys:
#pod
#pod =over
#pod
#pod =item metric
#pod
#pod The time series to read. For InfluxDB, this is the database, metric, and
#pod field separated by dots (C<.>). Field defaults to C<value>.
#pod
#pod =item start
#pod
#pod An ISO8601 date/time for the start of the series points to return,
#pod inclusive.
#pod
#pod =item end
#pod
#pod An ISO8601 date/time for the end of the series points to return,
#pod inclusive.
#pod
#pod =item tags
#pod
#pod An optional hashref of tags. If specified, only points matching all of
#pod these tags will be returned.
#pod
#pod =back
#pod
#pod =cut

sub read_ts {
    my ( $self, $query ) = @_;
    my $metric = $query->{ metric };
    ( my $db, $metric, my $field ) = split /\./, $metric;
    $field ||= "value";

    my $q = sprintf 'SELECT "%s" FROM "%s"', $field, $metric;
    my @where;
    my $tags = $query->{ tags };
    if ( $tags && keys %$tags ) {
        push @where, map { sprintf q{"%s"='%s'}, $_, $tags->{ $_ } } keys %$tags;
    }
    if ( my $start = $query->{start} ) {
        push @where, qq{time >= '$start'};
    }
    if ( my $end = $query->{end} ) {
        push @where, qq{time <= '$end'};
    }
    if ( @where ) {
        $q .= ' WHERE ' . join " AND ", @where;
    }

    my $url = URI->new( sprintf 'http://%s:%s/query', $self->{host}, $self->{port} );
    $url->query_form( db => $db, q => $q );

    #; say "Fetching $url";
    my $res = $self->client->GET( $url )->get;

    #; say $res->decoded_content;
    if ( $res->is_error ) {
        die sprintf "Error fetching metric '%s': " . $res->decoded_content . "\n", $metric;
    }

    my $result = decode_json( $res->decoded_content );
    my @points;
    for my $series ( map @{ $_->{series} }, @{ $result->{results} } ) {
        my $time_i = first { $series->{columns}[$_] eq 'time' } 0..$#{ $series->{columns} };
        my $value_i = first { $series->{columns}[$_] eq $field } 0..$#{ $series->{columns} };

        push @points, map {
            +{
                metric => join( ".", $db, $series->{name}, ( $field ne 'value' ? ( $field ) : () ) ),
                timestamp => $_->[ $time_i ],
                value => $_->[ $value_i ],
            }
        } @{ $series->{values} };
    }

    return @points;
}

#pod =method write_ts
#pod
#pod     $db->write_ts( @points );
#pod
#pod Write time series points to the database. C<@points> is an array
#pod of hashrefs with the following keys:
#pod
#pod =over
#pod
#pod =item metric
#pod
#pod The metric to write. For InfluxDB, this is the database, metric,
#pod and field separated by dots (C<.>). Field defaults to C<value>.
#pod
#pod =item timestamp
#pod
#pod An ISO8601 timestamp or UNIX epoch time. Optional. Defaults to the
#pod current time.
#pod
#pod =item value
#pod
#pod The metric value.
#pod
#pod =back
#pod
#pod =cut

sub write_ts {
    my ( $self, @points ) = @_;

    my %db_lines;
    for my $point ( @points ) {
        my ( $db, $metric, $field ) = split /\./, $point->{metric};
        my $tags = '';
        if ( $point->{tags} ) {
            $tags = join ",", '', map { join "=", $_, $point->{tags}{$_} } keys %{ $point->{tags} };
        }

        my $ts = '';
        if ( my $epoch = $point->{timestamp} || time ) {
            if ( !looks_like_number( $epoch ) ) {
                $epoch =~ s/[.]\d+Z?$//; # We do not support nanoseconds
                $epoch = Time::Piece->strptime( $epoch, '%Y-%m-%dT%H:%M:%S' )->epoch;
            }
            $ts = " " . ( $epoch * 10**9 );
        }

        push @{ $db_lines{ $db } }, sprintf '%s%s %s=%s%s',
            $metric, $tags, $field || "value",
            $point->{value}, $ts;
    }

    for my $db ( keys %db_lines ) {
        my @lines = @{ $db_lines{ $db } };
        my $body = join "\n", @lines;
        my $url = URI->new( sprintf 'http://%s:%s/write?db=%s', $self->{host}, $self->{port}, $db );
        my $res = $self->client->POST( $url, $body, content_type => 'text/plain' )->get;
        if ( $res->is_error ) {
            my $result = decode_json( $res->decoded_content );
            die "Error writing metric '%s': $result->{error}\n";
        }
    }

    return;
}

1;

__END__

=pod

=head1 NAME

ETL::Yertl::Adapter::influxdb - Adapter to read/write from InfluxDB time series database

=head1 VERSION

version 0.044

=head1 SYNOPSIS

    my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost:8086' );
    my @points = $db->read_ts( { metric => 'db.cpu_load.1m' } );
    $db->write_ts( { metric => 'db.cpu_load.1m', value => 1.23 } );

=head1 DESCRIPTION

This class allows Yertl to read and write time series from L<the InfluxDB
time series database|https://www.influxdata.com>.

This adapter is used by the L<yts> command.

=head2 Metric Name Format

InfluxDB has databases, metrics, and fields. In Yertl, the time series
is identified by joining the database, metric, and field with periods (C<.>).
The field is optional, and defaults to C<value>.

    # Database "foo", metric "bar", field "baz"
    yts influxdb://localhost foo.bar.baz

    # Database "foo", metric "bar", field "value"
    yts influxdb://localhost foo.bar

=head1 METHODS

=head2 new

    my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost' );
    my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost:8086' );

Construct a new InfluxDB adapter for the database on the given host and port.
Port is optional and defaults to C<8086>.

=head2 read_ts

    my @points = $db->read_ts( $query );