View | Details | Raw Unified | Return to bug 35920
Collapse All | Expand All

(-)a/Koha/BackgroundWorker.pm (+193 lines)
Line 0 Link Here
1
package Koha::BackgroundWorker;
2
3
# Copyright 2024 Rijksmuseum
4
#
5
# This file is part of Koha.
6
#
7
# Koha is free software; you can redistribute it and/or modify it
8
# under the terms of the GNU General Public License as published by
9
# the Free Software Foundation; either version 3 of the License, or
10
# (at your option) any later version.
11
#
12
# Koha is distributed in the hope that it will be useful, but
13
# WITHOUT ANY WARRANTY; without even the implied warranty of
14
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15
# GNU General Public License for more details.
16
#
17
# You should have received a copy of the GNU General Public License
18
# along with Koha; if not, see <http://www.gnu.org/licenses>.
19
20
use Modern::Perl;
21
use JSON qw( decode_json );
22
use Time::HiRes;
23
use Parallel::ForkManager;
24
use Try::Tiny qw(try catch);
25
26
use C4::Context;
27
use Koha::BackgroundJobs;
28
use Koha::Logger;
29
30
=head1 NAME
31
32
Koha::BackgroundWorker - Centralized code for background worker scripts
33
34
=head1 SYNOPSIS
35
36
    use Koha::BackgroundWorker;
37
    Koha::BackgroundWorker->run($params);
38
39
=head1 DESCRIPTION
40
41
    This module allows you to use centralized worker code reading
42
    from Rabbit queues and specifying your own callback.
43
44
45
=head1 METHODS
46
47
=head2 run
48
49
    Koha::BackgroundWorker->run($params);
50
51
    Main method to start the worker.
52
    Available parameters:
53
    - callback
54
    - queues
55
    - max_processes
56
    - max_retries
57
    - mq_timeout
58
59
=cut
60
61
sub run {
62
    my ( $class, $params ) = @_;
63
    my $self = bless $params // {}, $class;
64
    $self->_init;
65
    my @queues   = @{ $self->{queues} };
66
    my $callback = $self->{callback};
67
68
    my $pm                = Parallel::ForkManager->new( $self->{max_processes} );
69
    my $not_found_retries = {};
70
71
    my $conn;
72
    try {
73
        $conn = Koha::BackgroundJob->connect;
74
    } catch {
75
        warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
76
    };
77
    if ($conn) {
78
79
        # FIXME cf note in Koha::BackgroundJob about $namespace
80
        my $namespace = C4::Context->config('memcached_namespace');
81
        for my $queue (@queues) {
82
            $conn->subscribe(
83
                {
84
                    destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
85
                    ack              => 'client',
86
                    'prefetch-count' => 1,
87
                }
88
            );
89
        }
90
    }
91
92
    ### The main loop
93
    while (1) {
94
        if ($conn) {
95
            my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } );
96
            if ( !defined $frame ) {
97
98
                # timeout or connection issue?
99
                $pm->reap_finished_children;
100
                next;    # will reconnect automatically
101
            }
102
103
            my $args = try {
104
                my $body = $frame->body;
105
                decode_json($body);    # TODO Should this be from_json? Check utf8 flag.
106
            } catch {
107
                Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ );
108
                return;
109
            };
110
111
            unless ($args) {
112
                Koha::Logger->get( { interface => 'worker' } )
113
                    ->warn( sprintf "Frame does not have correct args, ignoring it" );
114
                $conn->nack( { frame => $frame, requeue => 'false' } );
115
                next;
116
            }
117
118
            my $job = Koha::BackgroundJobs->find( $args->{job_id} );
119
120
            if ( $job && $job->status ne 'new' ) {
121
                Koha::Logger->get( { interface => 'worker' } )
122
                    ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
123
124
                # nack without requeue, we do not want to process this frame again
125
                $conn->nack( { frame => $frame, requeue => 'false' } );
126
                next;
127
            }
128
129
            unless ($job) {
130
                $not_found_retries->{ $args->{job_id} } //= 0;
131
                if ( ++$not_found_retries->{ $args->{job_id} } >= $self->{max_retries} ) {
132
                    Koha::Logger->get( { interface => 'worker' } )
133
                        ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
134
135
                    # nack without requeue, we do not want to process this frame again
136
                    $conn->nack( { frame => $frame, requeue => 'false' } );
137
                    next;
138
                }
139
140
                Koha::Logger->get( { interface => 'worker' } )
141
                    ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
142
143
                # nack to force requeue
144
                $conn->nack( { frame => $frame, requeue => 'true' } );
145
                Time::HiRes::sleep(0.5);
146
                next;
147
            }
148
            $conn->ack( { frame => $frame } );
149
150
            $pm->start and next;
151
            srand();    # ensure each child process begins with a new seed
152
            &$callback( $job, $args );
153
            $pm->finish;
154
155
        } else {
156
            my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } );
157
            while ( my $job = $jobs->next ) {
158
                my $args = try {
159
                    $job->json->decode( $job->data );
160
                } catch {
161
                    Koha::Logger->get( { interface => 'worker' } )
162
                        ->warn( sprintf "Cannot decode data for job id=%s", $job->id );
163
                    $job->status('failed')->store;
164
                    return;
165
                };
166
167
                next unless $args;
168
169
                $pm->start and next;
170
                srand();    # ensure each child process begins with a new seed
171
                &$callback( $job, { job_id => $job->id, %$args } );
172
                $pm->finish;
173
174
            }
175
            sleep 10;
176
        }
177
    }
178
179
    # Work is done
180
    $conn->disconnect;
181
    $pm->wait_all_children;
182
}
183
184
sub _init {
185
    my ($self) = @_;
186
    $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes}
