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__ |