File: mk-slave-delay

package info (click to toggle)
maatkit 1877-1
links: PTS, VCS
area: main
in suites: lenny
size: 1,428 kB
ctags: 1,190
sloc: perl: 27,341; cpp: 70; makefile: 38
file content (1221 lines) | stat: -rwxr-xr-x 39,756 bytes
#!/usr/bin/perl

# This is mk-slave-delay, a program that makes a MySQL slave lag its master.
#
# This program is copyright (c) 2007 Sergey Zhuravlev and Baron Schwartz.
# Feedback and improvements are welcome.
#
# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
# WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
# MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
# systems, you can issue `man perlgpl' or `man perlartistic' to read these
# licenses.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place, Suite 330, Boston, MA  02111-1307  USA.

use strict;
use warnings FATAL => 'all';

our $VERSION = '1.0.6';
our $DISTRIB = '1877';
our $SVN_REV = sprintf("%d", map { $_ || 0 } q$Revision: 1871 $ =~ m/(\d+)/g);

# ###########################################################################
# OptionParser package 1844
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package OptionParser;

use Getopt::Long;
use List::Util qw(max);
use English qw(-no_match_vars);

sub new {
   my ( $class, @opts ) = @_;
   my %key_seen;
   my %long_seen;
   my %key_for;
   my %defaults;
   my @mutex;
   my @atleast1;
   my %long_for;
   my %disables;
   my %copyfrom;
   unshift @opts,
      { s => 'help',    d => 'Show this help message' },
      { s => 'version', d => 'Output version information and exit' };
   foreach my $opt ( @opts ) {
      if ( ref $opt ) {
         my ( $long, $short ) = $opt->{s} =~ m/^([\w-]+)(?:\|([^!+=]*))?/;
         $opt->{k} = $short || $long;
         $key_for{$long} = $opt->{k};
         $long_for{$opt->{k}} = $long;
         $long_for{$long} = $long;
         $opt->{l} = $long;
         die "Duplicate option $opt->{k}" if $key_seen{$opt->{k}}++;
         die "Duplicate long option $opt->{l}" if $long_seen{$opt->{l}}++;
         $opt->{t} = $short;
         $opt->{n} = $opt->{s} =~ m/!/;
         $opt->{g} ||= 'o';
         if ( (my ($y) = $opt->{s} =~ m/=([mdHhAaz])/) ) {
            $ENV{MKDEBUG} && _d("Option $opt->{k} type: $y");
            $opt->{y} = $y;
            $opt->{s} =~ s/=./=s/;
         }
         if ( $opt->{d} =~ m/required/ ) {
            $opt->{r} = 1;
            $ENV{MKDEBUG} && _d("Option $opt->{k} is required");
         }
         if ( (my ($def) = $opt->{d} =~ m/default\b(?: ([^)]+))?/) ) {
            $defaults{$opt->{k}} = defined $def ? $def : 1;
            $ENV{MKDEBUG} && _d("Option $opt->{k} has a default");
         }
         if ( (my ($dis) = $opt->{d} =~ m/(disables .*)/) ) {
            $disables{$opt->{k}} = [ $class->get_participants($dis) ];
            $ENV{MKDEBUG} && _d("Option $opt->{k} $dis");
         }
      }
      else { # It's an instruction.

         if ( $opt =~ m/at least one|mutually exclusive|one and only one/ ) {
            my @participants = map {
                  die "No such option '$_' in $opt" unless $long_for{$_};
                  $long_for{$_};
               } $class->get_participants($opt);
            if ( $opt =~ m/mutually exclusive|one and only one/ ) {
               push @mutex, \@participants;
               $ENV{MKDEBUG} && _d(@participants, ' are mutually exclusive');
            }
            if ( $opt =~ m/at least one|one and only one/ ) {
               push @atleast1, \@participants;
               $ENV{MKDEBUG} && _d(@participants, ' require at least one');
            }
         }
         elsif ( $opt =~ m/default to/ ) {
            my @participants = map {
                  die "No such option '$_' in $opt" unless $long_for{$_};
                  $key_for{$_};
               } $class->get_participants($opt);
            $copyfrom{$participants[0]} = $participants[1];
            $ENV{MKDEBUG} && _d(@participants, ' copy from each other');
         }

      }
   }

   if ( $ENV{MKDEBUG} ) {
      my $text = do {
         local $RS = undef;
         open my $fh, "<", $PROGRAM_NAME
            or die "Can't open $PROGRAM_NAME: $OS_ERROR";
         <$fh>;
      };
      my %used = map { $_ => 1 } $text =~ m/\$opts\{'?([\w-]+)'?\}/g;
      my @unused;
      my @undefined;
      my %option_exists;
      foreach my $opt ( @opts ) {
         next unless ref $opt;
         my $key = $opt->{k};
         $option_exists{$key}++;
         next if $opt->{l} =~ m/^(?:help|version|defaults-file|database|charset
                                    |password|port|socket|user|host)$/x
              || $disables{$key};
         push @unused, $key unless $used{$key};
      }
      foreach my $key ( keys %used ) {
         push @undefined, $key unless $option_exists{$key};
      }
      if ( @unused || @undefined ) {
         die "The following command-line options are unused: "
            . join(',', @unused)
            . ' The following are undefined: '
            . join(',', @undefined);
      }
   }

   foreach my $dis ( keys %disables ) {
      $disables{$dis} = [ map {
            die "No such option '$_' while processing $dis" unless $long_for{$_};
            $long_for{$_};
         } @{$disables{$dis}} ];
   }

   return bless {
      specs => [ grep { ref $_ } @opts ],
      notes => [],
      instr => [ grep { !ref $_ } @opts ],
      mutex => \@mutex,
      defaults => \%defaults,
      long_for => \%long_for,
      atleast1 => \@atleast1,
      disables => \%disables,
      key_for  => \%key_for,
      copyfrom => \%copyfrom,
      strict   => 1,
      groups   => [ { k => 'o', d => 'Options' } ],
   }, $class;
}

