PT-175 slave-restart support replication channels

- Added support for MySQL 5.7+ replication channels to slave-restart
- Modified lib/MasterSlave.pm to improve support replication channels

bin/pt-slave-restart

@@ -2984,26 +2984,30 @@ sub get_slave_status {
       my ($sss_rows) = $sth->fetchall_arrayref({}); # Show Slave Status rows
 
       my $ss;
+      my @valid_channel_names;
       if ( $sss_rows && @$sss_rows ) {
-          if (scalar @$sss_rows > 1) {
-              if (!$self->{channel}) {
-                  warn 'This server returned more than one row for SHOW SLAVE STATUS but "channel" was not specified on the command line';
-                  return undef;
+          my $channel_name = $self->{channel} || '';
+          for my $row (@$sss_rows) {
+              $row = { map { lc($_) => $row->{$_} } keys %$row }; # lowercase the keys
+              push @valid_channel_names, $row->{channel_name};
+              if ($row->{channel_name} eq $channel_name) {
+                  $ss = $row;
+                  last;
               }
-              for my $row (@$sss_rows) {
-                  $row = { map { lc($_) => $row->{$_} } keys %$row }; # lowercase the keys
-                  if ($row->{channel_name} eq $self->{channel}) {
-                      $ss = $row;
-                      last;
-                  }
-              }
-          } else {
-              $ss = $sss_rows->[0];
           }
 
           if ( $ss && %$ss ) {
              $ss = { map { lc($_) => $ss->{$_} } keys %$ss }; # lowercase the keys
              return $ss;
+          } else {
+             my $msg;
+             if (!$self->{channel}) {
+                 $msg = "cannot get slave status without using a channel name.\n";
+             } else {
+                 $msg = "cannot get slave status using channel name '$self->{channel}'.\n";
+             }
+             $msg .= "SHOW SLAVE STATUS returned these channel names: ".join(", ", @valid_channel_names) if @valid_channel_names;
+             die $msg;
           }
       }
 
@@ -4977,7 +4981,9 @@ sub main {
       OptionParser => $o,
       DSNParser    => $dp,
       Quoter       => $q,
+      channel      => $o->get('channel'),
    );
+
    $ms->recurse_to_slaves(
       {  dbh      => $dbh,
          dsn      => $dsn,
@@ -4988,15 +4994,17 @@ sub main {
                my $stat = $ms->get_slave_status($dbh);
                if ( $stat ) {
                   push @servers_to_watch, { dsn => $dsn, dbh => $dbh };
-               }
-               else {
-                  die "could not find slave status on this server\n";
+               } else {
+                  die "could not find slave status on this server.\n";
                }
             };
             if ( $EVAL_ERROR ) {
                chomp $EVAL_ERROR;
-               PTDEBUG && _d('Not watching', $dp->as_string($dsn),
-                  'because', $EVAL_ERROR);
+               if ($level == 0) {
+                   print 'Not watching '. $dp->as_string($dsn). ' because ' . $EVAL_ERROR;
+               } else {
+                   PTDEBUG && _d('Not watching', $dp->as_string($dsn), 'because', $EVAL_ERROR);
+               }
             }
          },
          skip_callback => sub {
@@ -5059,6 +5067,9 @@ sub main {
 sub watch_server {
    my ( $dsn, $dbh, $was_forked, $ms ) = @_;
 
+   my $server_version = VersionParser->new($dbh);
+   my $channel_sql = $server_version > '5.6' && $o->get('channel') ? " FOR CHANNEL '".$o->get('channel')."' " : '';
+
    PTDEBUG && _d('Watching server', $dp->as_string($dsn),
       'forked:', $was_forked);
 
@@ -5073,8 +5084,9 @@ sub watch_server {
       $start_sql .= " UNTIL RELAY_LOG_FILE = '$file', RELAY_LOG_POS = $pos";
    }
 
+   $start_sql .= $channel_sql;
    my $start = $dbh->prepare($start_sql);
-   my $stop  = $dbh->prepare('STOP SLAVE');
+   my $stop  = $dbh->prepare('STOP SLAVE'.$channel_sql);
 
    # ########################################################################
    # Detect if GTID is enabled. Skipping an event is done differently.
@@ -5131,7 +5143,8 @@ sub watch_server {
          # https://bugs.launchpad.net/percona-toolkit/+bug/932614
          my $sql = "CHANGE MASTER TO "
                  . "MASTER_LOG_FILE='$stat->{relay_master_log_file}', "
-                 . "MASTER_LOG_POS=$stat->{exec_master_log_pos}";
+                 . "MASTER_LOG_POS=$stat->{exec_master_log_pos}"
+                 . $channel_sql;
          PTDEBUG && _d($sql);
          $dbh->do($sql);
       },
@@ -5532,6 +5545,18 @@ pt-slave-restart will not let you stop the slave manually if you want to!
 
 Prompt for a password when connecting to MySQL.
 
+=item --channel
+
+type: string
+
+Channel name used when connected to a server using replication channels.
+Suppose you have two masters, master_a at port 12345, master_b at port 1236 and
+a slave connected to both masters using channels chan_master_a and chan_master_b.
+If you want to run pt-table-sync to syncronize the slave against master_a, pt-table-sync
+won't be able to determine what's the correct master since SHOW SLAVE STATUS
+will return 2 rows. In this case, you can use --channel=chan_master_a to specify
+the channel name to use in the SHOW SLAVE STATUS command.
+
 =item --charset
 
 short form: -A; type: string

lib/MasterSlave.pm

@@ -443,31 +443,35 @@ sub get_slave_status {
       # If we don't have a channel name as a parameter, there is no way to know what the correct master is so,
       # return an error.
       my $ss;
+      my @valid_channel_names;
       if ( $sss_rows && @$sss_rows ) {
-          if (scalar @$sss_rows > 1) {
-              if (!$self->{channel}) {
-                  warn 'This server returned more than one row for SHOW SLAVE STATUS but "channel" was not specified on the command line';
-                  return undef;
-              }
-              for my $row (@$sss_rows) {
-                  $row = { map { lc($_) => $row->{$_} } keys %$row }; # lowercase the keys
-                  if ($row->{channel_name} eq $self->{channel}) {
-                      $ss = $row;
-                      last;
-                  }
-              }
-          } else {
-              if ($sss_rows->[0]->{channel_name} && $sss_rows->[0]->{channel_name} ne $self->{channel}) {
-                  warn 'This server is using replication channels but "channel" was not specified on the command line';
-                  return undef;
-              } else {
-                  $ss = $sss_rows->[0];
+          my $channel_name = $self->{channel} || '';
+          for my $row (@$sss_rows) {
+              $row = { map { lc($_) => $row->{$_} } keys %$row }; # lowercase the keys
+              push @valid_channel_names, $row->{channel_name};
+              if ($row->{channel_name} eq $channel_name) {
+                  $ss = $row;
+                  last;
               }
           }
+          if (scalar @$sss_rows > 1 && !$self->{channel}) {
+             my $msg = "cannot get slave status without using a channel name.\n"
+                     . "SHOW SLAVE STATUS returned these channel names: ".join(", ", @valid_channel_names);
+             die $msg;
+          }
 
           if ( $ss && %$ss ) {
              $ss = { map { lc($_) => $ss->{$_} } keys %$ss }; # lowercase the keys
              return $ss;
+          } else {
+             my $msg;
+             if (!$self->{channel}) {
+                 $msg = "cannot get slave status without using a channel name.\n";
+             } else {
+                 $msg = "cannot get slave status using channel name '$self->{channel}'.\n";
+             }
+             $msg .= "SHOW SLAVE STATUS returned these channel names: ".join(", ", @valid_channel_names);
+             die $msg;
           }
       }
 
@@ -572,8 +576,11 @@ sub wait_for_master {
 # Executes STOP SLAVE.
 sub stop_slave {
    my ( $self, $dbh ) = @_;
+   my $server_version = VersionParser->new($dbh);
+   my $channel_sql = $server_version > '5.6' && $self->{channel} ? " FOR CHANNEL '".$self->{channel}."' " : '';
+
    my $sth = $self->{sths}->{$dbh}->{STOP_SLAVE}
-         ||= $dbh->prepare('STOP SLAVE');
+         ||= $dbh->prepare('STOP SLAVE'.$channel_sql);
    PTDEBUG && _d($dbh, $sth->{Statement});
    $sth->execute();
 }
@@ -581,16 +588,19 @@ sub stop_slave {
 # Executes START SLAVE, optionally with UNTIL.
 sub start_slave {
    my ( $self, $dbh, $pos ) = @_;
+   my $server_version = VersionParser->new($dbh);
+   my $channel_sql = $server_version > '5.6' && $self->{channel} ? " FOR CHANNEL '".$self->{channel}."' " : '';
    if ( $pos ) {
       # Just like with CHANGE MASTER TO, you can't quote the position.
       my $sql = "START SLAVE UNTIL MASTER_LOG_FILE='$pos->{file}', "
-              . "MASTER_LOG_POS=$pos->{position}";
+              . "MASTER_LOG_POS=$pos->{position}"
+              . $channel_sql;
       PTDEBUG && _d($dbh, $sql);
       $dbh->do($sql);
    }
    else {
       my $sth = $self->{sths}->{$dbh}->{START_SLAVE}
-            ||= $dbh->prepare('START SLAVE');
+            ||= $dbh->prepare('START SLAVE' . $channel_sql);
       PTDEBUG && _d($dbh, $sth->{Statement});
       $sth->execute();
    }

sandbox/slave_channels_ports.sql

@@ -0,0 +1,15 @@
+STOP SLAVE FOR CHANNEL '';
+SET GLOBAL master_info_repository = 'TABLE';
+SET @@GLOBAL.relay_log_info_repository = 'TABLE';
+SET @@GLOBAL.ENFORCE_GTID_CONSISTENCY=ON;
+SET @@GLOBAL.GTID_MODE = OFF_PERMISSIVE;
+SET @@GLOBAL.GTID_MODE = ON_PERMISSIVE;
+SET @@GLOBAL.GTID_MODE = ON;
+
+CHANGE MASTER TO master_host='127.0.0.1', master_port=2900, master_user='msandbox', master_password='msandbox', master_auto_position=1 FOR CHANNEL 'masterchan1';  
+
+CHANGE MASTER TO master_host='127.0.0.1', master_port=2901, master_user='msandbox', master_password='msandbox', master_auto_position=1 FOR CHANNEL 'masterchan2';  
+
+START SLAVE for channel 'masterchan1';
+START SLAVE for channel 'masterchan2';
+

sandbox/test-env

@@ -296,8 +296,11 @@ case $opt in
          if [ "${2:-""}" = "channels" ] && [ "$MYSQL_VERSION" '>' "5.6" ]; then
             ./start-sandbox master 12346 
             exit_status=$((exit_status | $?))
+            echo "Starting slave using port 12347. It will be connected to master 1 (port: 12345) and master 2 (port: 12346), using replication channels"
             ./start-sandbox master 12347 
             exit_status=$((exit_status | $?))
+            ./start-sandbox slave 12348 12345 
+            exit_status=$((exit_status | $?))
             /tmp/12345/use < $PERCONA_TOOLKIT_BRANCH/sandbox/gtid_on.sql
             exit_status=$?
             /tmp/12346/use < $PERCONA_TOOLKIT_BRANCH/sandbox/gtid_on.sql

t/lib/MasterSlave.t

@@ -786,31 +786,33 @@ SKIP: {
       },
    );
 
-   our $message;
-   local $SIG{__WARN__} = sub {
-      $message = shift;
+   my $ss;
+   eval {
+       $ss = $ms->get_slave_status($slave1_dbh);
    };
-   my $css = $ms->get_slave_status($slave1_dbh);
-   local $SIG{__WARN__} = undef;
+
    is (
-       $css,
+       $ss,
        undef,
        'Cannot determine slave in a multi source config without --channel param'
    );
+
    like (
-       $message,
-       qr/This server returned more than one row for SHOW SLAVE STATUS/,
-       'Got warning message if we cannot determine slave in a multi source config without --channel param',
+       $EVAL_ERROR,
+       qr/cannot get slave status without using a channel name/,
+       'Got error message if we cannot determine slave in a multi source config without --channel param',
    );
 
-   my $wfm = $ms->wait_for_master(
-      master_status => $ms->get_master_status($dbh),
-      slave_dbh     => $slave1_dbh,
-      timeout       => 1,
-   );
+   eval {
+       my $wfm = $ms->wait_for_master(
+          master_status => $ms->get_master_status($dbh),
+          slave_dbh     => $slave1_dbh,
+          timeout       => 1,
+       );
+   };
    like(
-       $wfm->{error},
-       qr/"channel" was not specified on the command line/,
+       $EVAL_ERROR,
+       qr/cannot get slave status without using a channel name/,
        'Wait for master returned error',
    );
 
@@ -819,35 +821,47 @@ SKIP: {
    # It should return undef
    $slave1_dbh->do("STOP SLAVE for channel 'masterchan2'");
 
-   $css = $ms->get_slave_status($slave1_dbh);
-   is (
-       $css,
-       undef,
-       'Cannot determine slave in a multi source config without --channel param (only one server)'
-   );
+   eval {
+       my $css = $ms->get_slave_status($slave1_dbh);
+       is (
+           $css,
+           undef,
+           'Cannot determine slave in a multi source config without --channel param (only one server)'
+       );
+   };
 
    $slave1_dbh->do("START SLAVE for channel 'masterchan2'");
 
    # Now try specifying a channel name 
    $ms->{channel} = 'masterchan1';
-   $css = $ms->get_slave_status($slave1_dbh);
+   my $css = $ms->get_slave_status($slave1_dbh);
    is (
        $css->{channel_name},
        'masterchan1',
        'Returned the correct slave',
    );
 
-   $wfm = $ms->wait_for_master(
-      master_status => $ms->get_master_status($dbh),
-      slave_dbh     => $slave1_dbh,
-      timeout       => 1,
-   );
+   my $wfm;
+   eval {
+       $wfm = $ms->wait_for_master(
+          master_status => $ms->get_master_status($dbh),
+          slave_dbh     => $slave1_dbh,
+          timeout       => 1,
+       );
+   };
+
    is(
-       $wfm->{error},
-       undef,
+       $EVAL_ERROR,
+       '',
        'Wait for master returned no error',
    );
 
+   isnt(
+       $wfm,
+       undef,
+       'Wait for master returned a hash',
+   );
+
    $sb->stop_sandbox(qw(chan_master1 chan_master2 chan_slave1));
 }
 # #############################################################################

t/pt-slave-restart/pt-slave-restart.t

@@ -132,6 +132,69 @@ is(
    'No error with --quiet (issue 673)'
 );
 
+SKIP: {
+
+   skip "Only test on mysql 5.7" if ( $sandbox_version lt '5.7' );
+
+   my ($master1_dbh, $master1_dsn) = $sb->start_sandbox(
+      server => 'chan_master1',
+      type   => 'master',
+   );
+   my ($master2_dbh, $master2_dsn) = $sb->start_sandbox(
+      server => 'chan_master2',
+      type   => 'master',
+   );
+   my ($slave1_dbh, $slave1_dsn) = $sb->start_sandbox(
+      server => 'chan_slave1',
+      type   => 'master',
+   );
+   my $slave1_port = $sb->port_for('chan_slave1');
+   
+   $sb->load_file('chan_master1', "sandbox/gtid_on.sql", undef, no_wait => 1);
+   $sb->load_file('chan_master2', "sandbox/gtid_on.sql", undef, no_wait => 1);
+   $sb->load_file('chan_slave1', "sandbox/slave_channels_ports.sql", undef, no_wait => 1);
+   my $slave_port = $sb->port_for('chan_slave1');
+                                                             
+   eval {
+      $output = `$trunk/bin/pt-slave-restart -h 127.0.0.1 -P $slave_port -u msandbox -p msandbox  --error-numbers 1205,1317 --run-time 1`;
+   };
+
+   like(
+      $output,
+      qr/cannot get slave status without using a channel name/,
+      'Error on slave using replication channels and no --channel param'
+   );
+
+   $master1_dbh->do('DROP DATABASE IF EXISTS test');
+   $master1_dbh->do('CREATE DATABASE test');
+   $master1_dbh->do('CREATE TABLE test.t (a INT)');
+   sleep(2);
+   
+   # Bust replication
+   $slave1_dbh->do('DROP TABLE test.t');
+   $master1_dbh->do('INSERT INTO test.t SELECT 1');
+   wait_until(
+      sub {
+         my $row = $slave1_dbh->selectrow_hashref('show slave status for channel "masterchan1"');
+         return $row->{last_sql_errno};
+      }
+   );
+   
+   my $r = $slave1_dbh->selectrow_hashref('show slave status');
+   like($r->{last_error}, qr/Table 'test.t' doesn't exist'/, 'It is busted');
+   
+   # Start an instance
+   my $cmd = "$trunk/bin/pt-slave-restart --max-sleep 0.25 -h 127.0.0.1 -P $slave_port "
+           . "-u msandbox -p msandbox --daemonize --channel masterchan1 "
+           . "--pid /tmp/pt-slave-restart.pid --log /tmp/pt-slave-restart.log";
+   diag(`$cmd`);
+   my $output = `ps x | grep 'pt-slave-restart \-\-max\-sleep ' | grep -v grep | grep -v pt-slave-restart.t`;
+   like($output, qr/pt-slave-restart --max/, 'It lives');
+   
+   unlike($output, qr/Table 'test.t' doesn't exist'/, 'It is not busted');
+
+   $sb->stop_sandbox(qw(chan_master1 chan_master2 chan_slave1));
+}
 # #############################################################################
 # Done.
 # #############################################################################