scripts/radsqlrelay

   1 #!/usr/bin/perl
   2 ##
   3 ##  radsqlrelay.pl      This program tails a SQL logfile and forwards
   4 ##                      the queries to a database server. Used to
   5 ##                      replicate accounting records to one (central)
   6 ##                      database, even if the database has extended
   7 ##                      downtime.
   8 ##
   9 ##  Version:    $Id$
  10 ##
  11 ##  Author:     Nicolas Baradakis <nicolas.baradakis@cegetel.net>
  12 ##
  13 ##  Copyright (C) 2005 Cegetel
  14 ##
  15 ##  This program is free software; you can redistribute it and/or
  16 ##  modify it under the terms of the GNU General Public License
  17 ##  as published by the Free Software Foundation; either version 2
  18 ##  of the License, or (at your option) any later version.
  19 ##
  20 ##  This program is distributed in the hope that it will be useful,
  21 ##  but WITHOUT ANY WARRANTY; without even the implied warranty of
  22 ##  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  23 ##  GNU General Public License for more details.
  24 ##
  25 ##  You should have received a copy of the GNU General Public License
  26 ##  along with this program; if not, write to the Free Software
  27 ##  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  28 ##
  29
  30 use DBI;
  31 use Fcntl;
  32 use Getopt::Std;
  33 use POSIX qw(:unistd_h :errno_h);
  34
  35 use warnings;
  36 use strict;
  37
  38 my $maxcollect = 100;    # tunable, works for MySQL!
  39
  40 my $lastinsert;
  41 my @values;
  42
  43 my $need_exit = 0;
  44
  45 sub got_signal()
  46 {
  47     $need_exit = 1;
  48 }
  49
  50 # /!\ OS-dependent structure
  51 # Linux struct flock
  52 #   short l_type;
  53 #   short l_whence;
  54 #   off_t l_start;
  55 #   off_t l_len;
  56 #   pid_t l_pid;
  57 # c2ph says: typedef='s2 l2 i', sizeof=16
  58 my $FLOCK_STRUCT = 's2l2i';
  59
  60 sub setlock($;$$)
  61 {
  62     my ($fh, $start, $len) = @_;
  63     $start = 0 unless defined $start;
  64     $len = 0 unless defined $len;
  65
  66                                     #type     whence    start   till  pid
  67     my $packed = pack($FLOCK_STRUCT, F_WRLCK, SEEK_SET, $start, $len, 0);
  68     if (fcntl($fh, F_SETLKW, $packed)) { return 1 }
  69     else { return 0 }
  70 }
  71
  72 sub usage()
  73 {
  74     print STDERR <<HERE;
  75 usage: radsqlrelay [options] file_path
  76 options:
  77         -?              Print this help message.
  78         -1              One-shot mode: push the file to database and exit.
  79         -b database     Name of the database to use.
  80         -d sql_driver   Driver to use: mysql, pg, oracle.
  81         -f file         Read password from file, instead of command line.
  82         -h host         Connect to host.
  83         -P port         Port number to use for connection.
  84         -p passord      Password to use when connecting to server.
  85         -u user         User for login.
  86         -x              Turn on debugging.
  87 HERE
  88 }
  89
  90 sub connect_wait($)
  91 {
  92     my $dbinfo = shift;
  93     my $dbh;
  94     while (!$dbh) {
  95         $dbh = DBI->connect($dbinfo->{base}, $dbinfo->{user}, $dbinfo->{pass},
  96                             { RaiseError => 0, PrintError => 0,
  97                               AutoCommit => 1 });
  98         sleep (1) if !$dbh;
  99         exit if $need_exit;
 100     }
 101     $dbinfo->{handle} = $dbh;
 102 }
 103
 104
 105
 106 sub process_file($$)
 107 {
 108     my ($dbinfo, $path) = @_;
 109
 110     sub do_inserts($) {
 111         my $dbinfo = shift;
 112         if (scalar(@values) > 0) {
 113             my $query = $lastinsert . " ";
 114             $query .= join(" ), ( ",@values);
 115             $query .= " );";
 116             do_query($dbinfo,$query);
 117         }
 118         @values = ();
 119     }
 120
 121     sub do_query($$) {
 122         my ($dbinfo,$query) = @_;
 123         until ($dbinfo->{handle}->do($query)) {
 124             print $dbinfo->{handle}->errstr."\n";
 125             if ($dbinfo->{handle}->ping) {
 126                 sleep (1);
 127             } else {
 128                 print "error: Lost connection to database\n";
 129                 $dbinfo->{handle}->disconnect;
 130                 connect_wait($dbinfo);
 131             }
 132         }
 133     }
 134
 135     unless (-e $path.'.work') {
 136         until (rename($path, $path.'.work')) {
 137             if ($! == ENOENT) {
 138                 sleep(1);
 139                 return if $need_exit;
 140             } else {
 141                 print STDERR "error: Couldn't move $path to $path.work: $!\n";
 142                 exit 1;
 143             }
 144         }
 145     }
 146
 147     open(FILE, "+< $path.work") or die "error: Couldn't open $path.work: $!\n";
 148     setlock(\*FILE) or die "error: Couldn't lock $path.work: $!\n";
 149
 150     $lastinsert = "";
 151     @values = ();
 152
 153     while (<FILE>) {
 154         chomp (my $line = $_);
 155
 156         if (!($line =~ /^\s*insert\s+into\s+`?\w+`?\s+(?:\(.*?\)\s+)?
 157                             values\s*\(.*\)\s*;\s*$/ix)) {
 158             # This is no INSERT, so start new collection
 159             do_inserts($dbinfo);
 160             $lastinsert = "";
 161             # must output this line
 162             do_query($dbinfo, "$line");
 163
 164         } else {
 165             # This is an INSERT, so collect it
 166             my $insert = $line;
 167             my $values = $line;
 168             $insert =~ s/^\s*(insert\s+into\s+`?\w+`?\s+(?:\(.*?\)\s+)?
 169                               values\s*\().*\)\s*;\s*$/$1/ix;
 170             $values =~ s/^\s*insert\s+into\s+`?\w+`?\s+(?:\(.*?\)\s+)?
 171                              values\s*\((.*)\)\s*;\s*$/$1/ix;
 172
 173             if (($lastinsert ne "") && ($insert ne $lastinsert)) {
 174                 # This is different from the last one
 175                 do_inserts($dbinfo);
 176             }
 177             push(@values, $values);
 178             $lastinsert = $insert; # start new collection
 179         }
 180
 181         # limit to $maxcollect collected lines
 182         if (scalar(@values) >= $maxcollect) {
 183             do_inserts($dbinfo);
 184         }
 185     }
 186
 187     # Cleanup
 188     do_inserts($dbinfo);
 189
 190     unlink($path.'.work');
 191     close(FILE); # and unlock
 192 }
 193
 194 # sub main()
 195
 196 my %args = (
 197             b => 'radius',
 198             d => 'mysql',
 199             h => 'localhost',
 200             p => 'radius',
 201             u => 'radius',
 202 );
 203 my $ret = getopts("b:d:f:h:P:p:u:x1?", \%args);
 204 if (!$ret or @ARGV != 1) {
 205     usage();
 206     exit 1;
 207 }
 208 if ($args{'?'}) {
 209     usage();
 210     exit 0;
 211 }
 212
 213 my $data_source;
 214 if (lc($args{d}) eq 'mysql') {
 215     $data_source = "DBI:mysql:database=$args{b};host=$args{h}";
 216 } elsif (lc($args{d}) eq 'pg') {
 217     $data_source = "DBI:Pg:dbname=$args{b};host=$args{h}";
 218 } elsif (lc($args{d}) eq 'oracle') {
 219     $data_source = "DBI:Oracle:$args{b}";
 220     # Oracle does not conform to the SQL standard for multirow INSERTs
 221     $maxcollect = 1;
 222 } else {
 223     print STDERR "error: SQL driver not supported yet: $args{d}\n";
 224     exit 1;
 225 }
 226 $data_source .= ";port=$args{P}" if $args{'P'};
 227
 228 my $pw;
 229 if($args{f}) {
 230     open(FILE, "< $args{f}") or die "error: Couldn't open $args{f}: $!\n";
 231     $pw = <FILE>;
 232     chomp($pw);
 233     close(FILE);
 234 } else {
 235     # args{p} is always defined.
 236     $pw = $args{p};
 237 }
 238
 239 $SIG{INT} = \&got_signal;
 240 $SIG{TERM} = \&got_signal;
 241
 242 my %dbinfo = (
 243               base => $data_source,
 244               user => $args{u},
 245               pass => $pw,
 246 );
 247 connect_wait(\%dbinfo);
 248
 249 my $path = shift @ARGV;
 250
 251 until ($need_exit) {
 252     process_file(\%dbinfo, $path);
 253     last if ($args{1} || $need_exit);
 254     sleep(10);
 255 }
 256
 257 $dbinfo{handle}->disconnect;
 258