sub get_participants {
   my ( $self, $str ) = @_;
   my @participants;
   foreach my $thing ( $str =~ m/(--?[\w-]+)/g ) {
      if ( (my ($long) = $thing =~ m/--(.+)/) ) {
         push @participants, $long;
      }
      else {
         foreach my $short ( $thing =~ m/([^-])/g ) {
            push @participants, $short;
         }
      }
   }
   $ENV{MKDEBUG} && _d("Participants for $str: ", @participants);
   return @participants;
}

sub parse {
   my ( $self, %defaults ) = @_;
   my @specs = @{$self->{specs}};
   my %factor_for = (k => 1_024, M => 1_048_576, G => 1_073_741_824);

   my %opt_seen;
   my %vals = %{$self->{defaults}};
   @vals{keys %defaults} = values %defaults;
   foreach my $spec ( @specs ) {
      $vals{$spec->{k}} = undef unless defined $vals{$spec->{k}};
      $opt_seen{$spec->{k}} = 1;
   }

   foreach my $key ( keys %defaults ) {
      die "Cannot set default for non-existent option '$key'\n"
         unless $opt_seen{$key};
   }

   Getopt::Long::Configure('no_ignore_case', 'bundling');
   GetOptions( map { $_->{s} => \$vals{$_->{k}} } @specs )
      or $self->error('Error parsing options');

   if ( $vals{version} ) {
      my $prog = $self->prog;
      printf("%s  Ver %s Distrib %s Changeset %s\n",
         $prog, $main::VERSION, $main::DISTRIB, $main::SVN_REV);
      exit(0);
   }

   if ( @ARGV && $self->{strict} ) {
      $self->error("Unrecognized command-line options @ARGV");
   }

   foreach my $dis ( grep { defined $vals{$_} } keys %{$self->{disables}} ) {
      my @disses = map { $self->{key_for}->{$_} } @{$self->{disables}->{$dis}};
      $ENV{MKDEBUG} && _d("Unsetting options: ", @disses);
      @vals{@disses} = map { undef } @disses;
   }

   foreach my $spec ( grep { $_->{r} } @specs ) {
      if ( !defined $vals{$spec->{k}} ) {
         $self->error("Required option --$spec->{l} must be specified");
      }
   }

   foreach my $mutex ( @{$self->{mutex}} ) {
      my @set = grep { defined $vals{$self->{key_for}->{$_}} } @$mutex;
      if ( @set > 1 ) {
         my $note = join(', ',
            map { "--$self->{long_for}->{$_}" }
                @{$mutex}[ 0 .. scalar(@$mutex) - 2] );
         $note .= " and --$self->{long_for}->{$mutex->[-1]}"
               . " are mutually exclusive.";
         $self->error($note);
      }
   }

   foreach my $required ( @{$self->{atleast1}} ) {
      my @set = grep { defined $vals{$self->{key_for}->{$_}} } @$required;
      if ( !@set ) {
         my $note = join(', ',
            map { "--$self->{long_for}->{$_}" }
                @{$required}[ 0 .. scalar(@$required) - 2] );
         $note .= " or --$self->{long_for}->{$required->[-1]}";
         $self->error("Specify at least one of $note");
      }
   }

   foreach my $spec ( grep { $_->{y} && defined $vals{$_->{k}} } @specs ) {
      my $val = $vals{$spec->{k}};
      if ( $spec->{y} eq 'm' ) {
         my ( $num, $suffix ) = $val =~ m/(\d+)([a-z])?$/;
         if ( !$suffix ) {
            my ( $s ) = $spec->{d} =~ m/\(suffix (.)\)/;
            $suffix = $s || 's';
            $ENV{MKDEBUG} && _d("No suffix given; using $suffix for $spec->{k} "
               . "(value: '$val')");
         }
         if ( $suffix =~ m/[smhd]/ ) {
            $val = $suffix eq 's' ? $num            # Seconds
                 : $suffix eq 'm' ? $num * 60       # Minutes
                 : $suffix eq 'h' ? $num * 3600     # Hours
                 :                  $num * 86400;   # Days
            $vals{$spec->{k}} = $val;
            $ENV{MKDEBUG} && _d("Setting option $spec->{k} to $val");
         }
         else {
            $self->error("Invalid --$spec->{l} argument");
         }
      }
      elsif ( $spec->{y} eq 'd' ) {
         $ENV{MKDEBUG} && _d("Parsing option $spec->{y} as a DSN");
         my $from_key = $self->{copyfrom}->{$spec->{k}};
         my $default = {};
         if ( $from_key ) {
            $ENV{MKDEBUG} && _d("Option $spec->{y} DSN copies from option $from_key");
            $default = $self->{dsn}->parse($self->{dsn}->as_string($vals{$from_key}));
         }
         $vals{$spec->{k}} = $self->{dsn}->parse($val, $default);
      }
      elsif ( $spec->{y} eq 'z' ) {
         my ($pre, $num, $factor) = $val =~ m/^([+-])?(\d+)([kMG])?$/;
         if ( defined $num ) {
            if ( $factor ) {
               $num *= $factor_for{$factor};
               $ENV{MKDEBUG} && _d("Setting option $spec->{y} to num * factor");
            }
            $vals{$spec->{k}} = ($pre || '') . $num;
         }
         else {
            $self->error("Invalid --$spec->{l} argument");
         }
      }
   }

   foreach my $spec ( grep { $_->{y} } @specs ) {
      $ENV{MKDEBUG} && _d("Treating option $spec->{k} as a list");
      my $val = $vals{$spec->{k}};
      if ( $spec->{y} eq 'H' || (defined $val && $spec->{y} eq 'h') ) {
         $vals{$spec->{k}} = { map { $_ => 1 } split(',', ($val || '')) };
      }
      elsif ( $spec->{y} eq 'A' || (defined $val && $spec->{y} eq 'a') ) {
         $vals{$spec->{k}} = [ split(',', ($val || '')) ];
      }
   }

   return %vals;
}

