File Coverage

blib/lib/ETL/Yertl/Adapter/influxdb.pm
Criterion Covered Total %
statement 87 98 88.7
branch 16 24 66.6
condition 7 15 46.6
subroutine 14 14 100.0
pod 3 4 75.0
total 127 155 81.9


line stmt bran cond sub pod time code
1             package ETL::Yertl::Adapter::influxdb;
2             our $VERSION = '0.035';
3             # ABSTRACT: Adapter to read/write from InfluxDB time series database
4              
5             #pod =head1 SYNOPSIS
6             #pod
7             #pod my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost:8086' );
8             #pod my @points = $db->read_ts( { metric => 'db.cpu_load.1m' } );
9             #pod $db->write_ts( { metric => 'db.cpu_load.1m', value => 1.23 } );
10             #pod
11             #pod =head1 DESCRIPTION
12             #pod
13             #pod This class allows Yertl to read and write time series from L
14             #pod time series database|https://www.influxdata.com>.
15             #pod
16             #pod This adapter is used by the L command.
17             #pod
18             #pod =head2 Metric Name Format
19             #pod
20             #pod InfluxDB has databases, metrics, and fields. In Yertl, the time series
21             #pod is identified by joining the database, metric, and field with periods (C<.>).
22             #pod The field is optional, and defaults to C.
23             #pod
24             #pod # Database "foo", metric "bar", field "baz"
25             #pod yts influxdb://localhost foo.bar.baz
26             #pod
27             #pod # Database "foo", metric "bar", field "value"
28             #pod yts influxdb://localhost foo.bar
29             #pod
30             #pod =head1 SEE ALSO
31             #pod
32             #pod L, L,
33             #pod L,
34             #pod L,
35             #pod L
36             #pod
37             #pod =cut
38              
39 1     1   156561 use ETL::Yertl;
  1         2  
  1         9  
40 1     1   371 use Net::Async::HTTP;
  1         38235  
  1         32  
41 1     1   6 use URI;
  1         2  
  1         18  
42 1     1   240 use JSON::MaybeXS qw( decode_json );
  1         3882  
  1         52  
43 1     1   7 use List::Util qw( first );
  1         2  
  1         41  
44 1     1   5 use IO::Async::Loop;
  1         2  
  1         15  
45 1     1   282 use Time::Piece ();
  1         6141  
  1         1020  
