Skip to content

Commit 02f7989

Browse files
moseshllmoseshll
authored
ETT-316 issues with glacier expiration job (#186)
* ETT-316 issues with glacier expiration job - `BackupExpiration.pm` forks/execs worker processes to handle deletion in parallel. - Existing test suite left as intact as possible. - Add `--job-size`, `--limit`, and `--workers` flags to `expire_backups.pl` - Fix brittle tests with potential collisions from `old_random_timestamp` and `new_random_timestamp`
1 parent 4c64518 commit 02f7989

6 files changed

Lines changed: 427 additions & 74 deletions

File tree

‎Dockerfile‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
3737
libreadonly-xs-perl \
3838
libroman-perl \
3939
libtest-class-perl \
40+
libtest-exception-perl \
4041
libtest-mockobject-perl \
4142
libtest-most-perl \
4243
libtest-spec-perl \

‎bin/expire_backups.pl‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,17 +11,31 @@
1111
use HTFeed::BackupExpiration;
1212

1313
my $dry_run = 0; # -d
14+
my $job_size = 10000; # -j
15+
my $limit = undef; # --limit
1416
my $storage_name = undef; # -s
17+
my $workers = 1;
1518
my $help = 0;
1619

1720
GetOptions(
1821
'dry-run|d' => \$dry_run,
22+
'job-size|j=i' => \$job_size,
23+
'limit|l=i' => \$limit,
1924
'storage|s=s' => \$storage_name,
25+
'workers|w=i' => \$workers,
2026
'help|?' => \$help
2127
) or pod2usage(2);
2228
pod2usage(1) if $help;
2329

24-
my $exp = HTFeed::BackupExpiration->new(storage_name => $storage_name, dry_run => $dry_run);
30+
$workers = 1 if $workers < 1;
31+
32+
my $exp = HTFeed::BackupExpiration->new(
33+
dry_run => $dry_run,
34+
job_size => $job_size,
35+
limit => $limit,
36+
max_workers => $workers,
37+
storage_name => $storage_name,
38+
);
2539
$exp->run();
2640

2741
__END__
@@ -32,8 +46,10 @@ =head1 NAME
3246
3347
=head1 SYNOPSIS
3448
35-
expire_backups.pl [--dry-run] -s STORAGE_NAME
36-
37-
STORAGE_NAME - storage class name matched against feed_backups.storage_name
49+
expire_backups.pl [--dry-run] [--job-size JOB_SIZE] [--limit LIMIT] [--workers WORKER_COUNT] -s STORAGE_NAME
3850
51+
JOB_SIZE - number of objects to delete, per worker. Default 10,000.
52+
LIMIT - maximum rows of feed_backups to fetch per run. Default is no SQL LIMIT. Can be 0.
53+
STORAGE_NAME - storage class name matched against feed_backups.storage_name
54+
WORKER_COUNT - maximum number of subprocesses to spawn
3955
=cut

‎bin/expire_versions.pl‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
#!/usr/bin/perl
2+
3+
use warnings;
4+
use strict;
5+
6+
use FindBin;
7+
use Getopt::Long qw(:config no_ignore_case);
8+
use Pod::Usage;
9+
use YAML::XS;
10+
11+
use lib "$FindBin::Bin/../lib";
12+
use HTFeed::BackupExpirationBatch;
13+
use HTFeed::Log { root_logger => 'INFO, screen' };
14+
15+
my $dry_run = 0; # -d
16+
my $storage_config_file = undef; # -c
17+
my $storage_name = undef; # -s
18+
my $help = 0;
19+
20+
GetOptions(
21+
'config|c=s' => \$storage_config_file,
22+
'dry-run|d' => \$dry_run,
23+
'storage|s=s' => \$storage_name,
24+
'help|?' => \$help
25+
) or pod2usage(2);
26+
pod2usage(1) if $help;
27+
28+
if (scalar @ARGV != 1) {
29+
die "path to job file is required";
30+
}
31+
32+
my $storage_config = undef;
33+
if ($storage_config_file) {
34+
$storage_config = YAML::XS::LoadFile($storage_config_file);
35+
}
36+
37+
my $exp = HTFeed::BackupExpirationBatch->new(
38+
dry_run => $dry_run,
39+
job_file => $ARGV[0],
40+
storage_config => $storage_config,
41+
storage_name => $storage_name
42+
);
43+
$exp->run();
44+
45+
__END__
46+
47+
=head1 NAME
48+
49+
expire_versions.pl - remove a batch of superseded material from backup storage.
50+
51+
=head1 SYNOPSIS
52+
53+
expire_versions.pl [--config STORAGE_CONFIG_FILE] [--dry-run] -s STORAGE_NAME PATH_TO_JOB_FILE
54+
55+
STORAGE_CONFIG_FILE - path to a YAML file with config hash for STORAGE_NAME
56+
STORAGE_NAME - storage class name matched against feed_backups.storage_name
57+
PATH_TO_JOB_FILE - path to a TSV file with format namespace<TAB>id<TAB>version
58+
=cut

‎lib/HTFeed/BackupExpiration.pm‎

Lines changed: 144 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -3,103 +3,188 @@ package HTFeed::BackupExpiration;
33

44
use strict;
55
use warnings;
6+
67
use HTFeed::Config qw(get_config);
78
use HTFeed::DBTools qw(get_dbh);
89
use HTFeed::Volume;
9-
use Carp;
10+
11+
use Carp ();
12+
use File::Spec ();
13+
use File::Temp ();
1014
use Log::Log4perl qw(get_logger);
11-
use File::Temp;
15+
use YAML::XS ();
16+
17+
my $select_expired_sql = <<~'SQL';
18+
SELECT namespace,id
19+
FROM feed_backups
20+
WHERE deleted IS NULL
21+
AND storage_name=?
22+
AND version < DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 180 DAY),"%Y%m%d%H%i%S")
23+
GROUP BY namespace,id
24+
HAVING COUNT(*) > 1
25+
SQL
26+
27+
my $select_versions_sql = <<~'SQL';
28+
SELECT version
29+
FROM feed_backups
30+
WHERE deleted IS NULL
31+
AND storage_name=?
32+
AND namespace=?
33+
AND id=?
34+
AND version < DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 180 DAY),"%Y%m%d%H%i%S")
35+
ORDER BY version DESC
36+
SQL
1237