sub error {
   my ( $self, $note ) = @_;
   $self->{__error__} = 1;
   push @{$self->{notes}}, $note;
}

sub prog {
   (my $prog) = $PROGRAM_NAME =~ m/([.A-Za-z-]+)$/;
   return $prog || $PROGRAM_NAME;
}

sub prompt {
   my ( $self ) = @_;
   my $prog   = $self->prog;
   my $prompt = $self->{prompt} || '<options>';
   return "Usage: $prog $prompt\n";
}

sub descr {
   my ( $self ) = @_;
   my $prog = $self->prog;
   my $descr  = $prog . ' ' . ($self->{descr} || '')
          . "  For more details, please use the --help option, "
          . "or try 'perldoc $prog' for complete documentation.";
   $descr = join("\n", $descr =~ m/(.{0,80})(?:\s+|$)/g);
   $descr =~ s/ +$//mg;
   return $descr;
}

sub usage_or_errors {
   my ( $self, %opts ) = @_;
   if ( $opts{help} ) {
      print $self->usage(%opts);
      exit(0);
   }
   elsif ( $self->{__error__} ) {
      print $self->errors();
      exit(0);
   }
}

sub errors {
   my ( $self ) = @_;
   my $usage = $self->prompt() . "\n";
   if ( (my @notes = @{$self->{notes}}) ) {
      $usage .= join("\n  * ", 'Errors in command-line arguments:', @notes) . "\n";
   }
   return $usage . "\n" . $self->descr();
}

sub usage {
   my ( $self, %vals ) = @_;
   my @specs = @{$self->{specs}};

   my $maxl = max(map { length($_->{l}) + ($_->{n} ? 4 : 0)} @specs);

   my $maxs = max(0,
      map { length($_->{l}) + ($_->{n} ? 4 : 0)}
      grep { $_->{t} } @specs);

   my $lcol = max($maxl, ($maxs + 3));
   my $rcol = 80 - $lcol - 6;
   my $rpad = ' ' x ( 80 - $rcol );

   $maxs = max($lcol - 3, $maxs);

   my $usage = $self->descr() . "\n" . $self->prompt();
   foreach my $g ( @{$self->{groups}} ) {
      $usage .= "\n$g->{d}:\n";
      foreach my $spec (
         sort { $a->{l} cmp $b->{l} } grep { $_->{g} eq $g->{k} } @specs )
      {
         my $long  = $spec->{n} ? "[no]$spec->{l}" : $spec->{l};
         my $short = $spec->{t};
         my $desc  = $spec->{d};
         if ( $spec->{y} && $spec->{y} eq 'm' ) {
            my ($s) = $desc =~ m/\(suffix (.)\)/;
            $s    ||= 's';
            $desc =~ s/\s+\(suffix .\)//;
            $desc .= ".  Optional suffix s=seconds, m=minutes, h=hours, "
                   . "d=days; if no suffix, $s is used.";
         }
         $desc = join("\n$rpad", grep { $_ } $desc =~ m/(.{0,$rcol})(?:\s+|$)/g);
         $desc =~ s/ +$//mg;
         if ( $short ) {
            $usage .= sprintf("  --%-${maxs}s -%s  %s\n", $long, $short, $desc);
         }
         else {
            $usage .= sprintf("  --%-${lcol}s  %s\n", $long, $desc);
         }
      }
   }

   if ( (my @instr = @{$self->{instr}}) ) {
      $usage .= join("\n", map { "  $_" } @instr) . "\n";
   }
   if ( $self->{dsn} ) {
      $usage .= "\n" . $self->{dsn}->usage();
   }
   $usage .= "\nOptions and values after processing arguments:\n";
   foreach my $spec ( sort { $a->{l} cmp $b->{l} } @specs ) {
      my $val   = $vals{$spec->{k}};
      my $type  = $spec->{y} || '';
      my $bool  = $spec->{s} =~ m/^[\w-]+(?:\|[\w-])?!?$/;
      $val      = $bool                     ? ( $val ? 'TRUE' : 'FALSE' )
                : !defined $val             ? '(No value)'
                : $type eq 'd'              ? $self->{dsn}->as_string($val)
                : $type =~ m/H|h/           ? join(',', sort keys %$val)
                : $type =~ m/A|a/           ? join(',', @$val)
                :                             $val;
      $usage .= sprintf("  --%-${lcol}s  %s\n", $spec->{l}, $val);
   }
   return $usage;
}

