ETT-316 issues with glacier expiration job (mass_delete edition)

lib/HTFeed/BackupExpiration.pm

@@ -127,19 +127,16 @@ sub run {
 sub wait_for_available_worker {
   my $self = shift;
 
-  if (scalar keys %{$self->{workers}} >= $self->{max_workers}) {
-    my $pid = 0;
-    do {
-      # Wait for any worker. This blocks indefinitely but there's nothing else
-      # for this process to do but wait.
-      $pid = waitpid(-1, 0);
-      if ($pid > 0) {
-        my $job_file = $self->{workers}->{$pid};
-        get_logger->trace("worker [$pid] exited with status $? - removing $job_file");
-        unlink $job_file->filename;
-        delete $self->{workers}->{$pid};
-      }
-    } while ($pid > 0);
+  while (scalar keys %{$self->{workers}} && scalar keys %{$self->{workers}} >= $self->{max_workers}) {
+    # Wait for any worker. This blocks indefinitely but there's nothing else
+    # for this process to do but wait.
+    my $pid = waitpid(-1, 0);
+    if ($pid > 0) {
+      my $job_file = $self->{workers}->{$pid};
+      get_logger->trace("worker [$pid] exited with status $? - removing $job_file");
+      unlink $job_file->filename;
+      delete $self->{workers}->{$pid};
+    }
   }
 }
 
@@ -150,7 +147,7 @@ sub spawn_worker {
   my $job_file = File::Temp->new(
     DIR => $self->{temp_directory},
     SUFFIX => '.tsv',
-    CLEANUP => 0
+    UNLINK => 0
   );
   foreach my $version (@$job) {
     print $job_file join("\t", @$version) . "\n";

lib/HTFeed/BackupExpirationBatch.pm

@@ -66,14 +66,60 @@ sub run {
   my $self = shift;
 
   open(my $fh, '<:encoding(UTF-8)', $self->{job_file}) or die "could not open $$self->{job_file}: $!";
-  while (my $line = <$fh>) {
-    chomp $line;
-    my ($namespace, $id, $version) = split(/\t/, $line, 3);
-    $self->delete_version($namespace, $id, $version);
+  my $storage_deletes = [];
+  my $database_deletes = [];
+  while (1) {
+    my $line = <$fh>;
+    if ($line) {
+      chomp $line;
+      my ($namespace, $id, $version) = split(/\t/, $line, 3);
+      push @$storage_deletes, @{$self->storage_keys($namespace, $id, $version)};
+      push @$database_deletes, [$namespace, $id, $version];
+    }
+    # Now process the (sub)batch if max size or if at EOF.
+    # Maximum batch size is 1000 for glacier but buy some wiggle room by using 990,
+    # so we don't go over and have the whole batch fail.
+    if (!$line || scalar @$storage_deletes >= 990) {
+      $self->mass_delete($storage_deletes);
+      $self->mass_update($database_deletes);
+      $storage_deletes = [];
+      $database_deletes = [];
+    }
+    last unless $line;
   }
 }
 
-sub delete_version {
+# Use storage class `mass_delete` method to delete an arrayref of keys/files.
+sub mass_delete {
+  my $self = shift;
+  my $storage_deletes = shift;
+
+  return if $self->{dry_run};
+
+  unless ($self->{storage_config}->{class}->mass_delete(
+    config => $self->{storage_config},
+    keys => $storage_deletes
+  )) {
+    die sprintf("mass_delete: unable to delete %d volumes", scalar $storage_deletes);
+  }
+}
+
+# Update database to reflect deletion of arrayref of [namespace, id, version]
+sub mass_update {
+  my $self = shift;
+  my $database_deletes = shift;
+
+  return if $self->{dry_run};
+
+  foreach my $row (@$database_deletes) {
+    my ($namespace, $id, $version) = @$row;
+    get_logger->trace("setting deleted=1 for $namespace.$id version $version");
+    $self->{update_sth}->execute($namespace, $id, $version, $self->{storage_name});
+  }
+}
+
+# return arrayref of keys/filenames to delete, typically the mets and zip
+sub storage_keys {
   my $self = shift;
   my $namespace = shift;
   my $id = shift;
@@ -94,14 +140,7 @@ sub delete_version {
   }
   $storage->{timestamp} = $version;
   $storage->{zip_suffix} = '.gpg';
-  get_logger->trace("deleting archive for $volume->{namespace}.$volume->{objid} version $version" . $self->{dry_run_text});
-  return if $self->{dry_run};
-
-  unless ($storage->delete_objects) {
-    die "Unable to delete $volume->{namespace}.$volume->{objid} version $version";
-  }
-  get_logger->trace("setting deleted=1 for $volume->{namespace}.$volume->{objid} version $version");
-  $self->{update_sth}->execute($namespace, $id, $version, $self->{storage_name});
+  return $storage->object_keys;
 }
 
 1;

lib/HTFeed/Storage/ObjectStore.pm

@@ -13,6 +13,56 @@ use MIME::Base64 qw(decode_base64);
 use POSIX qw(strftime);
 use JSON::XS ();
 
+# Class method for doing mass deletion of multiple objects at a time.
+# Pass arrayref of keys to delete.
+sub mass_delete {
+  my ($class, %args)  = @_;
+
+  $args{keys} || die("Missing required argument 'keys'");
+  $args{config} || die("Missing required argument 'config'");
+
+  if (scalar @{$args{keys}} > 1000) {
+    die "can't use s3api delete-objects for more than 1000 objects"
+  }
+
+  my $json_xs = JSON::XS->new;
+  my $s3 = HTFeed::Storage::S3->new(
+    bucket => $args{config}{bucket},
+    awscli => $args{config}{awscli}
+  );
+  get_logger->trace("deleting objects");
+  my $payload = {
+    'Objects' => [],
+    'Quiet' => \1
+  };
+  foreach my $key (@{$args{keys}}) {
+    push @{$payload->{Objects}}, { Key => $key };
+  }
+  my $json_file = File::Temp->new(
+    SUFFIX => '.json',
+    UNLINK => 1
+  );
+
+  print $json_file $json_xs->encode($payload);
+  $json_file->close;
+
+  eval {
+    $s3->s3api('delete-objects', '--delete', "file://$json_file");
+  };
+  if ($@) {
+    get_logger->error("could not delete $json_file: $@");
+    return;
+  }
+  return 1;
+}
+
+# Class method: collect the keys for a mass_delete call.
+sub object_keys {
+  my $self = shift;
+
+  return [$self->mets_key, $self->zip_key];
+}
+
 sub new {
     my $class = shift;
     my $self  = $class->SUPER::new(@_);
@@ -50,7 +100,6 @@ sub delete_objects {
       $self->{s3}->s3api('delete-objects', '--delete', $self->{json_xs}->encode($payload));
     };
     if ($@) {
-    print STDERR "\n\n\nWAHAPPA? $@\n\n\n";
         $self->set_error(
             'OperationFailed',
             detail => "delete_objects failed: $@"

lib/HTFeed/Storage/PrefixedVersions.pm

@@ -29,6 +29,34 @@ sub zip_audit_class {
   return 'HTFeed::StorageAudit::PrefixedVersions';
 }
 
+# Class method for doing mass deletion of multiple objects at a time.
+# Pass arrayref of keys to delete.
+sub mass_delete {
+  my ($class, %args)  = @_;
+
+  $args{keys} || die("Missing required argument 'keys'");
+  $args{config} || die("Missing required argument 'config'");
+
+  eval {
+    foreach my $key (@{$args{keys}}) {
+      unlink $key;
+    }
+  };
+
+  if ($@) {
+    get_logger->error("could not delete: $@");
+    return;
+  }
+  return 1;
+}
+
+# Class method: collect the keys (paths in this case) for a mass_delete call.
+sub object_keys {
+  my $self = shift;
+
+  return [$self->zip_obj_path, $self->mets_obj_path];
+}
+
 sub delete_objects {
   my $self = shift;