Bugzilla – Attachment 162951 Details for
Bug 35920
Centralize code from workers
Home
|
New
|
Browse
|
Search
|
[?]
|
Reports
|
Help
|
New Account
|
Log In
[x]
|
Forgot Password
Login:
[x]
[patch]
Bug 35920: Centralize worker code
Bug-35920-Centralize-worker-code.patch (text/plain), 12.29 KB, created by
Marcel de Rooy
on 2024-03-08 11:03:28 UTC
(
hide
)
Description:
Bug 35920: Centralize worker code
Filename:
MIME Type:
Creator:
Marcel de Rooy
Created:
2024-03-08 11:03:28 UTC
Size:
12.29 KB
patch
obsolete
>From 403083ee6e3951f4e93985f73a0d9147b2b1fdb9 Mon Sep 17 00:00:00 2001 >From: Marcel de Rooy <m.de.rooy@rijksmuseum.nl> >Date: Fri, 8 Mar 2024 08:45:15 +0000 >Subject: [PATCH] Bug 35920: Centralize worker code >Content-Type: text/plain; charset=utf-8 > >Test plan: >Confirm that running jobs still works as expected. >Restart workers and try to stage a MARC file and import it. >--- > Koha/BackgroundWorker.pm | 193 +++++++++++++++++++++++++ > misc/workers/background_jobs_worker.pl | 143 ++---------------- > 2 files changed, 206 insertions(+), 130 deletions(-) > create mode 100644 Koha/BackgroundWorker.pm > >diff --git a/Koha/BackgroundWorker.pm b/Koha/BackgroundWorker.pm >new file mode 100644 >index 0000000000..760ab8fe2a >--- /dev/null >+++ b/Koha/BackgroundWorker.pm >@@ -0,0 +1,193 @@ >+package Koha::BackgroundWorker; >+ >+# Copyright 2024 Rijksmuseum >+# >+# This file is part of Koha. >+# >+# Koha is free software; you can redistribute it and/or modify it >+# under the terms of the GNU General Public License as published by >+# the Free Software Foundation; either version 3 of the License, or >+# (at your option) any later version. >+# >+# Koha is distributed in the hope that it will be useful, but >+# WITHOUT ANY WARRANTY; without even the implied warranty of >+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the >+# GNU General Public License for more details. >+# >+# You should have received a copy of the GNU General Public License >+# along with Koha; if not, see <http://www.gnu.org/licenses>. >+ >+use Modern::Perl; >+use JSON qw( decode_json ); >+use Time::HiRes; >+use Parallel::ForkManager; >+use Try::Tiny qw(try catch); >+ >+use C4::Context; >+use Koha::BackgroundJobs; >+use Koha::Logger; >+ >+=head1 NAME >+ >+Koha::BackgroundWorker - Centralized code for background worker scripts >+ >+=head1 SYNOPSIS >+ >+ use Koha::BackgroundWorker; >+ Koha::BackgroundWorker->run($params); >+ >+=head1 DESCRIPTION >+ >+ This module allows you to use centralized worker code reading >+ from Rabbit queues and specifying your own callback. >+ >+ >+=head1 METHODS >+ >+=head2 run >+ >+ Koha::BackgroundWorker->run($params); >+ >+ Main method to start the worker. >+ Available parameters: >+ - callback >+ - queues >+ - max_processes >+ - max_retries >+ - mq_timeout >+ >+=cut >+ >+sub run { >+ my ( $class, $params ) = @_; >+ my $self = bless $params // {}, $class; >+ $self->_init; >+ my @queues = @{ $self->{queues} }; >+ my $callback = $self->{callback}; >+ >+ my $pm = Parallel::ForkManager->new( $self->{max_processes} ); >+ my $not_found_retries = {}; >+ >+ my $conn; >+ try { >+ $conn = Koha::BackgroundJob->connect; >+ } catch { >+ warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; >+ }; >+ if ($conn) { >+ >+ # FIXME cf note in Koha::BackgroundJob about $namespace >+ my $namespace = C4::Context->config('memcached_namespace'); >+ for my $queue (@queues) { >+ $conn->subscribe( >+ { >+ destination => sprintf( "/queue/%s-%s", $namespace, $queue ), >+ ack => 'client', >+ 'prefetch-count' => 1, >+ } >+ ); >+ } >+ } >+ >+ ### The main loop >+ while (1) { >+ if ($conn) { >+ my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } ); >+ if ( !defined $frame ) { >+ >+ # timeout or connection issue? >+ $pm->reap_finished_children; >+ next; # will reconnect automatically >+ } >+ >+ my $args = try { >+ my $body = $frame->body; >+ decode_json($body); # TODO Should this be from_json? Check utf8 flag. >+ } catch { >+ Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ ); >+ return; >+ }; >+ >+ unless ($args) { >+ Koha::Logger->get( { interface => 'worker' } ) >+ ->warn( sprintf "Frame does not have correct args, ignoring it" ); >+ $conn->nack( { frame => $frame, requeue => 'false' } ); >+ next; >+ } >+ >+ my $job = Koha::BackgroundJobs->find( $args->{job_id} ); >+ >+ if ( $job && $job->status ne 'new' ) { >+ Koha::Logger->get( { interface => 'worker' } ) >+ ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status ); >+ >+ # nack without requeue, we do not want to process this frame again >+ $conn->nack( { frame => $frame, requeue => 'false' } ); >+ next; >+ } >+ >+ unless ($job) { >+ $not_found_retries->{ $args->{job_id} } //= 0; >+ if ( ++$not_found_retries->{ $args->{job_id} } >= $self->{max_retries} ) { >+ Koha::Logger->get( { interface => 'worker' } ) >+ ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} ); >+ >+ # nack without requeue, we do not want to process this frame again >+ $conn->nack( { frame => $frame, requeue => 'false' } ); >+ next; >+ } >+ >+ Koha::Logger->get( { interface => 'worker' } ) >+ ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} ); >+ >+ # nack to force requeue >+ $conn->nack( { frame => $frame, requeue => 'true' } ); >+ Time::HiRes::sleep(0.5); >+ next; >+ } >+ $conn->ack( { frame => $frame } ); >+ >+ $pm->start and next; >+ srand(); # ensure each child process begins with a new seed >+ &$callback( $job, $args ); >+ $pm->finish; >+ >+ } else { >+ my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } ); >+ while ( my $job = $jobs->next ) { >+ my $args = try { >+ $job->json->decode( $job->data ); >+ } catch { >+ Koha::Logger->get( { interface => 'worker' } ) >+ ->warn( sprintf "Cannot decode data for job id=%s", $job->id ); >+ $job->status('failed')->store; >+ return; >+ }; >+ >+ next unless $args; >+ >+ $pm->start and next; >+ srand(); # ensure each child process begins with a new seed >+ &$callback( $job, { job_id => $job->id, %$args } ); >+ $pm->finish; >+ >+ } >+ sleep 10; >+ } >+ } >+ >+ # Work is done >+ $conn->disconnect; >+ $pm->wait_all_children; >+} >+ >+sub _init { >+ my ($self) = @_; >+ $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes} >+ if C4::Context->config('background_jobs_worker'); >+ $self->{max_processes} ||= 1; >+ $self->{mq_timeout} //= 10; >+ $self->{max_retries} //= 10; >+} >+ >+1; >diff --git a/misc/workers/background_jobs_worker.pl b/misc/workers/background_jobs_worker.pl >index 5b432ec11f..fd28dd2b1a 100755 >--- a/misc/workers/background_jobs_worker.pl >+++ b/misc/workers/background_jobs_worker.pl >@@ -54,149 +54,32 @@ The different values available are: > =cut > > use Modern::Perl; >-use JSON qw( decode_json ); >-use Try::Tiny; > use Pod::Usage; > use Getopt::Long; >-use Parallel::ForkManager; >-use Time::HiRes; >+use Try::Tiny qw(try catch); > >-use C4::Context; >-use Koha::Logger; >-use Koha::BackgroundJobs; >-use C4::Context; >+use Koha::BackgroundWorker; > > $SIG{'PIPE'} = 'IGNORE'; # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu. > >-my ( $help, @queues ); >- >-my $max_processes = $ENV{MAX_PROCESSES}; >-$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker'); >-$max_processes ||= 1; >-my $mq_timeout = $ENV{MQ_TIMEOUT} // 10; >- >-my $not_found_retries = {}; >-my $max_retries = $ENV{MAX_RETRIES} || 10; >- >+my ( $max_processes, $help, @queues ); > GetOptions( > 'm|max-processes=i' => \$max_processes, > 'h|help' => \$help, > 'queue=s' => \@queues, > ) || pod2usage(1); >- >- > pod2usage(0) if $help; >- >-unless (@queues) { >- push @queues, 'default'; >-} >- >-my $conn; >-try { >- $conn = Koha::BackgroundJob->connect; >-} catch { >- warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; >-}; >- >-my $pm = Parallel::ForkManager->new($max_processes); >- >-if ( $conn ) { >- # FIXME cf note in Koha::BackgroundJob about $namespace >- my $namespace = C4::Context->config('memcached_namespace'); >- for my $queue (@queues) { >- $conn->subscribe( >- { >- destination => sprintf( "/queue/%s-%s", $namespace, $queue ), >- ack => 'client', >- 'prefetch-count' => 1, >- } >- ); >+push @queues, 'default' unless @queues; >+ >+Koha::BackgroundWorker->run( >+ { >+ max_processes => $max_processes // $ENV{MAX_PROCESSES}, >+ mq_timeout => $ENV{MQ_TIMEOUT}, >+ max_retries => $ENV{MAX_RETRIES}, >+ callback => \&process_job, >+ queues => \@queues, > } >-} >-while (1) { >- if ( $conn ) { >- my $frame = $conn->receive_frame( { timeout => $mq_timeout } ); >- if ( !defined $frame ) { >- # timeout or connection issue? >- $pm->reap_finished_children; >- next; # will reconnect automatically >- } >- >- my $args = try { >- my $body = $frame->body; >- decode_json($body); # TODO Should this be from_json? Check utf8 flag. >- } catch { >- Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_); >- return; >- }; >- >- unless ( $args ) { >- Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it"); >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- my $job = Koha::BackgroundJobs->find( $args->{job_id} ); >- >- if ( $job && $job->status ne 'new' ) { >- Koha::Logger->get( { interface => 'worker' } ) >- ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status ); >- >- # nack without requeue, we do not want to process this frame again >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- unless ($job) { >- $not_found_retries->{ $args->{job_id} } //= 0; >- if ( ++$not_found_retries->{ $args->{job_id} } >= $max_retries ) { >- Koha::Logger->get( { interface => 'worker' } ) >- ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} ); >- >- # nack without requeue, we do not want to process this frame again >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- Koha::Logger->get( { interface => 'worker' } ) >- ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} ); >- >- # nack to force requeue >- $conn->nack( { frame => $frame, requeue => 'true' } ); >- Time::HiRes::sleep(0.5); >- next; >- } >- $conn->ack( { frame => $frame } ); >- >- $pm->start and next; >- srand(); # ensure each child process begins with a new seed >- process_job( $job, $args ); >- $pm->finish; >- >- } else { >- my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); >- while ( my $job = $jobs->next ) { >- my $args = try { >- $job->json->decode($job->data); >- } catch { >- Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Cannot decode data for job id=%s", $job->id); >- $job->status('failed')->store; >- return; >- }; >- >- next unless $args; >- >- $pm->start and next; >- srand(); # ensure each child process begins with a new seed >- process_job( $job, { job_id => $job->id, %$args } ); >- $pm->finish; >- >- } >- sleep 10; >- } >-} >-$conn->disconnect; >-$pm->wait_all_children; >+); > > sub process_job { > my ( $job, $args ) = @_; >-- >2.30.2
You cannot view the attachment while viewing its details because your browser does not support IFRAMEs.
View the attachment on a separate page
.
View Attachment As Diff
View Attachment As Raw
Actions:
View
|
Diff
|
Splinter Review
Attachments on
bug 35920
:
161544
|
162951
|
162952