sub prompt_noecho {
   shift @_ if ref $_[0] eq __PACKAGE__;
   my ( $prompt ) = @_;
   local $OUTPUT_AUTOFLUSH = 1;
   print $prompt;
   my $response;
   eval {
      require Term::ReadKey;
      Term::ReadKey::ReadMode('noecho');
      chomp($response = <STDIN>);
      Term::ReadKey::ReadMode('normal');
      print "\n";
   };
   if ( $EVAL_ERROR ) {
      die "Cannot read response; is Term::ReadKey installed? $EVAL_ERROR";
   }
   return $response;
}

sub groups {
   my ( $self, @groups ) = @_;
   push @{$self->{groups}}, @groups;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# OptionParser:$line $PID ", @_, "\n";
}

if ( $ENV{MKDEBUG} ) {
   print '# ', $^X, ' ', $], "\n";
   my $uname = `uname -a`;
   if ( $uname ) {
      $uname =~ s/\s+/ /g;
      print "# $uname\n";
   }
   printf("# %s  Ver %s Distrib %s Changeset %s line %d\n",
      $PROGRAM_NAME, ($main::VERSION || ''), ($main::DISTRIB || ''),
      ($main::SVN_REV || ''), __LINE__);
}

1;

# ###########################################################################
# End OptionParser package
# ###########################################################################

# ###########################################################################
# VersionParser package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package VersionParser;

use English qw(-no_match_vars);

sub new {
   my ( $class ) = @_;
   bless {}, $class;
}

sub parse {
   my ( $self, $str ) = @_;
   my $result = sprintf('%03d%03d%03d', $str =~ m/(\d+)/g);
   $ENV{MKDEBUG} && _d("$str parses to $result");
   return $result;
}

sub version_ge {
   my ( $self, $dbh, $target ) = @_;
   if ( !$self->{$dbh} ) {
      $self->{$dbh} = $self->parse(
         $dbh->selectrow_array('SELECT VERSION()'));
   }
   my $result = $self->{$dbh} ge $self->parse($target) ? 1 : 0;
   $ENV{MKDEBUG} && _d("$self->{$dbh} ge $target: $result");
   return $result;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# VersionParser:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End VersionParser package
# ###########################################################################

# ###########################################################################
# DSNParser package 1868
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package DSNParser;

use DBI;
use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;
use English qw(-no_match_vars);

sub new {
   my ( $class, @opts ) = @_;
   my $self = {
      opts => {
         A => {
            desc => 'Default character set',
            dsn  => 'charset',
            copy => 1,
         },
         D => {
            desc => 'Database to use',
            dsn  => 'database',
            copy => 1,
         },
         F => {
            desc => 'Only read default options from the given file',
            dsn  => 'mysql_read_default_file',
            copy => 1,
         },
         h => {
            desc => 'Connect to host',
            dsn  => 'host',
            copy => 1,
         },
         p => {
            desc => 'Password to use when connecting',
            dsn  => 'password',
            copy => 1,
         },
         P => {
            desc => 'Port number to use for connection',
            dsn  => 'port',
            copy => 1,
         },
         S => {
            desc => 'Socket file to use for connection',
            dsn  => 'mysql_socket',
            copy => 1,
         },
         u => {
            desc => 'User for login if not current user',
            dsn  => 'user',
            copy => 1,
         },
      },
   };
   foreach my $opt ( @opts ) {
      $ENV{MKDEBUG} && _d('Adding extra property ' . $opt->{key});
      $self->{opts}->{$opt->{key}} = { desc => $opt->{desc}, copy => $opt->{copy} };
   }
   return bless $self, $class;
}

sub prop {
   my ( $self, $prop, $value ) = @_;
   if ( @_ > 2 ) {
      $ENV{MKDEBUG} && _d("Setting $prop property");
      $self->{$prop} = $value;
   }
   return $self->{$prop};
}

sub parse {
   my ( $self, $dsn, $prev, $defaults ) = @_;
   if ( !$dsn ) {
      $ENV{MKDEBUG} && _d('No DSN to parse');
      return;
   }
   $ENV{MKDEBUG} && _d("Parsing $dsn");
   $prev     ||= {};
   $defaults ||= {};
   my %vals;
   my %opts = %{$self->{opts}};
   if ( $dsn !~ m/=/ && (my $p = $self->prop('autokey')) ) {
      $ENV{MKDEBUG} && _d("Interpreting $dsn as $p=$dsn");
      $dsn = "$p=$dsn";
   }
   my %hash = map { m/^(.)=(.*)$/g } split(/,/, $dsn);
   foreach my $key ( keys %opts ) {
      $ENV{MKDEBUG} && _d("Finding value for $key");
      $vals{$key} = $hash{$key};
      if ( !defined $vals{$key} && defined $prev->{$key} && $opts{$key}->{copy} ) {
         $vals{$key} = $prev->{$key};
         $ENV{MKDEBUG} && _d("Copying value for $key from previous DSN");
      }
      if ( !defined $vals{$key} ) {
         $vals{$key} = $defaults->{$key};
         $ENV{MKDEBUG} && _d("Copying value for $key from defaults");
      }
   }
   foreach my $key ( keys %hash ) {
      die "Unrecognized DSN part '$key' in '$dsn'\n"
         unless exists $opts{$key};
   }
   if ( (my $required = $self->prop('required')) ) {
      foreach my $key ( keys %$required ) {
         die "Missing DSN part '$key' in '$dsn'\n" unless $vals{$key};
      }
   }
   return \%vals;
}

sub as_string {
   my ( $self, $dsn ) = @_;
   return $dsn unless ref $dsn;
   return join(',',
      map  { "$_=" . ($_ eq 'p' ? '...' : $dsn->{$_}) }
      grep { defined $dsn->{$_} && $self->{opts}->{$_} }
      sort keys %$dsn );
}

sub usage {
   my ( $self ) = @_;
   my $usage
      = "DSN syntax is key=value[,key=value...]  Allowable DSN keys:\n"
      . "  KEY  COPY  MEANING\n"
      . "  ===  ====  =============================================\n";
   my %opts = %{$self->{opts}};
   foreach my $key ( sort keys %opts ) {
      $usage .= "  $key    "
             .  ($opts{$key}->{copy} ? 'yes   ' : 'no    ')
             .  ($opts{$key}->{desc} || '[No description]')
             . "\n";
   }
   if ( (my $key = $self->prop('autokey')) ) {
      $usage .= "  If the DSN is a bareword, the word is treated as the '$key' key.\n";
   }
   return $usage;
}

sub get_cxn_params {
   my ( $self, $info ) = @_;
   my $dsn;
   my %opts = %{$self->{opts}};
   my $driver = $self->prop('dbidriver') || '';
   if ( $driver eq 'Pg' ) {
      $dsn = 'DBI:Pg:dbname=' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(h P));
   }
   else {
      $dsn = 'DBI:mysql:' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(F h P S A))
         . ';mysql_read_default_group=mysql';
   }
   $ENV{MKDEBUG} && _d($dsn);
   return ($dsn, $info->{u}, $info->{p});
}