187
        if C4::Context->config('background_jobs_worker');
188
    $self->{max_processes} ||= 1;
189
    $self->{mq_timeout}  //= 10;
190
    $self->{max_retries} //= 10;
191
}
192
193
1;
(-)a/misc/workers/background_jobs_worker.pl (-131 / +13 lines)
Lines 54-202 The different values available are: Link Here
54
=cut
54
=cut
55
55
56
use Modern::Perl;
56
use Modern::Perl;
57
use JSON qw( decode_json );
58
use Try::Tiny;
59
use Pod::Usage;
57
use Pod::Usage;
60
use Getopt::Long;
58
use Getopt::Long;
61
use Parallel::ForkManager;
59
use Try::Tiny qw(try catch);
62
use Time::HiRes;
63
60
64
use C4::Context;
61
use Koha::BackgroundWorker;
65
use Koha::Logger;
66
use Koha::BackgroundJobs;
67
use C4::Context;
68
62
69
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
63
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
70
64
71
my ( $help, @queues );
65
my ( $max_processes, $help, @queues );
72
73
my $max_processes = $ENV{MAX_PROCESSES};
74
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker');
75
$max_processes ||= 1;
76
my $mq_timeout = $ENV{MQ_TIMEOUT} // 10;
77
78
my $not_found_retries = {};
79
my $max_retries = $ENV{MAX_RETRIES} || 10;
80
81
GetOptions(
66
GetOptions(
82
    'm|max-processes=i' => \$max_processes,
67
    'm|max-processes=i' => \$max_processes,
83
    'h|help' => \$help,
68
    'h|help' => \$help,
84
    'queue=s' => \@queues,
69
    'queue=s' => \@queues,
85
) || pod2usage(1);
70
) || pod2usage(1);
86
87
88
pod2usage(0) if $help;
71
pod2usage(0) if $help;
89
72
push @queues, 'default' unless @queues;
90
unless (@queues) {
73
91
    push @queues, 'default';
74
Koha::BackgroundWorker->run(
92
}
75
    {
93
76
        max_processes => $max_processes // $ENV{MAX_PROCESSES},
94
my $conn;
77
        mq_timeout    => $ENV{MQ_TIMEOUT},
95
try {
78
        max_retries   => $ENV{MAX_RETRIES},
96
    $conn = Koha::BackgroundJob->connect;
79
        callback      => \&process_job,
97
} catch {
80
        queues        => \@queues,
98
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
99
};
100
101
my $pm = Parallel::ForkManager->new($max_processes);
102
103
if ( $conn ) {
104
    # FIXME cf note in Koha::BackgroundJob about $namespace
105
    my $namespace = C4::Context->config('memcached_namespace');
106
    for my $queue (@queues) {
107
        $conn->subscribe(
108
            {
109
                destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
110
                ack              => 'client',
111
                'prefetch-count' => 1,
112
            }
113
        );
114
    }
81
    }
115
}
82
);
116
while (1) {
117
    if ( $conn ) {
118
        my $frame = $conn->receive_frame( { timeout => $mq_timeout } );
119
        if ( !defined $frame ) {
120
            # timeout or connection issue?
121
            $pm->reap_finished_children;
122
            next;    # will reconnect automatically
123
        }
124
125
        my $args = try {
126
            my $body = $frame->body;
127
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
128
        } catch {
129
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_);
130
            return;
131
        };
132
133
        unless ( $args ) {
134
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it");
135
            $conn->nack( { frame => $frame, requeue => 'false' } );
136
            next;
137
        }
138
139
        my $job = Koha::BackgroundJobs->find( $args->{job_id} );
140
141
        if ( $job && $job->status ne 'new' ) {
142
            Koha::Logger->get( { interface => 'worker' } )
143
                ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
144
145
            # nack without requeue, we do not want to process this frame again
146
            $conn->nack( { frame => $frame, requeue => 'false' } );
147
            next;
148
        }
149
150
        unless ($job) {
151
            $not_found_retries->{ $args->{job_id} } //= 0;
152
            if ( ++$not_found_retries->{ $args->{job_id} } >= $max_retries ) {
153
                Koha::Logger->get( { interface => 'worker' } )
154
                    ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
155
156
                # nack without requeue, we do not want to process this frame again
157
                $conn->nack( { frame => $frame, requeue => 'false' } );
158
                next;
159
            }
160
161
            Koha::Logger->get( { interface => 'worker' } )
162
                ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
163
164
            # nack to force requeue
165
            $conn->nack( { frame => $frame, requeue => 'true' } );
166
            Time::HiRes::sleep(0.5);
167
            next;
168
        }
169
        $conn->ack( { frame => $frame } );
170
171
        $pm->start and next;
172
        srand();    # ensure each child process begins with a new seed
173
        process_job( $job, $args );
174
        $pm->finish;
175
176
    } else {
177
        my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues });
178
        while ( my $job = $jobs->next ) {
179
            my $args = try {
180
                $job->json->decode($job->data);
181
            } catch {
182
                Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Cannot decode data for job id=%s", $job->id);
183
                $job->status('failed')->store;
184
                return;
185
            };
186
187
            next unless $args;
188
189
            $pm->start and next;
190
            srand();    # ensure each child process begins with a new seed
191
            process_job( $job, { job_id => $job->id, %$args } );
192
            $pm->finish;
193
194
        }
195
        sleep 10;
196
    }
197
}
198
$conn->disconnect;
199
$pm->wait_all_children;
200
83
201
sub process_job {
84
sub process_job {
202
    my ( $job, $args ) = @_;
85
    my ( $job, $args ) = @_;
203
- 

Return to bug 35920