13-
use HTFeed::Storage::PrefixedVersions;
14-
use HTFeed::Storage::ObjectStore;
1538

1639
sub new {
1740
my $class = shift;
1841

1942
my $self = {
2043
storage_name => undef,
44+
custom_storage_config => 0,
45+
max_workers => 8,
46+
job_size => 10000,
47+
limit => undef,
2148
@_
2249
};
2350

2451
unless ($self->{storage_name}) {
25-
croak "$class cannot be constructed without a storage name";
52+
Carp::croak "$class cannot be constructed without a storage name";
53+
}
54+
55+
# Test can init with `storage_config` because it is transient and must be known to workers.
56+
# Production just reads the config as normal
57+
if ($self->{storage_config}) {
58+
$self->{custom_storage_config} = 1;
59+
} else {
60+
my $config = get_config('storage_classes');
61+
my $storage_config = $config->{$self->{storage_name}};
62+
die("Can't find storage configuration for " . $self->{storage_name}) unless $storage_config;
63+
$self->{storage_config} = $storage_config;
2664
}
2765

66+
$self->{temp_directory} = File::Temp->newdir;
67+
$self->{workers} = {};
68+
2869
bless($self, $class);
2970
return $self;
3071
}
3172

3273
sub run {
3374
my $self = shift;
3475

35-
my $dry_run = $self->{dry_run};
36-
my $dry_run_text = "";
37-
$dry_run_text = " (DRY RUN)" if $dry_run;
38-
39-
my $config = get_config('storage_classes');
40-
my $storage_config = $config->{$self->{storage_name}};
41-
die("Can't find storage configuration for " . $self->{storage_name}) unless $storage_config;
42-
43-
# find everything with more than one version that is at least 6 months old
44-
# delete all but the most recent > 6 months old version
45-
my $sth = get_dbh()->prepare(<<'SQL');
46-
SELECT namespace,id
47-
FROM feed_backups
48-
WHERE deleted IS NULL
49-
AND storage_name=?
50-
AND version < DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 180 DAY),"%Y%m%d%H%i%S")
51-
GROUP BY namespace,id
52-
HAVING COUNT(*) > 1
53-
SQL
54-
55-
my $versions_sth = get_dbh()->prepare(<<'SQL');
56-
SELECT version
57-
FROM feed_backups
58-
WHERE deleted IS NULL
59-
AND storage_name=?
60-
AND namespace=?
61-
AND id=?
62-
AND version < DATE_FORMAT(DATE_SUB(NOW(), INTERVAL 180 DAY),"%Y%m%d%H%i%S")
63-
ORDER BY version DESC
64-
SQL
76+
# Write storage config to the temp directory for child processes to get at it.
77+
# Unnecessary for production, needed for testing because it is generated as part
78+
# of the test suite.
79+
if ($self->{custom_storage_config}) {
80+
$self->{storage_config_file} = File::Spec->catfile($self->{temp_directory}, 'storage_config.yml');
81+
my $yaml = YAML::XS::Dump($self->{storage_config});
82+
open(my $fh, '>', $self->{storage_config_file}) or die "Could not open storage config YAML $!";
83+
print $fh $yaml;
84+
close $fh;
85+
}
6586

