percona/percona-toolkit
1
0
Fork 0
mirror of https://github.com/percona/percona-toolkit.git synced 2025-09-02 02:34:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Cxn.pm. Make MasterSlave, NibbleIterator, and ReplicaLagWaiter use Cxn. Rewrite, simplify Retry.

Browse Source
This commit is contained in:
Daniel Nichter
2011-10-06 12:47:35 -06:00
parent bd900c5ab8
commit e2e40488c5
9 changed files with 310 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
lib/Cxn.pm Normal file
View File

@@ -0,0 +1,121 @@
# This program is copyright 2011 Percona Inc.
# 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.
# ###########################################################################
# Cxn package
# ###########################################################################
{
# Package: Cxn
# Cxn creates a connection to MySQL and initializes it properly.
package Cxn;
use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant MKDEBUG => $ENV{MKDEBUG} || 0;
sub new {
my ( $class, %args ) = @_;
my @required_args = qw(DSNParser OptionParser);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
};
die "I need a dsn or dsn_string argument"
unless $args{dsn} || $args{dsn_string};
my ($dp, $o) = @args{@required_args};
my $dsn = $args{dsn};
if ( !$dsn ) {
$dsn = $dp->parse(
$args{dsn_string}, $args{prev_dsn}, $dp->parse_options($o));
}
my $self = {
dsn_string => $args{dsn_string},
dsn => $dsn,
dbh => $args{dbh},
OptionParser => $o,
DSNParser => $dp,
};
MKDEBUG && _d('New connection to', $dsn->{n});
return bless $self, $class;
}
sub connect {
my ( $self ) = @_;
my $dsn = $self->{dsn};
my $dp = $self->{DSNParser};
my $o = $self->{OptionParser};
my $dbh = $self->{dbh};
if ( !$dbh ) {
if ( $o->get('ask-pass') ) {
$dsn->{p} = OptionParser::prompt_noecho("Enter password: ");
}
$dbh = $dp->get_dbh($dp->get_cxn_params($dsn), { AutoCommit => 1 });
MKDEBUG && _d('Connected dbh', $dbh, $dsn->{n});
}
return $self->set_dbh($dbh);
}
sub set_dbh {
my ($self, $dbh) = @_;
# Don't set stuff twice on the same dbh.
return $dbh if $self->{dbh} && $self->{dbh} == $dbh;
# Set stuff for this dbh (i.e. initialize it).
$dbh->{FetchHashKeyName} = 'NAME_lc';
$self->{dbh} = $dbh;
return $dbh;
}
sub dbh {
my ($self) = @_;
return $self->{dbh};
}
sub dsn {
my ($self) = @_;
return $self->{dsn};
}
sub DESTROY {
my ($self) = @_;
if ( $self->{dbh} ) {
MKDEBUG && _d('Disconnecting dbh', $self->{dbh}, $self->{dsn}->{n});
$self->{dbh}->disconnect();
}
return;
}
sub _d {
my ($package, undef, $line) = caller 0;
@_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
map { defined $_ ? $_ : 'undef' }
@_;
print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}
1;
}
# ###########################################################################
# End Cxn package
# ###########################################################################

32
lib/MasterSlave.pm
View File