46              
47             #pod =method new
48             #pod
49             #pod my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost' );
50             #pod my $db = ETL::Yertl::Adapter::influxdb->new( 'influxdb://localhost:8086' );
51             #pod
52             #pod Construct a new InfluxDB adapter for the database on the given host and port.
53             #pod Port is optional and defaults to C<8086>.
54             #pod
55             #pod =cut
56              
57             sub new {
58 5     5 1 27585 my $class = shift;
59              
60 5         10 my %args;
61 5 100       13 if ( @_ == 1 ) {
62 3 100       19 if ( $_[0] =~ m{://([^:]+)(?::([^/]+))?} ) {
63 2         10 ( $args{host}, $args{port} ) = ( $1, $2 );
64             }
65             }
66             else {
67 2         7 %args = @_;
68             }
69              
70 5 100       20 die "Host is required" unless $args{host};
71              
72 4   100     17 $args{port} ||= 8086;
73              
74 4         10 return bless \%args, $class;
75             }
76              
77             sub _loop {
78 2     2   4 my ( $self ) = @_;
79 2   33     14 return $self->{_loop} ||= IO::Async::Loop->new;
80             }
81              
82             sub client {
83 2     2 0 5 my ( $self ) = @_;
84 2   33     7 return $self->{http_client} ||= do {
85 2         18 my $http = Net::Async::HTTP->new;
86 2         157 $self->_loop->add( $http );
87 2         116 $http;
88             };
89             }
90              
91             #pod =method read_ts
92             #pod
93             #pod my @points = $db->read_ts( $query );
94             #pod
95             #pod Read a time series from the database. C<$query> is a hash reference
96             #pod with the following keys:
97             #pod
98             #pod =over
99             #pod
100             #pod =item metric
101             #pod
102             #pod The time series to read. For InfluxDB, this is the database, metric, and
103             #pod field separated by dots (C<.>). Field defaults to C.
104             #pod
105             #pod =item start
106             #pod
107             #pod An ISO8601 date/time for the start of the series points to return,
108             #pod inclusive.
109             #pod
110             #pod =item end
111             #pod
112             #pod An ISO8601 date/time for the end of the series points to return,
113             #pod inclusive.
114             #pod
115             #pod =item tags
116             #pod
117             #pod An optional hashref of tags. If specified, only points matching all of
118             #pod these tags will be returned.
119             #pod
120             #pod =back
121             #pod
122             #pod =cut
123              
124             sub read_ts {
125 1     1 1 573 my ( $self, $query ) = @_;
126 1         3 my $metric = $query->{ metric };
127 1         4 ( my $db, $metric, my $field ) = split /\./, $metric;
128 1   50     3 $field ||= "value";
129              
130 1         4 my $q = sprintf 'SELECT "%s" FROM "%s"', $field, $metric;
131 1         2 my @where;
132 1         2 my $tags = $query->{ tags };
133 1 50 33     4 if ( $tags && keys %$tags ) {
134 0         0 push @where, map { sprintf q{"%s"='%s'}, $_, $tags->{ $_ } } keys %$tags;
  0         0  
135             }
136 1 50       4 if ( my $start = $query->{start} ) {
137 0         0 push @where, qq{time >= '$start'};
138             }
139 1 50       3 if ( my $end = $query->{end} ) {
140 0         0 push @where, qq{time <= '$end'};
141             }
142 1 50       3 if ( @where ) {
143 0         0 $q .= ' WHERE ' . join " AND ", @where;
144             }
145              
146 1         9 my $url = URI->new( sprintf 'http://%s:%s/query', $self->{host}, $self->{port} );
147 1         5086 $url->query_form( db => $db, q => $q );
148              
149             #; say "Fetching $url";
150 1         169 my $res = $self->client->GET( $url )->get;
151              
152             #; say $res->decoded_content;
153 1 50       213 if ( $res->is_error ) {
154 0         0 die sprintf "Error fetching metric '%s': " . $res->decoded_content . "\n", $metric;
155             }
156              
157 1         14 my $result = decode_json( $res->decoded_content );
158 1         1470 my @points;
159 1         2 for my $series ( map @{ $_->{series} }, @{ $result->{results} } ) {
  1         3  
  1         3  
160 1     1   5 my $time_i = first { $series->{columns}[$_] eq 'time' } 0..$#{ $series->{columns} };
  1         3  
  1         5  
161 1     2   5 my $value_i = first { $series->{columns}[$_] eq $field } 0..$#{ $series->{columns} };
  2         4  
  1         3  
162              
163             push @points, map {
164             +{
165 2 50       13 metric => join( ".", $db, $series->{name}, ( $field ne 'value' ? ( $field ) : () ) ),
166             timestamp => $_->[ $time_i ],
167             value => $_->[ $value_i ],
168             }
169 1         3 } @{ $series->{values} };
  1         2  
170             }
171              
172 1         11 return @points;
173             }
174              
175             #pod =method write_ts
176             #pod
177             #pod $db->write_ts( @points );
178             #pod
179             #pod Write time series points to the database. C<@points> is an array
180             #pod of hashrefs with the following keys:
181             #pod
182             #pod =over
183             #pod
184             #pod =item metric
185             #pod
186             #pod The metric to write. For InfluxDB, this is the database, metric,
187             #pod and field separated by dots (C<.>). Field defaults to C.
188             #pod
189             #pod =item timestamp
190             #pod
191             #pod An ISO8601 timestamp. Optional. Defaults to the current time on the
192             #pod InfluxDB server.
193             #pod
194             #pod =item value
195             #pod
196             #pod The metric value.
197             #pod
198             #pod =back
199             #pod
200             #pod =cut
201              
202             sub write_ts {
203 1     1 1 45 my ( $self, @points ) = @_;
204              
205 1         2 my %db_lines;
206 1         3 for my $point ( @points ) {
207 2         9 my ( $db, $metric, $field ) = split /\./, $point->{metric};
208 2         3 my $tags = '';
209 2 50       5 if ( $point->{tags} ) {
210 0         0 $tags = join ",", '', map { join "=", $_, $point->{tags}{$_} } keys %{ $point->{tags} };
  0         0  
  0         0  
211             }
212              
213 2         2 my $ts = '';
214 2 100       5 if ( $point->{timestamp} ) {
215 1         2 $point->{timestamp} =~ s/[.]\d+Z?$//; # We do not support nanoseconds
216             $ts = " " . (
217 1         8 Time::Piece->strptime( $point->{timestamp}, '%Y-%m-%dT%H:%M:%S' )->epoch * 10**9
218             );
219             }
220              
221 2         19 push @{ $db_lines{ $db } }, sprintf '%s%s %s=%s%s',
222             $metric, $tags, $field || "value",
223 2   50     130 $point->{value}, $ts;
224             }
225              
226 1         3 for my $db ( keys %db_lines ) {
227 1         2 my @lines = @{ $db_lines{ $db } };
  1         4  
228 1         3 my $body = join "\n", @lines;
229 1         6 my $url = URI->new( sprintf 'http://%s:%s/write?db=%s', $self->{host}, $self->{port}, $db );
230 1         65 my $res = $self->client->POST( $url, $body, content_type => 'text/plain' )->get;
231 1 50       130 if ( $res->is_error ) {
232 0         0 my $result = decode_json( $res->decoded_content );
233 0         0 die "Error writing metric '%s': $result->{error}\n";
234             }
235             }
236              
237 1         11 return;
238             }
239              
240             1;
241              
242             __END__