66-
my $update_sth = get_dbh()->prepare(<<'SQL');
67-
UPDATE feed_backups SET deleted=1
68-
WHERE namespace=?
69-
AND id=?
70-
AND version=?
71-
AND storage_name=?
72-
SQL
87+
my $sth = get_dbh()->prepare($self->select_expired_sql);
88+
my $versions_sth = get_dbh()->prepare($select_versions_sql);
7389

90+
my $job = [];
91+
# Iterate over the entirety of feed_backups
92+
# Reaching the end and restarting the query must take place at a
93+
# higher level, perhaps with $self->run called repeatedly.
7494
$sth->execute($self->{storage_name});
7595
while (my $row = $sth->fetchrow_hashref) {
7696
$versions_sth->execute($self->{storage_name}, $row->{namespace}, $row->{id});
7797
my @versions = map { $_->[0]; } @{$versions_sth->fetchall_arrayref};
7898
shift @versions; # jettison the most recent
7999
foreach my $version (@versions) {
80-
my $volume = new HTFeed::Volume(namespace => $row->{namespace},
81-
objid => $row->{id},
82-
package_type => 'ht');
83-
my $storage = $storage_config->{class}->new(volume => $volume,
84-
config => $storage_config,
85-
name => $self->{storage_name});
86-
unless (defined $storage) {
87-
die "Unable to get storage for $volume->{namespace}.$volume->{objid}";
100+
push(@$job, [$row->{namespace}, $row->{id}, $version]);
101+
# Do we have enough to spawn a worker?
102+
if (scalar @$job >= $self->{job_size}) {
103+
$self->wait_for_available_worker;
104+
$self->spawn_worker($job);
105+
$job = [];
88106
}
89-
$storage->{timestamp} = $version;
90-
$storage->{zip_suffix} = '.gpg';
91-
get_logger->trace("deleting archive for $volume->{namespace}.$volume->{objid} version $version" . $dry_run_text);
92-
next if $dry_run;
93-
unless ($storage->delete_objects) {
94-
die "Unable to delete $volume->{namespace}.$volume->{objid}";
107+
}
108+
}
109+
# Submit the leftovers if any
110+
if (scalar @$job > 0) {
111+
$self->wait_for_available_worker;
112+
$self->spawn_worker($job);
113+
$job = [];
114+
}
115+
# Set max workers to 0 so we wait for all of them to finish.
116+
$self->{max_workers} = 0;
117+
# Wait for all the workers to finish.
118+
while (scalar keys %{$self->{workers}} > 0) {
119+
$self->wait_for_available_worker;
120+
}
121+
}
122+
123+
# waitpid on existing workers (if any) until one finishes up
124+
# but only if we are at maximum capacity.
125+
# Only waits for workers to finish if we already have the maximum number on the go,
126+
# or if we are finished and have set the maximum to 0.
127+
sub wait_for_available_worker {
128+
my $self = shift;
129+
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};
95141
}
96-
get_logger->trace("setting deleted=1 for $volume->{namespace}.$volume->{objid} version $version");
97-
$update_sth->execute($row->{namespace}, $row->{id},
98-
$version, $self->{storage_name});
142+
} while ($pid > 0);
143+
}
144+
}
145+
146+
sub spawn_worker {
147+
my $self = shift;
148+
my $job = shift;
149+
150+
my $job_file = File::Temp->new(
151+
DIR => $self->{temp_directory},
152+
SUFFIX => '.tsv',
153+
CLEANUP => 0
154+
);
155+
foreach my $version (@$job) {
156+
print $job_file join("\t", @$version) . "\n";
157+
}
158+
$job_file->close;
159+
my $pid = fork();
160+
if (!defined $pid) {
161+
die "Fork failed: $!";
162+
} elsif ($pid == 0) {
163+
# WORKER PROCESS
164+
my $worker_script = File::Spec->catfile($ENV{FEED_HOME}, 'bin', 'expire_versions.pl');
165+
my @cmd = ('perl', $worker_script, '-s', $self->{storage_name});
166+
if ($self->{custom_storage_config}) {
167+
push @cmd, '--config', $self->{storage_config_file};
168+
}
169+
if ($self->{dry_run}) {
170+
push @cmd, '--dry-run';
99171
}
172+
push @cmd, $job_file->filename;
173+
exec(@cmd) or die "worker [$$] exec failed to run: $!\n";
174+
} else {
175+
# PARENT PROCESS
176+
get_logger->trace("worker [$pid] started with $job_file (" . scalar(@$job) . " items)");
177+
$self->{workers}->{$pid} = $job_file;
100178
}
101179
}
102180

181+
sub select_expired_sql {
182+
my $self = shift;
183+
184+
my $limit_clause = (defined $self->{limit}) ? " LIMIT $self->{limit}" : '';
185+
return $select_expired_sql . $limit_clause;
186+
}
187+
103188
1;
104189

105190
__END__

0 commit comments

Comments
 (0)