sub get_dbh {
   my ( $self, $cxn_string, $user, $pass, $opts ) = @_;
   $opts ||= {};
   my $defaults = {
      AutoCommit        => 0,
      RaiseError        => 1,
      PrintError        => 0,
      mysql_enable_utf8 => ($cxn_string =~ m/charset=utf8/ ? 1 : 0),
   };
   @{$defaults}{ keys %$opts } = values %$opts;
   $ENV{MKDEBUG} && _d($cxn_string, ' ', $user, ' ', $pass, ' {',
      join(', ', map { "$_=>$defaults->{$_}" } keys %$defaults ), '}');
   my $dbh = DBI->connect($cxn_string, $user, $pass, $defaults);
   if ( my ($charset) = $cxn_string =~ m/charset=(\w+)/ ) {
      my $sql = "/*!40101 SET NAMES $charset*/";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
      $ENV{MKDEBUG} && _d('Enabling charset for STDOUT');
      if ( $charset eq 'utf8' ) {
         binmode(STDOUT, ':utf8')
            or die "Can't binmode(STDOUT, ':utf8'): $OS_ERROR";
      }
      else {
         binmode(STDOUT) or die "Can't binmode(STDOUT): $OS_ERROR";
      }
   }
   my $setvars = $self->prop('setvars');
   if ( $setvars ) {
      my $sql = "SET $setvars";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
   }
   $ENV{MKDEBUG} && _d('DBH info: ',
      $dbh,
      Dumper($dbh->selectrow_hashref(
         'SELECT DATABASE(), CONNECTION_ID(), VERSION()/*!50038 , @@hostname*/')),
      ' Connection info: ', ($dbh->{mysql_hostinfo} || 'undef'),
      ' Character set info: ',
      Dumper($dbh->selectall_arrayref(
         'SHOW VARIABLES LIKE "character_set%"', { Slice => {}})),
      ' $DBD::mysql::VERSION: ', $DBD::mysql::VERSION,
      ' $DBI::VERSION: ', $DBI::VERSION,
   );
   return $dbh;
}

sub get_hostname {
   my ( $self, $dbh ) = @_;
   if ( my ($host) = ($dbh->{mysql_hostinfo} || '') =~ m/^(\w+) via/ ) {
      return $host;
   }
   my ( $hostname, $one ) = $dbh->selectrow_array(
      'SELECT /*!50038 @@hostname, */ 1');
   return $hostname;
}

sub disconnect {
   my ( $self, $dbh ) = @_;
   $ENV{MKDEBUG} && $self->print_active_handles($dbh);
   $dbh->disconnect;
}

