diff --git a/lib/HTFeed/BackupExpiration.pm b/lib/HTFeed/BackupExpiration.pm index a37657e7..678a1363 100644 --- a/lib/HTFeed/BackupExpiration.pm +++ b/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"; diff --git a/lib/HTFeed/BackupExpirationBatch.pm b/lib/HTFeed/BackupExpirationBatch.pm index 7c078fb9..feb097ca 100644 --- a/lib/HTFeed/BackupExpirationBatch.pm +++ b/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; diff --git a/lib/HTFeed/Storage/ObjectStore.pm b/lib/HTFeed/Storage/ObjectStore.pm index 0659acd5..f3b61c5e 100644 --- a/lib/HTFeed/Storage/ObjectStore.pm +++ b/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: $@" diff --git a/lib/HTFeed/Storage/PrefixedVersions.pm b/lib/HTFeed/Storage/PrefixedVersions.pm index fdf9d78f..87b29da3 100644 --- a/lib/HTFeed/Storage/PrefixedVersions.pm +++ b/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;