@@ -38,11 +38,11 @@ sub new {
sub get_slaves {
my ($self, %args) = @_;
my @required_args = qw(OptionParser DSNParser Quoter);
my @required_args = qw(make_cxn OptionParser DSNParser Quoter);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
}
my ($o, $dp) = @args{@required_args};
my ($make_cxn, $o, $dp) = @args{@required_args};
my $slaves = [];
my $method = $o->get('recursion-method');
@@ -63,9 +63,7 @@ sub get_slaves {
my ( $dsn, $dbh, $level, $parent ) = @_;
return unless $level;
MKDEBUG && _d('Found slave:', $dp->as_string($dsn));
$dbh->{InactiveDestroy} = 1; # Prevent destroying on fork.
$dbh->{FetchHashKeyName} = 'NAME_lc';
push @$slaves, { dsn=>$dsn, dbh=>$dbh };
push @$slaves, $make_cxn->(dsn => $dsn, dbh => $dbh);
return;
},
}
@@ -849,11 +847,11 @@ sub reset_known_replication_threads {
sub get_cxn_from_dsn_table {
my ($self, %args) = @_;
my @required_args = qw(dsn_table_dsn DSNParser Quoter);
my @required_args = qw(dsn_table_dsn make_cxn DSNParser Quoter);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
}
my ($dsn_table_dsn, $dp, $q) = @args{@required_args};
my ($dsn_table_dsn, $make_cxn, $dp, $q) = @args{@required_args};
MKDEBUG && _d('DSN table DSN:', $dsn_table_dsn);
my $dsn = $dp->parse($dsn_table_dsn);
@@ -869,20 +867,18 @@ sub get_cxn_from_dsn_table {
. "or a database-qualified table (t)";
}
my @cxn;
my $dbh = $dp->get_dbh($dp->get_cxn_params($dsn));
my $sql = "SELECT dsn FROM $dsn_table ORDER BY id";
my $dsn_tbl_cxn = $make_cxn->(dsn => $dsn);
my $dbh = $dsn_tbl_cxn->connect();
my $sql = "SELECT dsn FROM $dsn_table ORDER BY id";
MKDEBUG && _d($sql);
my $dsns = $dbh->selectcol_arrayref($sql);
if ( $dsns ) {
foreach my $dsn ( @$dsns ) {
MKDEBUG && _d('DSN from DSN table:', $dsn);
my $dsn = $dp->parse($dsn);
my $dbh = $dp->get_dbh($dp->get_cxn_params($dsn));
push @cxn, {dsn=>$dsn, dbh=>$dbh};
my $dsn_strings = $dbh->selectcol_arrayref($sql);
my @cxn;
if ( $dsn_strings ) {
foreach my $dsn_string ( @$dsn_strings ) {
MKDEBUG && _d('DSN from DSN table:', $dsn_string);
push @cxn, $make_cxn->(dsn_string => $dsn_string);
}
}
$dbh->disconnect();
return \@cxn;
}

35
lib/NibbleIterator.pm
View File

@@ -35,7 +35,7 @@ $Data::Dumper::Quotekeys = 0;
# Sub: new
#
# Required Arguments:
# dbh - dbh
# cxn - <Cxn> object
# tbl - Standard tbl ref
# chunk_size - Number of rows to nibble per chunk
# OptionParser - <OptionParser> object
@@ -51,18 +51,18 @@ $Data::Dumper::Quotekeys = 0;
# NibbleIterator object
sub new {
my ( $class, %args ) = @_;
my @required_args = qw(dbh tbl chunk_size OptionParser Quoter TableNibbler TableParser);
my @required_args = qw(Cxn tbl chunk_size OptionParser Quoter TableNibbler TableParser);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
}
my ($dbh, $tbl, $chunk_size, $o, $q) = @args{@required_args};
my ($cxn, $tbl, $chunk_size, $o, $q) = @args{@required_args};
my $one_nibble = !defined $args{one_nibble} || $args{one_nibble}
? _can_nibble_once(%args)
? _can_nibble_once(dbh => $cxn->dbh(), %args)
: 0;
# Get an index to nibble by. We'll order rows by the index's columns.
my $index = _find_best_index(%args);
my $index = _find_best_index(dbh => $cxn->dbh(), %args);
if ( !$index && !$one_nibble ) {
die "There is no good index and the table is oversized.";
}
@@ -221,7 +221,7 @@ sub next {
my ($self) = @_;
my %callback_args = (
dbh => $self->{dbh},
Cxn => $self->{Cxn},
tbl => $self->{tbl},
NibbleIterator => $self,
);
@@ -456,14 +456,17 @@ sub _can_nibble_once {
sub _prepare_sths {
my ($self) = @_;
MKDEBUG && _d('Preparing statement handles');
$self->{nibble_sth}
= $self->{dbh}->prepare($self->{nibble_sql});
$self->{explain_nibble_sth}
= $self->{dbh}->prepare($self->{explain_nibble_sql});
my $dbh = $self->{Cxn}->dbh();
$self->{nibble_sth} = $dbh->prepare($self->{nibble_sql});
$self->{explain_nibble_sth} = $dbh->prepare($self->{explain_nibble_sql});
if ( !$self->{one_nibble} ) {
$self->{ub_sth} = $self->{dbh}->prepare($self->{ub_sql});
$self->{explain_ub_sth} = $self->{dbh}->prepare($self->{explain_ub_sql});
$self->{ub_sth} = $dbh->prepare($self->{ub_sql});
$self->{explain_ub_sth} = $dbh->prepare($self->{explain_ub_sql});
}
return;
}
@@ -471,10 +474,12 @@ sub _get_bounds {
my ($self) = @_;
return if $self->{one_nibble};
$self->{next_lower} = $self->{dbh}->selectrow_arrayref($self->{first_lb_sql});
my $dbh = $self->{Cxn}->dbh();
$self->{next_lower} = $dbh->selectrow_arrayref($self->{first_lb_sql});
MKDEBUG && _d('First lower boundary:', Dumper($self->{next_lower}));
$self->{last_upper} = $self->{dbh}->selectrow_arrayref($self->{last_ub_sql});
$self->{last_upper} = $dbh->selectrow_arrayref($self->{last_ub_sql});
MKDEBUG && _d('Last upper boundary:', Dumper($self->{last_upper}));
return;
@@ -521,7 +526,7 @@ sub _next_boundaries {
if ( my $callback = $self->{callbacks}->{next_boundaries} ) {
my $oktonibble = $callback->(
dbh => $self->{dbh},
Cxn => $self->{Cxn},
tbl => $self->{tbl},
NibbleIterator => $self,
);

4
lib/ReplicaLagWaiter.pm
View File

@@ -37,7 +37,7 @@ use Data::Dumper;
# get_lag - Callback passed slave dbh and returns slave's lag
# sleep - Callback to sleep between checking lag.
# max_lag - Max lag
# slaves - Arrayref of slave cxn, like [{dsn=>{...}, dbh=>...},...]
# slaves - Arrayref of <Cxn> objects
#
# Returns:
# ReplicaLagWaiter object
@@ -101,7 +101,7 @@ sub wait {
MKDEBUG && _d('Checking slave lag');
for my $i ( 0..$#lagged_slaves ) {
my $slave = $lagged_slaves[$i];
my $lag = $get_lag->($slave->{dbh});
my $lag = $get_lag->($slave->dbh());
MKDEBUG && _d($slave->{dsn}->{n}, 'slave lag:', $lag);
if ( !defined $lag || $lag > $max_lag ) {
$slave->{lag} = $lag;

72
lib/Retry.pm
View File

@@ -19,7 +19,7 @@
# ###########################################################################
{
# Package: Retry
# Retry retries code until a condition succeeds.
# Retry retries code that may die and handles those deaths nicely.
package Retry;
use strict;
@@ -35,64 +35,58 @@ sub new {
return bless $self, $class;
}
# Required arguments:
# * try coderef: code to try; return true on success
# * wait coderef: code that waits in between tries
# Sub: retry
# Retry a code block that may die several times.
#
# Required Arguments:
# try - Callback to try.
# fail - Callback when try dies.
# final_fail - Callback when try dies for the last time.
#
# Optional arguments:
# * tries scalar: number of retries to attempt (default 3)
# * retry_on_die bool: retry try code if it dies (default no)
# * on_success coderef: code to call if try is successful
# * on_failure coderef: code to call if try does not succeed
# Retries the try code until either it returns true or we exhaust
# the number of retry attempts. The args are passed to the coderefs
# (try, wait, on_success, on_failure). If the try code dies, that's
# a final failure (no more retries) unless retry_on_die is true.
# Returns either whatever the try code returned or undef on failure.
# tries - Number of try attempts (default 3)
# wait - Callback after fail if tries remain (default sleep 1s)
#
# Returns:
# Return value of try code when it doesn't die.
sub retry {
my ( $self, %args ) = @_;
my @required_args = qw(try wait);
my @required_args = qw(try fail final_fail);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
};
my ($try, $wait) = @args{@required_args};
my ($try, $fail, $final_fail) = @args{@required_args};
my $wait = $args{wait} || sub { sleep 1; };
my $tries = $args{tries} || 3;
my $last_error;
my $tryno = 0;
TRY:
while ( ++$tryno <= $tries ) {
MKDEBUG && _d("Retry", $tryno, "of", $tries);
MKDEBUG && _d("Try", $tryno, "of", $tries);
my $result;
eval {
$result = $try->(tryno=>$tryno);
};
if ( $EVAL_ERROR ) {
MKDEBUG && _d("Try code failed:", $EVAL_ERROR);
$last_error = $EVAL_ERROR;
if ( defined $result ) {
MKDEBUG && _d("Try code succeeded");
if ( my $on_success = $args{on_success} ) {
MKDEBUG && _d("Calling on_success code");
$on_success->(tryno=>$tryno, result=>$result);
if ( $tryno < $tries ) { # more retries
my $retry = $fail->(tryno=>$tryno, error=>$last_error);
last TRY unless $retry;
MKDEBUG && _d("Calling wait code");
$wait->(tryno=>$tryno);
}
}
else {
MKDEBUG && _d("Try code succeeded");
return $result;
}
if ( $EVAL_ERROR ) {
MKDEBUG && _d("Try code died:", $EVAL_ERROR);
die $EVAL_ERROR unless $args{retry_on_die};
}
# Wait if there's more retries, else end immediately.
if ( $tryno < $tries ) {
MKDEBUG && _d("Try code failed, calling wait code");
$wait->(tryno=>$tryno);
}
}
MKDEBUG && _d("Try code did not succeed");
if ( my $on_failure = $args{on_failure} ) {
MKDEBUG && _d("Calling on_failure code");
$on_failure->();
}
return;
MKDEBUG && _d('Try code did not succeed');
return $final_fail->(error=>$last_error);
}
sub _d {

34
t/lib/MasterSlave.t
View File

@@ -16,6 +16,7 @@ use DSNParser;
use VersionParser;
use OptionParser;
use Quoter;
use Cxn;
use Sandbox;
use PerconaTest;
@@ -50,10 +51,19 @@ SKIP: {
OptionParser => $o,
DSNParser => $dp,
Quoter => $q,
make_cxn => sub {
my $cxn = new Cxn(
@_,
DSNParser => $dp,
OptionParser => $o,
);
$cxn->connect();
return $cxn;
},
);
is_deeply(
$slaves->[0]->{dsn},
$slaves->[0]->dsn(),
{ A => undef,
D => undef,
F => undef,
@@ -63,6 +73,7 @@ SKIP: {
p => 'msandbox',
t => undef,
u => 'msandbox',
n => 'h=127.0.0.1,P=12346',
server_id => 12346,
master_id => 12345,
source => 'hosts',
@@ -70,14 +81,12 @@ SKIP: {
'get_slaves() from recurse_to_slaves()'
);
my ($id) = $slaves->[0]->{dbh}->selectrow_array('SELECT @@SERVER_ID');
my ($id) = $slaves->[0]->dbh()->selectrow_array('SELECT @@SERVER_ID');
is(
$id,
'12346',
'dbh created from get_slaves()'
);
$slaves->[0]->{dbh}->disconnect();
}
# #############################################################################
@@ -170,6 +179,7 @@ is_deeply(
D => undef,
A => undef,
t => undef,
n => 'h=127.0.0.1,P=2900',
},
'Got master DSN',
);
@@ -531,6 +541,15 @@ my $slaves = $ms->get_slaves(
OptionParser => $o,
DSNParser => $dp,
Quoter => $q,
make_cxn => sub {
my $cxn = new Cxn(
@_,
DSNParser => $dp,
OptionParser => $o,
);
$cxn->connect();
return $cxn;
},
);
is_deeply(
@@ -543,20 +562,19 @@ is_deeply(
h => '127.1',
p => 'msandbox',
t => undef,
u => 'msandbox'
u => 'msandbox',
n => 'h=127.1,P=12346',
},
'get_slaves() from DSN table'
);
my ($id) = $slaves->[0]->{dbh}->selectrow_array('SELECT @@SERVER_ID');
my ($id) = $slaves->[0]->dbh()->selectrow_array('SELECT @@SERVER_ID');
is(
$id,
'12346',
'dbh created from DSN table works'
);
$slaves->[0]->{dbh}->disconnect();
# #############################################################################
# Done.
# #############################################################################

19
t/lib/NibbleIterator.t
View File

@@ -21,6 +21,7 @@ use TableParser;
use TableNibbler;
use RowChecksum;
use NibbleIterator;
use Cxn;
use PerconaTest;
use constant MKDEBUG => $ENV{MKDEBUG} || 0;
@@ -41,11 +42,17 @@ else {
plan tests => 26;
}
my $q = new Quoter();
my $tp = new TableParser(Quoter=>$q);
my $nb = new TableNibbler(TableParser=>$tp, Quoter=>$q);
my $o = new OptionParser(description => 'NibbleIterator');
my $rc = new RowChecksum(OptionParser => $o, Quoter=>$q);
my $q = new Quoter();
my $tp = new TableParser(Quoter=>$q);
my $nb = new TableNibbler(TableParser=>$tp, Quoter=>$q);
my $o = new OptionParser(description => 'NibbleIterator');
my $rc = new RowChecksum(OptionParser => $o, Quoter=>$q);
my $cxn = new Cxn(
dbh => $dbh,
dsn => { h=>'127.1', P=>'12345', n=>'h=127.1,P=12345' },
DSNParser => $dp,
OptionParser => $o,
);
$o->get_specs("$trunk/bin/pt-table-checksum");
@@ -77,7 +84,7 @@ sub make_nibble_iter {
1 while $si->next();
my $ni = new NibbleIterator(
dbh => $dbh,
Cxn => $cxn,
tbl => $schema->get_table($args{db}, $args{tbl}),
chunk_size => $o->get('chunk-size'),
callbacks => $args{callbacks},

15
t/lib/ReplicaLagWaiter.t
View File

@@ -12,6 +12,7 @@ use English qw(-no_match_vars);
use Test::More tests => 5;
use ReplicaLagWaiter;
use Cxn;
use PerconaTest;
my $oktorun = 1;
@@ -36,13 +37,13 @@ sub sleep {
}
my $rll = new ReplicaLagWaiter(
oktorun => \&oktorun,
get_lag => \&get_lag,
sleep => \&sleep,
max_lag => 1,
slaves => [
{ dsn=>{n=>'slave1'}, dbh=>1 },
{ dsn=>{n=>'slave2'}, dbh=>2 },
oktorun => \&oktorun,
get_lag => \&get_lag,
sleep => \&sleep,
max_lag => 1,
slaves => [
new Cxn(dsn=>{n=>'slave1'}, dbh=>1, DSNParser=>1, OptionParser=>1),
new Cxn(dsn=>{n=>'slave2'}, dbh=>2, DSNParser=>1, OptionParser=>1),
],
);

181
t/lib/Retry.t
View File

@@ -9,138 +9,103 @@ BEGIN {
use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use Test::More tests => 14;
use Test::More tests => 6;
use Retry;
use PerconaTest;
my $success;
my $failure;
my $waitno;
my $tryno;
my $tries;
my $die;
my $rt = new Retry();
my @called = ();
my @retry = ();
my @die = ();
my $try = sub {
if ( $die ) {
$die = 0;
die "I die!\n";
}
return $tryno++ == $tries ? "succeed" : undef;
push @called, 'try';
die if shift @die;
};
my $fail = sub {
push @called, 'fail';
return shift @retry;
};
my $wait = sub {
$waitno++;
push @called, 'wait';
};
my $on_success = sub {
$success = "succeed on $tryno";
my $final_fail = sub {
push @called, 'final_fail';
return;
};
my $on_failure = sub {
$failure = "failed on $tryno";
};
sub try_it {
my ( %args ) = @_;
$success = "";
$failure = "";
$waitno = $args{wainot} || 0;
$tryno = $args{tryno} || 1;
$tries = $args{tries} || 3;
sub try_it {
return $rt->retry(
try => $try,
wait => $wait,
on_success => $on_success,
on_failure => $on_failure,
retry_on_die => $args{retry_on_die},
try => $try,
fail => $fail,
wait => $wait,
final_fail => $final_fail,
);
}
my $retval = try_it();
is(
$retval,
"succeed",
"Retry succeeded"
# Success on first try;
@called = ();
@retry = ();
@die = ();
try_it();
is_deeply(
\@called,
['try'],
'Success on first try'
);
is(
$success,
"succeed on 4",
"Called on_success code"
# Success on 2nd try.
@called = ();
@retry = (1),
@die = (1);
try_it();
is_deeply(
\@called,
['try', 'fail', 'wait',
'try'
],
'Success on second try'
);
is(
$waitno,
2,
"Called wait code"
# Success on 3rd, last try.
@called = ();
@retry = (1, 1),
@die = (1, 1);
try_it();
is_deeply(
\@called,
['try', 'fail', 'wait',
'try', 'fail', 'wait',
'try'
],
'Success on third, final try'
);
# Default tries is 3 so allowing ourself 4 tries will cause the retry
# to fail and the on_failure code should be called.
$retval = try_it(tries=>4);
ok(
!defined $retval,
"Returned undef on failure"
# Failure.
@called = ();
@retry = (1, 1, 1);
@die = (1, 1, 1);
try_it();
is_deeply(
\@called,
['try', 'fail', 'wait',
'try', 'fail', 'wait',
'try', 'final_fail',
],
'Failure'
);
is(
$failure,
"failed on 4",
"Called on_failure code"
);
is(
$success,
"",
"Did not call on_success code"
);
# Test what happens if the try code dies. try_it() will reset $die to 0.
$die = 1;
eval { try_it(); };
is(
$EVAL_ERROR,
"I die!\n",
"Dies if code dies without retry_on_die"
);
ok(
!defined $retval,
"Returned undef on try die"
);
is(
$failure,
"",
"Did not call on_failure code on try die without retry_on_die"
);
is(
$success,
"",
"Did not call on_success code"
);
# Test retry_on_die. This should work with tries=2 because the first
# try will die leaving with only 2 more retries.
$die = 1;
$retval = try_it(retry_on_die=>1, tries=>2);
is(
$retval,
"succeed",
"Retry succeeded with retry_on_die"
);
is(
$success,
"succeed on 3",
"Called on_success code with retry_on_die"
);
is(
$waitno,
2,
"Called wait code with retry_on_die"
# Fail and no retry.
@called = ();
@retry = (0);
@die = (1);
try_it();
is_deeply(
\@called,
['try', 'fail', 'final_fail'],
"Fail, don't retry"
);
# #############################################################################