sub print_active_handles {
   my ( $self, $thing, $level ) = @_;
   $level ||= 0;
   printf("# Active %sh: %s %s %s\n", $thing->{Type}, "\t" x $level,
      $thing, ($thing->{Type} eq 'st' ? $thing->{Statement} || '' : ''));
   foreach my $handle ( grep {defined} @{ $thing->{ChildHandles} } ) {
      $self->print_active_handles->( $handle, $level + 1 );
   }
}

sub _d {
   my ( $line ) = (caller(0))[2];
   @_ = map { defined $_ ? $_ : 'undef' } @_;
   print "# DSNParser:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End DSNParser package
# ###########################################################################

package main;

use English qw(-no_match_vars);
use List::Util qw(max);
use sigtrap qw(handler finish untrapped normal-signals);

my @opt_spec = (
   { s => 'askpass',       d => 'Prompt for password for connections' },
   { s => 'continue|c!',   d => 'Continue replication normally on exit (default)' },
   { s => 'daemonize',     d => 'Fork to background and detach (POSIX only)' },
   { s => 'delay|d=m',     d => 'Slave delay (default 1h)' },
   { s => 'interval|i=m',  d => 'Sleep interval (default 1m)' },
   { s => 'quiet|q',       d => 'Suppress normal output' },
   { s => 'setvars=s',     d => 'Set these MySQL variables (default wait_timeout=10000)' },
   { s => 'time|t=m',      d => 'Time to run before exiting' },
   { s => 'usemaster|u',   d => 'Get binlog positions from master, not slave' },
);

my $dsn_parser = new DSNParser();
$dsn_parser->prop('required', { h => 1 });
$dsn_parser->prop('autokey', 'h' );
my $vp = new VersionParser();
my $opt_parser = new OptionParser(@opt_spec);
$opt_parser->{dsn} = $dsn_parser;
$opt_parser->{strict} = 0;
$opt_parser->{prompt} = '[OPTION]... SLAVE-HOST [MASTER-HOST]';
$opt_parser->{descr}  = 'starts and stops a slave server as needed '
                      . 'to make it lag behind the master.  The SLAVE-HOST '
                      . 'and MASTER-HOST use DSN syntax, and values are copied '
                      . 'from the SLAVE-HOST to the MASTER-HOST if omitted.';
my %opts = $opt_parser->parse();
$dsn_parser->prop('setvars', $opts{setvars});

my ($slave_dsn, $master_dsn);
if ( @ARGV ) {
   $slave_dsn  = $dsn_parser->parse(shift @ARGV);
   $master_dsn = $dsn_parser->parse(shift(@ARGV), $slave_dsn) if @ARGV;
}

if ( !$opts{help} ) {
   if ( !$slave_dsn ) {
      $opt_parser->error("Missing or invalid slave host");
   }
}

$opts{i} = max($opts{i}, 1);
if ( $opts{t} ) {
   $opts{t} = max($opts{t}, 1);
}

$opt_parser->usage_or_errors(%opts);

# ############################################################################
# Ready to work now.
# ############################################################################
my ( $TS, $FILE, $POS ) = ( 0, 1, 2 );

my @positions;
my $now        = time();
my $next_start = 0;
my $end        = $now + ( $opts{t} || 0 );    # When we should exit
my $oktorun    = 1;

# Connect before daemonizing, in case --askpass is needed.
my ( $slave, $master );
$slave     = get_dbh($slave_dsn);
my $status = $slave->selectrow_hashref("SHOW SLAVE STATUS");
if ( !$status || ! %$status ) {
   die "No SLAVE STATUS found.\n";
}

if ( $master_dsn ) {
   $ENV{MKDEBUG} && _d('Connecting to master via DSN from cmd-line');
   $master = get_dbh($master_dsn);
}
elsif ( $opts{u} || $status->{slave_io_state} =~ m/free enough relay log/ ) {
   # Try to connect to the slave's master just by looking at its SLAVE STATUS.
   $ENV{MKDEBUG} && _d('The I/O thread is waiting, connecting to master');
   my $spec = "h=$status->{master_host},P=$status->{master_port}";
   $master  = get_dbh($dsn_parser->parse($spec, $slave_dsn));
}

# Daemonize only after (potentially) asking for passwords for --askpass.
if ( $opts{daemonize} ) {
   require POSIX;
   chdir '/'                 or die "Can't chdir to /: $OS_ERROR";
   open STDIN, '/dev/null'   or die "Can't read /dev/null: $OS_ERROR";
   open STDOUT, '>/dev/null' or die "Can't write to /dev/null: $OS_ERROR";
   defined( my $pid = fork ) or die "Can't fork: $OS_ERROR";
   exit if $pid;
   POSIX::setsid()           or die "Can't start a new session: $OS_ERROR";
   open STDERR, '>&STDOUT'   or die "Can't dup STDOUT: $OS_ERROR";
}

while (                       # Quit if:
   (!$opts{t} || $now < $end) # time is exceeded
   && $oktorun                # or instructed to quit
) {

   $now = time();

   $status = $slave->selectrow_hashref("SHOW SLAVE STATUS");
   if ( !$status || ! %$status ) {
      die "No SLAVE STATUS found.\n";
   }

   if ( !$master && $status->{slave_io_state} =~ m/free enough relay log/ ) {
      $ENV{MKDEBUG} && _d("The I/O thread is stuck, connecting to master");
      # If we're daemonized and --askpass is given, there's no way to ask for
      # a password.
      if ( $opts{daemonize} && $opts{askpass} ) {
         die "Cannot ask for password while daemonized";
      }
      my $spec = "h=$status->{master_host},P=$status->{master_port}";
      $master  = get_dbh($dsn_parser->parse($spec, $slave_dsn));
   }

   if ( defined $status->{seconds_behind_master} ) {
      info("slave running $status->{seconds_behind_master} seconds behind");
   }

   # Get binlog position.
   if ( $master ) {
      $ENV{MKDEBUG} && _d('Getting binlog pos from master');
      my $res = $master->selectrow_hashref("SHOW MASTER STATUS");
      my $pos = $positions[-1];
      if ( !@positions || $pos->[$FILE] ne $res->{file}
         || $pos->[$POS] != $res->{position} )
      {
         push @positions,
            [ $now, $res->{file}, $res->{position} ];
      }
   }
   else {
      $ENV{MKDEBUG} && _d('Getting binlog pos from slave');
      # Use the position on master at which the I/O thread is reading.  If the
      # I/O thread is not far behind, which it usually is not, this is basically
      # the same as the master's File/Position, but it's more efficient -- one
      # fewer connections to keep open.
      my $pos = $positions[-1];
      if ( !@positions
         || $pos->[$FILE] ne $status->{master_log_file}
         || $pos->[$POS] != $status->{read_master_log_pos} )
      {
         push @positions,
            [ $now, $status->{master_log_file},
              $status->{read_master_log_pos} ];
      }
   }

   if ( ( $status->{slave_sql_running} || '' ) eq 'No' ) {
      $ENV{MKDEBUG} && _d('Slave not running');
      # Find the most recent binlog position that's older than the delay amount.
      my $pos;
      my $i = 0;
      while ( $i < @positions && $positions[$i]->[$TS] <= $now - $opts{d} ) {
         $pos = $i;
         $i++;
      }

      if ( $pos ) {
         my $position = $positions[$pos];
         $ENV{MKDEBUG} && _d('Chosen position: ', ts($position->[$TS]),
               " $position->[$FILE]/$position->[$POS]");
      }
      else {
         $ENV{MKDEBUG} && _d('No position found');
      }

      # Make the slave server delay if possible; otherwise sleep and check
      # again.
      if ( $now >= $next_start && defined $pos ) {
         my $position = $positions[$pos];
         if ( $position->[$FILE] ne $status->{master_log_file}
            || $position->[$POS] != $status->{read_master_log_pos} )
         {
            $slave->do(
               "START SLAVE SQL_THREAD UNTIL /*$position->[$TS]*/ "
                  . "MASTER_LOG_FILE = '$position->[$FILE]', "
                  . "MASTER_LOG_POS = $position->[$POS]"
            );

            info("START SLAVE until master "
               . ts($position->[$TS])
               . " $position->[$FILE]/$position->[$POS]");
         }
         else {
            info("no new binlog events");
         }

         # Throw away positions we're going to replicate past.
         @positions = @positions[$pos + 1 .. $#positions];
      }
      else {
         my $position = $positions[-1];
         info("slave stopped at master position "
            . "$position->[$FILE]/$position->[$POS]");
      }
   }
   elsif ( ($status->{seconds_behind_master} || 0) < $opts{d} ) {
      my $position = $positions[-1];
      my $behind = $status->{seconds_behind_master} || 0;
      $next_start = $now + $opts{d} - $behind;
      info("STOP SLAVE until "
         . ts($next_start)
         . " at master position $position->[$FILE]/$position->[$POS]");
      $slave->do("STOP SLAVE SQL_THREAD");
   }
   else {
      my $position = $positions[-1];
      my $behind = $status->{seconds_behind_master} || 0;
      info("slave running $behind seconds behind at"
         . " master position $position->[$FILE]/$position->[$POS]");
   }

   sleep($opts{i});
}

if ( $slave && $opts{c} ) {
   info("Setting slave to run normally");
   $slave->do("START SLAVE SQL_THREAD");
}

# ############################################################################
# Subroutines
# ############################################################################

sub info {
   my ( $message ) = @_;
   print ts($now), " ", $message, "\n" unless $opts{q};
}

# Catches signals so mk-slave-delay can exit gracefully.
sub finish {
   my ($signal) = @_;
   print STDERR "Exiting on SIG$signal.\n";
   $oktorun = 0;
}

sub ts {
   my ( $time ) = @_;
   my ( $sec, $min, $hour, $mday, $mon, $year )
      = localtime($time);
   $mon  += 1;
   $year += 1900;
   return sprintf("%d-%02d-%02dT%02d:%02d:%02d",
      $year, $mon, $mday, $hour, $min, $sec);
}

sub get_dbh {
   my ( $info, $db ) = @_;

   if ( $opts{askpass} ) {
      $info->{p} = OptionParser::prompt_noecho("Enter password for $info->{h}: ");
   }

   my $dbh = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($info), {AutoCommit => 1});
   $dbh->{FetchHashKeyName} = 'NAME_lc'; # Lowercases all column names
   $dbh->{InactiveDestroy}  = 1;         # Don't disconnect on fork
   return $dbh;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# main:$line ", @_, "\n";
}

# ############################################################################
# Documentation.
# ############################################################################

=pod

=head1 NAME

mk-slave-delay - Make a MySQL slave server lag behind its master.

=head1 SYNOPSIS

To hold slavehost one minute behind its master for ten minutes:

 mk-slave-delay --delay 1m --interval 15s --time 10m slavehost

=head1 DESCRIPTION

C<mk-slave-delay> watches a slave and starts and stops its replication SQL
thread as necessary to hold it at least as far behind the master as you
request.  In practice, it will typically cause the slave to lag between
L<"--delay"> and L<"--delay">+L<"--interval"> behind the master.

It bases the delay on binlog positions in the slave's relay logs by default,
so there is no need to connect to the master.  This works well if the IO
thread doesn't lag the master much, which is typical in most replication
setups; the IO thread lag is usually milliseconds on a fast network.  If your
IO thread's lag is too large for your purposes, C<mk-slave-delay> can also
connect to the master for information about binlog positions.

If the slave's I/O thread reports that it is waiting for the SQL thread to
free some relay log space, C<mk-slave-delay> will automatically connect to the
master to find binary log positions.  If L<"--askpass"> and L<"--daemonize">
are given, it is possible that this could cause it to ask for a password while
daemonized.  In this case, it exits.  Therefore, if you think your slave might
encounter this condition, you should be sure to either specify
L<"--usemaster"> explicitly when daemonizing, or don't specify L<"--askpass">.

Note that since C<mk-slave-delay> starts and stops the SQL thread, monitoring
systems may think the slave is having trouble when it's just being held back
intentionally.

There is a special syntax for connecting to MySQL servers.  Each server name
on the command line can be either just a hostname, or a key=value,key=value
string.  Keys are a single letter:

   KEY MEANING
   === =======
   h   Connect to host
   P   Port number to use for connection
   S   Socket file to use for connection
   u   User for login if not current user
   p   Password to use when connecting
   F   Only read default options from the given file

If you omit any values in MASTER-HOST, they are filled in with defaults from
SLAVE-HOST, so you don't need to specify them in both places.  C<mk-slave-delay>
reads all normal MySQL option files, such as ~/.my.cnf, so you may not need to
specify username, password and other common options at all.

C<mk-slave-delay> tries to exit gracefully by trapping signals such as Ctrl-C.
You cannot bypass L<"--continue"> with a trappable signal.

=head1 OPTIONS

Some options are negatable by specifying them in their long form with a --no
prefix.

=over

=item --askpass

Prompts the user for a password when connecting to MySQL.

=item --continue

After exiting, restart the slave's SQL thread with no UNTIL condition, so it
will run as usual and catch up to the master.  This is enabled by default and
works even if you terminate C<mk-slave-delay> with Control-C.

=item --daemonize

Fork to the background and detach from the shell.  This probably doesn't work
on Microsoft Windows.

=item --delay

How far the slave should lag its master.

=item --help

Displays a help message.

=item --interval

How frequently C<mk-slave-delay> should check whether the slave needs to be
started or stopped.

=item --quiet

Do not output regular status messages.

=item --setvars

Specify any variables you want to be set immediately after connecting to MySQL.
These will be included in a C<SET> command.

=item --time

How long C<mk-slave-delay> should run before exiting.  Default is to run
forever.

=item --usemaster

Don't trust the binlog positions in the slave's relay log.  Connect to the
master and get binlog positions instead.  If you specify this option without
giving a MASTER-HOST on the command line, C<mk-slave-delay> examines the
slave's SHOW SLAVE STATUS to determine the hostname and port for connecting to
the master.

C<mk-slave-delay> uses only the MASTER_HOST and MASTER_PORT values from SHOW
SLAVE STATUS for the master connection.  It does not use the MASTER_USER
value.  If you want to specify a different username for the master than the
one you use to connect to the slave, you should specify the MASTER-HOST option
explicitly on the command line.

=item --version

Output version information and exit.

=back

=head1 SYSTEM REQUIREMENTS

You need Perl, DBI, DBD::mysql, and some core packages that ought to be
installed in any reasonably new version of Perl.

=head1 OUTPUT

If you specify L<"--quiet">, there is no output.  Otherwise, the normal output
is a status message consisting of a timestamp and information about what
C<mk-slave-delay> is doing: starting the slave, stopping the slave, or just
observing.

=head1 ENVIRONMENT

The environment variable C<MKDEBUG> enables verbose debugging output in all of
the Maatkit tools:

   MKDEBUG=1 mk-....

=head1 BUGS

Please use the Sourceforge bug tracker, forums, and mailing lists to request
support or report bugs: L<http://sourceforge.net/projects/maatkit/>.

=head1 COPYRIGHT, LICENSE AND WARRANTY

This program is copyright (c) 2007 Sergey Zhuravlev and Baron Schwartz.
Feedback and improvements are welcome.

THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.

This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
systems, you can issue `man perlgpl' or `man perlartistic' to read these
licenses.

You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc., 59 Temple
Place, Suite 330, Boston, MA  02111-1307  USA.

=head1 AUTHOR

Sergey Zhuravlev and Baron Schwartz.

=head1 VERSION

This manual page documents Ver 1.0.6 Distrib 1877 $Revision: 1871 $.

=cut