From 403083ee6e3951f4e93985f73a0d9147b2b1fdb9 Mon Sep 17 00:00:00 2001 From: Marcel de Rooy 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 . + +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