four kids, no delays, RT#10574
[freeside.git] / FS / bin / freeside-torrus-srvderive
1 #!/usr/bin/perl -w
2
3 use strict;
4 use POSIX qw( :sys_wait_h );
5 use Sys::SigAction qw( set_sig_handler );
6 use Date::Parse;
7 use Date::Format;
8 use IPC::Run qw( run ); #timeout );
9 use FS::Daemon ':all'; #daemonize1 drop_root daemonize2 myexit logfile sig*
10 use FS::UID qw( adminsuidsetup forksuidsetup dbh driver_name );
11 use FS::Record qw( qsearch str2time_sql str2time_sql_closing concat_sql );
12 use FS::torrus_srvderive;
13
14 our $DEBUG = 2;
15 our $max_kids = 4;
16 our $search_timeout = 2*60*60; #60*60; #30*60; # 15*60 5*60; #$torrus_srvderive->last_srv_date ? 5*60 : 15*60);
17 our %kids;
18
19 my $user = shift or die &usage;
20 $FS::Daemon::PID_NEWSTYLE = 1;
21 daemonize1('torrus-srvderive');
22
23 drop_root();
24
25 adminsuidsetup($user);
26
27 logfile( "%%%FREESIDE_LOG%%%/torrus-srvderive-log.". $FS::UID::datasrc );
28
29 daemonize2();
30
31 our $conf = new FS::Conf;
32
33 die "not running: network_monitoring_system not Torrus_Internal\n"
34   unless _shouldrun();
35
36 #--
37
38 my $str2time = str2time_sql();
39 my $c = str2time_sql_closing();
40
41 my $_date = concat_sql([ 'srvexport.srv_date', "' '", 'srvexport.srv_time' ]);
42 $_date = "CAST( $_date AS TIMESTAMP )" if driver_name =~ /^Pg/i;
43 $_date = str2time_sql. $_date.  str2time_sql_closing;
44
45 my $other_date = concat_sql([ 'other.srv_date', "' '", 'other.srv_time' ]);
46 $other_date = "CAST( $other_date AS TIMESTAMP )" if driver_name =~ /^Pg/i;
47 $other_date = str2time_sql. $other_date.  str2time_sql_closing;
48
49 my $in  = concat_sql([ '?', "'_IN'" ]);
50 my $out = concat_sql([ '?', "'_OUT'" ]);
51
52 my $sql = "
53   SELECT DISTINCT srv_date, srv_time FROM srvexport
54     WHERE NOT EXISTS (
55                        SELECT 1 FROM srvexport AS other
56                          WHERE other.serviceid IN ( $in, $out )
57                            AND srvexport.srv_date = other.srv_date
58                            AND ABS( $_date - $other_date ) <= 60
59                      )
60 ";
61
62 my $orderlimit = "
63     ORDER BY srv_date, srv_time
64     LIMIT 200
65 ";
66
67 if ( driver_name =~ /mysql/i ) {
68   dbh->do('SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED');
69   dbh->commit or die dbh->errstr;
70 }
71
72 our $kids = 0;
73
74 #MAIN: while (1) {
75 while (1) {
76
77   my $found = 0;
78
79   #SERVICEID: foreach my $torrus_srvderive ( qsearch('torrus_srvderive', {}) ) {
80   foreach my $torrus_srvderive ( qsearch('torrus_srvderive', {}) ) {
81
82     &reap_kids;
83     if ( $kids >= $max_kids ) {
84       sleep 5;
85       myexit() if sigterm() || sigint();
86       redo;
87     }
88
89     defined( my $pid = fork ) or do {
90       #warn "WARNING: can't fork: $!\n";
91       #next; #don't increment the kid counter
92       die "can't fork: $!\n";
93     };
94
95     if ( $pid ) {
96       $kids++;
97       $kids{$pid} = 1;
98     } else { #kid time
99
100       #get new db handle
101       $FS::UID::dbh->{InactiveDestroy} = 1;
102
103       forksuidsetup($user);
104
105       my $serviceid = $torrus_srvderive->serviceid;
106
107       my @serviceids = $torrus_srvderive->component_serviceids;
108       exit unless @serviceids; #don't try to search for empty virtual ports
109
110       #nonlocking select statements; rows in this table never change
111       if ( driver_name =~ /mysql/i ) {
112         dbh->do('SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED');
113         dbh->commit or die dbh->errstr;
114       }
115
116       my @in = ();
117       for my $dir ('_IN', '_OUT') {
118         push @in, map dbh->quote("$_$dir"), @serviceids;
119       }
120       my $in = join(',', @in);
121
122       if ( ! $torrus_srvderive->last_srv_date ) {
123         warn "finding initial last_srv_date for $serviceid\n" if $DEBUG;
124         my $dsql = "SELECT srv_date FROM srvexport WHERE serviceid IN ($in)
125                     ORDER BY srv_date LIMIT 1";
126         my $dsth = dbh->prepare($dsql) or die $DBI::errstr;
127         $dsth->execute or die $dsth->errstr;
128         my $date = $dsth->fetchrow_arrayref->[0];
129         if ( $date ) {
130           warn "found initial last_srv_date of $date; updating $serviceid\n"
131             if $DEBUG;
132           $torrus_srvderive->last_srv_date($date);
133           my $error = $torrus_srvderive->replace;
134           die $error if $error;
135         } else {
136           warn "no initial last_srv_date for $serviceid; skipping\n" if $DEBUG;
137           exit;
138         }
139       }
140
141       #alternate (hopefully more efficient) path: if we're more than a day
142       #behind, call native torrus srvderive
143       my $last_srv_date = $torrus_srvderive->last_srv_date;
144       my $currdate = time2str('%Y-%m-%d', time);
145       if ( $last_srv_date ne $currdate
146            && time - str2time($last_srv_date) > (24+6)*60*60 #6 hour window for
147          ) {                                                 #yesterday
148         warn "serviceid $serviceid last_srv_date != $currdate\n" if $DEBUG;
149
150         #delete records from last_srv_date
151         my $dsql =
152           'DELETE FROM srvexport WHERE srv_date = ? AND serviceid IN(?,?)';
153         my $dsth = dbh->prepare($dsql) or die $DBI::errstr;
154         warn "$dsql $last_srv_date ${serviceid}_IN ${serviceid}_OUT\n"
155           if $DEBUG;
156         $dsth->execute($last_srv_date, $serviceid.'_IN', $serviceid.'_OUT')
157           or die $dsth->errstr;
158         dbh->commit or die dbh->errstr;
159
160         foreach my $dir ('_IN','_OUT') {
161
162           #XX TORRUS_PATH
163           my @cmd = ( '/usr/local/bin/torrus', 'srvderive',
164                         "--start=$last_srv_date", "--end=$currdate",
165                         "--out=$serviceid$dir", "--func=SUM",
166                         map "$_$dir", @serviceids
167                     );
168
169           warn "calling native srvderive: ". join(' ', @cmd). "\n" if $DEBUG;
170           my $out_and_err = '';
171           run( \@cmd, '>&', \$out_and_err ) #verbose?  timeout?
172             or die "error calling native srvder: $out_and_err\n";
173
174         }
175         warn "native srvderive done" if $DEBUG;
176
177         $torrus_srvderive->last_srv_date($currdate);
178         my $error = $torrus_srvderive->replace;
179         die $error if $error;
180
181         #don't bother trying today's immediately
182         warn "done with $serviceid for now\n" if $DEBUG;
183         exit; #end-of-kid
184
185       }
186
187       my $ssql = "
188         $sql AND EXISTS (
189                          SELECT 1 FROM srvexport AS other
190                            WHERE other.serviceid IN ($in)
191                              AND srvexport.srv_date = other.srv_date
192                              AND ABS( $_date - $other_date ) <= 60
193                        )
194       ";
195
196       $ssql .= " AND srv_date >= '". $torrus_srvderive->last_srv_date. "' "
197         if $torrus_srvderive->last_srv_date;
198
199       $ssql .= $orderlimit;
200
201       warn "searching for times to add $serviceid\n" if $DEBUG;
202       warn $ssql if $DEBUG > 2;
203       my $sth = dbh->prepare($ssql) or die $DBI::errstr; #better recovery here?
204
205       eval {
206         my $h = set_sig_handler( 'ALRM', sub { die "_timeout\n"; } );
207         alarm($search_timeout);
208         $sth->execute($serviceid, $serviceid) or die $sth->errstr;
209         alarm(0);
210       };
211       alarm(0);
212       
213       if ( $@ && $@ eq "_timeout\n" ) {
214         #warn "search timed out; reconnecting and restarting\n";
215         warn "search timed out\n";
216         dbh->clone()->do("KILL QUERY ". dbh->{"mysql_thread_id"})
217           if driver_name =~ /mysql/i;
218         dbh->rollback; #or die dbh->errstr;
219         #adminsuidsetup($user);
220         #next SERVICEID; #MAIN;
221         exit;
222       } elsif ( $@ ) {
223         die $@;
224       }
225
226       warn "search for $serviceid finished; checking results\n" if $DEBUG;
227
228       my $prev = 0;
229       while ( my $row = $sth->fetchrow_arrayref ) {
230         last if sigterm() || sigint();
231
232         my( $srv_date, $srv_time ) = @$row;
233         my $cur = str2time( "$srv_date $srv_time" );
234         next if $cur-$prev <= 60;
235         last if time - $cur <= 300;
236
237         warn "no $serviceid for $srv_date $srv_time; adding\n"
238           if $DEBUG;
239         $found++;
240
241         for my $dir ('_IN', '_OUT') {
242
243           my $sin = join(',', map dbh->quote("$_$dir"), @serviceids);
244
245           my $srv_date = time2str('%Y-%m-%d', $cur);
246
247           my $sum = "
248             SELECT COALESCE(SUM(value),0) FROM srvexport AS other
249               WHERE other.serviceid IN ($sin)
250                 AND other.srv_date = '$srv_date'
251                 AND ABS( $cur - $other_date ) <= 60
252           ";
253           my $ssth = dbh->prepare($sum) or die $DBI::errstr;
254           $ssth->execute or die $ssth->errstr; #better recovery?
255           my $value = $ssth->fetchrow_arrayref->[0];
256
257           my $isql = "
258             INSERT INTO srvexport (srv_date, srv_time, serviceid, value, intvl)
259               VALUES ( ?, ?, ?, ?, ? )
260           ";
261           my @param = ( $srv_date,
262                         time2str('%X', $cur), #srv_time
263                         "$serviceid$dir",     #serviceid
264                         $value,
265                         300,                  #intvl ... 
266                       );
267           warn $isql. ' with param '. join(',',@param). "\n"
268             if $DEBUG > 2;
269
270           my $isth = dbh->prepare($isql) or die $DBI::errstr; #better recovery?
271
272           #stupid mysql deadlocks all the time on insert, so we need to recover
273           unless ( $isth->execute(@param) ) {
274             #warn "Error inserting data for $serviceid$dir (restarting): ".
275             #     $isth->errstr;
276             warn "Error inserting data for $serviceid$dir: ". $isth->errstr;
277             dbh->rollback; #or die dbh->errstr;
278             #sleep 5;
279             #next SERVICEID; #MAIN;
280             exit;
281           }
282                           
283         }
284
285         if ( $srv_date ne $torrus_srvderive->last_srv_date ) {
286           warn "updating last_srv_date of $serviceid to $srv_date\n" if $DEBUG;
287           $torrus_srvderive->last_srv_date($srv_date);
288           my $error = $torrus_srvderive->replace;
289           die $error if $error;
290         }
291         dbh->commit or die dbh->errstr;
292
293         $prev = $cur;
294       }
295       warn "done with $serviceid\n" if $DEBUG;
296
297       exit;
298       #end-of-kid
299     }
300
301   } #foreach my $torrus_srvderive
302   dbh->commit or die dbh->errstr;
303
304   myexit() if sigterm() || sigint();
305   warn "restarting main loop\n" if $DEBUG > 1;
306   #sleep 60 unless $found;
307 }
308
309 sub _shouldrun {
310      $conf->exists('network_monitoring_system')
311   && $conf->config('network_monitoring_system') eq 'Torrus_Internal';
312 }
313
314 sub usage { 
315   die "Usage:\n\n  freeside-cdrrewrited user\n";
316 }
317
318 sub reap_kids {
319   foreach my $pid ( keys %kids ) {
320     my $kid = waitpid($pid, WNOHANG);
321     if ( $kid > 0 ) {
322       $kids--;
323       delete $kids{$kid};
324     }
325   }
326 }
327
328 =head1 NAME
329
330 freeside-torrus-srvderive - Freeside's Torrus virtual port daemon.
331
332 =head1 SYNOPSIS
333
334   freeside-torrus-srvderive
335
336 =head1 DESCRIPTION
337
338 Runs continuously, searches for samples in the srvexport table which do not
339 have an entry for combined virtual ports, and adds them.
340
341 =head1 SEE ALSO
342
343 =cut
344
345 1;
346