Skip to content

Commit 4ae9000

Browse files
moseshllmoseshll
authored
ETT-316 issues with glacier expiration job (mass_delete edition) (#190)
- Add a `mass_delete` class method to the `ObjectStore` and `PrefixedVersions` storage classes. - Intended to replace the `delete_objects` instance method. - Add companion `object_keys` instance so these can be amassed at a higher level and removed via `mass_delete`. - `BackupExpirationBatch` can still take an arbitrarily large batch size, but due to s3api delete-objects limitation of 1k objects it iterates sub-batches of just under 1k. This way there's no need to mess with the job-size parameter in the parent binary. - Switch to a single `mass_delete` storage call per subbatch.
1 parent 09161a1 commit 4ae9000

4 files changed

Lines changed: 141 additions & 28 deletions

File tree

‎lib/HTFeed/BackupExpiration.pm‎

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -127,19 +127,16 @@ sub run {
127127
sub wait_for_available_worker {
128128
my $self = shift;
129129

130-
if (scalar keys %{$self->{workers}} >= $self->{max_workers}) {
131-
my $pid = 0;
132-
do {
133-
# Wait for any worker. This blocks indefinitely but there's nothing else
134-
# for this process to do but wait.
135-
$pid = waitpid(-1, 0);
136-
if ($pid > 0) {
137-
my $job_file = $self->{workers}->{$pid};
138-
get_logger->trace("worker [$pid] exited with status $? - removing $job_file");
139-
unlink $job_file->filename;
140-
delete $self->{workers}->{$pid};
141-
}
142-
} while ($pid > 0);
130+
while (scalar keys %{$self->{workers}} && scalar keys %{$self->{workers}} >= $self->{max_workers}) {
131+
# Wait for any worker. This blocks indefinitely but there's nothing else
132+
# for this process to do but wait.
133+
my $pid = waitpid(-1, 0);
134+
if ($pid > 0) {
135+
my $job_file = $self->{workers}->{$pid};
136+
get_logger->trace("worker [$pid] exited with status $? - removing $job_file");
137+
unlink $job_file->filename;
138+
delete $self->{workers}->{$pid};
139+
}
143140
}
144141
}
145142

@@ -150,7 +147,7 @@ sub spawn_worker {
150147
my $job_file = File::Temp->new(
151148
DIR => $self->{temp_directory},
152149
SUFFIX => '.tsv',
153-
CLEANUP => 0
150+
UNLINK => 0
154151
);
155152
foreach my $version (@$job) {
156153
print $job_file join("\t", @$version) . "\n";

‎lib/HTFeed/BackupExpirationBatch.pm‎

Lines changed: 52 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -66,14 +66,60 @@ sub run {
6666
my $self = shift;
6767

6868
open(my $fh, '<:encoding(UTF-8)', $self->{job_file}) or die "could not open $$self->{job_file}: $!";
69-
while (my $line = <$fh>) {
70-
chomp $line;
71-
my ($namespace, $id, $version) = split(/\t/, $line, 3);
72-
$self->delete_version($namespace, $id, $version);
69+
my $storage_deletes = [];
70+
my $database_deletes = [];
71+
while (1) {
72+
my $line = <$fh>;
73+
if ($line) {
74+
chomp $line;
75+
my ($namespace, $id, $version) = split(/\t/, $line, 3);
76+
push @$storage_deletes, @{$self->storage_keys($namespace, $id, $version)};
77+
push @$database_deletes, [$namespace, $id, $version];
78+
}
79+
# Now process the (sub)batch if max size or if at EOF.
80+
# Maximum batch size is 1000 for glacier but buy some wiggle room by using 990,
81+
# so we don't go over and have the whole batch fail.
82+
if (!$line || scalar @$storage_deletes >= 990) {
83+
$self->mass_delete($storage_deletes);
84+
$self->mass_update($database_deletes);
85+
$storage_deletes = [];
86+
$database_deletes = [];
87+
}
88+
last unless $line;
7389
}
7490
}
7591

76-
sub delete_version {
92+
# Use storage class `mass_delete` method to delete an arrayref of keys/files.
93+
sub mass_delete {
94+
my $self = shift;
95+
my $storage_deletes = shift;
96+
97+
return if $self->{dry_run};
98+
99+
unless ($self->{storage_config}->{class}->mass_delete(
100+
config => $self->{storage_config},
101+
keys => $storage_deletes
102+
)) {
103+
die sprintf("mass_delete: unable to delete %d volumes", scalar $storage_deletes);
104+
}
105+
}
106+
107+
# Update database to reflect deletion of arrayref of [namespace, id, version]
108+
sub mass_update {
109+
my $self = shift;
110+
my $database_deletes = shift;
111+
112+
return if $self->{dry_run};
113+
114+
foreach my $row (@$database_deletes) {
115+
my ($namespace, $id, $version) = @$row;
116+
get_logger->trace("setting deleted=1 for $namespace.$id version $version");
117+
$self->{update_sth}->execute($namespace, $id, $version, $self->{storage_name});
118+
}
119+
}
120+
121+
# return arrayref of keys/filenames to delete, typically the mets and zip
122+
sub storage_keys {
77123
my $self = shift;
78124
my $namespace = shift;
79125
my $id = shift;
@@ -94,14 +140,7 @@ sub delete_version {
94140
}
95141
$storage->{timestamp} = $version;
96142
$storage->{zip_suffix} = '.gpg';
97-
get_logger->trace("deleting archive for $volume->{namespace}.$volume->{objid} version $version" . $self->{dry_run_text});
98-
return if $self->{dry_run};
99-
100-
unless ($storage->delete_objects) {
101-
die "Unable to delete $volume->{namespace}.$volume->{objid} version $version";
102-
}
103-
get_logger->trace("setting deleted=1 for $volume->{namespace}.$volume->{objid} version $version");
104-
$self->{update_sth}->execute($namespace, $id, $version, $self->{storage_name});
143+
return $storage->object_keys;
105144
}
106145

107146
1;

‎lib/HTFeed/Storage/ObjectStore.pm‎

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,56 @@ use MIME::Base64 qw(decode_base64);
1313
use POSIX qw(strftime);
1414
use JSON::XS ();
1515

16+
# Class method for doing mass deletion of multiple objects at a time.
17+
# Pass arrayref of keys to delete.
18+
sub mass_delete {
19+
my ($class, %args) = @_;
20+
21+
$args{keys} || die("Missing required argument 'keys'");
22+
$args{config} || die("Missing required argument 'config'");
23+
24+
if (scalar @{$args{keys}} > 1000) {
25+
die "can't use s3api delete-objects for more than 1000 objects"
26+
}
27+
28+
my $json_xs = JSON::XS->new;
29+
my $s3 = HTFeed::Storage::S3->new(
30+
bucket => $args{config}{bucket},
31+
awscli => $args{config}{awscli}
32+
);
33+
get_logger->trace("deleting objects");
34+
my $payload = {
35+
'Objects' => [],
36+
'Quiet' => \1
37+
};
38+
foreach my $key (@{$args{keys}}) {
39+
push @{$payload->{Objects}}, { Key => $key };
40+
}
41+
my $json_file = File::Temp->new(
42+
SUFFIX => '.json',
43+
UNLINK => 1
44+
);
45+
46+
print $json_file $json_xs->encode($payload);
47+
$json_file->close;
48+
49+
eval {
50+
$s3->s3api('delete-objects', '--delete', "file://$json_file");
51+
};
52+
if ($@) {
53+
get_logger->error("could not delete $json_file: $@");
54+
return;
55+
}
56+
return 1;
57+
}
58+
59+
# Class method: collect the keys for a mass_delete call.
60+
sub object_keys {
61+
my $self = shift;
62+
63+
return [$self->mets_key, $self->zip_key];
64+
}
65+
1666
sub new {
1767
my $class = shift;
1868
my $self = $class->SUPER::new(@_);
@@ -50,7 +100,6 @@ sub delete_objects {
50100
$self->{s3}->s3api('delete-objects', '--delete', $self->{json_xs}->encode($payload));
51101
};
52102
if ($@) {
53-
print STDERR "\n\n\nWAHAPPA? $@\n\n\n";
54103
$self->set_error(
55104
'OperationFailed',
56105
detail => "delete_objects failed: $@"

‎lib/HTFeed/Storage/PrefixedVersions.pm‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,34 @@ sub zip_audit_class {
2929
return 'HTFeed::StorageAudit::PrefixedVersions';
3030
}
3131

32+
# Class method for doing mass deletion of multiple objects at a time.
33+
# Pass arrayref of keys to delete.
34+
sub mass_delete {
35+
my ($class, %args) = @_;
36+
37+
$args{keys} || die("Missing required argument 'keys'");
38+
$args{config} || die("Missing required argument 'config'");
39+
40+
eval {
41+
foreach my $key (@{$args{keys}}) {
42+
unlink $key;
43+
}
44+
};
45+
46+
if ($@) {
47+
get_logger->error("could not delete: $@");
48+
return;
49+
}
50+
return 1;
51+
}
52+
53+
# Class method: collect the keys (paths in this case) for a mass_delete call.
54+
sub object_keys {
55+
my $self = shift;
56+
57+
return [$self->zip_obj_path, $self->mets_obj_path];
58+
}
59+
3260
sub delete_objects {
3361
my $self = shift;
3462

0 commit comments

Comments
 (0)