Implement retry_on_error for Pipeline processes. Retry iteration proc twice, then fail completely. Fix mirror.t.

bin/pt-query-digest

@@ -11745,6 +11745,9 @@ sub add {
 
    push @{$self->{procs}}, $process;
    push @{$self->{names}}, $name;
+   if ( my $n = $args{retry_on_error} ) {
+      $self->{retries}->{$name} = $n;
+   }
    if ( $self->{instrument} ) {
       $self->{instrumentation}->{$name} = { time => 0, calls => 0 };
    }
@@ -11809,10 +11812,28 @@ sub execute {
          }
       };
       if ( $EVAL_ERROR ) {
-         warn "Pipeline process $procno ("
+         my $name = $self->{names}->[$procno] || "";
-            . ($self->{names}->[$procno] || "")
+         my $msg  = "Pipeline process " . ($procno + 1)
-            . ") caused an error: $EVAL_ERROR";
+                  . " ($name) caused an error: "
-         die $EVAL_ERROR unless $self->{continue_on_error};
+                  . $EVAL_ERROR;
+         if ( defined $self->{retries}->{$name} ) {
+            my $n = $self->{retries}->{$name};
+            if ( $n ) {
+               warn $msg . "Will retry pipeline process $procno ($name) "
+                  . "$n more " . ($n > 1 ? "times" : "time") . ".\n";
+               $self->{retries}->{$name}--;
+            }
+            else {
+               die $msg . "Terminating pipeline because process $procno "
+                  . "($name) caused too many errors.\n";
+            }
+         }
+         elsif ( !$self->{continue_on_error} ) {
+            die $msg;
+         }
+         else {
+            warn $msg;
+         }
       }
    }
 
@@ -12562,6 +12583,12 @@ sub main {
 
    { # iteration
       $pipeline->add(
+         # This is a critical proc: if we die here, we probably need
+         # to stop, else an infinite loop can develop:
+         # https://bugs.launchpad.net/percona-toolkit/+bug/888114
+         # We'll retry twice in case the problem is just one bad
+         # query class, or something like that.
+         retry_on_error => 2,
          name           => 'iteration',
          process        => sub {
             my ( $args ) = @_;

lib/Pipeline.pm

@@ -71,6 +71,9 @@ sub add {
 
    push @{$self->{procs}}, $process;
    push @{$self->{names}}, $name;
+   if ( my $n = $args{retry_on_error} ) {
+      $self->{retries}->{$name} = $n;
+   }
    if ( $self->{instrument} ) {
       $self->{instrumentation}->{$name} = { time => 0, calls => 0 };
    }
@@ -156,10 +159,28 @@ sub execute {
          }
       };
       if ( $EVAL_ERROR ) {
-         warn "Pipeline process $procno ("
+         my $name = $self->{names}->[$procno] || "";
-            . ($self->{names}->[$procno] || "")
+         my $msg  = "Pipeline process " . ($procno + 1)
-            . ") caused an error: $EVAL_ERROR";
+                  . " ($name) caused an error: "
-         die $EVAL_ERROR unless $self->{continue_on_error};
+                  . $EVAL_ERROR;
+         if ( defined $self->{retries}->{$name} ) {
+            my $n = $self->{retries}->{$name};
+            if ( $n ) {
+               warn $msg . "Will retry pipeline process $procno ($name) "
+                  . "$n more " . ($n > 1 ? "times" : "time") . ".\n";
+               $self->{retries}->{$name}--;
+            }
+            else {
+               die $msg . "Terminating pipeline because process $procno "
+                  . "($name) caused too many errors.\n";
+            }
+         }
+         elsif ( !$self->{continue_on_error} ) {
+            die $msg;
+         }
+         else {
+            warn $msg;
+         }
       }
    }
 

t/lib/Pipeline.t

@@ -9,7 +9,7 @@ BEGIN {
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
-use Test::More tests => 10;
+use Test::More tests => 11;
 
 use Time::HiRes qw(usleep);
 
@@ -206,11 +206,13 @@ is(
 $oktorun  = 1;
 $pipeline = new Pipeline(continue_on_error=>1);
 
+my @called = ();
 my @die    = qw(1 0);
 $pipeline->add(
    name    => 'proc1',
    process => sub {
       my ( $args ) = @_;
+      push @called, 'proc1';
       die "I'm an error" if shift @die;
       return $args;
    },
@@ -218,7 +220,7 @@ $pipeline->add(
 $pipeline->add(
    name    => 'proc2',
    process => sub {
-      print "proc2";
+      push @called, 'proc2';
       $oktorun = 0;
       return;
    },
@@ -229,12 +231,48 @@ $output = output(
    stderr => 1,
 );
 
-like(
+is_deeply(
-   $output,
+   \@called,
-   qr/0 \(proc1\) caused an error.+proc2/s,
+   [qw(proc1 proc1 proc2)],
    "Continues on error"
 );
 
+# Override global for critical procs that must kill the pipeline if they die.
+$oktorun  = 1;
+$pipeline = new Pipeline(continue_on_error=>1);
+
+@called = ();
+$pipeline->add(
+   name    => 'proc1',
+   process => sub {
+      my ( $args ) = @_;
+      push @called, 'proc1';
+      $oktorun = 0 if @called > 8;
+      return $args;
+   },
+);
+$pipeline->add(
+   retry_on_error => 2,
+   name           => 'proc2',
+   process        => sub {
+      push @called, 'proc2';
+      die;
+   },
+);
+
+$output = output(
+   sub { $pipeline->execute(%args) },
+   stderr => 1,
+);
+
+is_deeply(
+   \@called,
+   [qw(proc1 proc2),  # first attempt
+    qw(proc1 proc2),  # retry 1/2
+    qw(proc1 proc2)], # retry 2/2
+   "Retry proc"
+) or diag($output);
+
 # #############################################################################
 # Done.
 # #############################################################################

t/pt-query-digest/continue_on_error.t

@@ -49,8 +49,8 @@ $output = output(
 
 like(
    $output,
-   qr/Query\s+1/,
+   qr/Argument \S+ isn't numeric/,
-   "No infinite loop in report crashes (bug 888114)"
+   "Report crashed, but no infinite loop (bug 888114)"
 ); 
 
 # #############################################################################

t/pt-query-digest/mirror.t

@@ -36,9 +36,6 @@ my $cmd;
 my $pid_file = "/tmp/pt-query-digest-mirror-test.pid";
 diag(`rm $pid_file 2>/dev/null`);
 
-my $pid_file = '/tmp/pt-query-digest.test.pid';
-`rm -rf $pid_file >/dev/null`;
-
 # ##########################################################################
 # Tests for swapping --processlist and --execute
 # ##########################################################################