From 66600a6d8d649347fbf38a5a8b36d3af074c7f78 Mon Sep 17 00:00:00 2001 From: Tomas Cohen Arazi Date: Tue, 29 Apr 2025 15:32:39 -0300 Subject: [PATCH] [Poll-only proposal] Bug 35920: Move worker code to class Signed-off-by: Tomas Cohen Arazi --- Koha/BackgroundJob.pm | 77 +------------ Koha/Worker.pm | 122 ++++++++++++++++++++ misc/workers/background_jobs_worker.pl | 152 +++---------------------- 3 files changed, 140 insertions(+), 211 deletions(-) create mode 100644 Koha/Worker.pm diff --git a/Koha/BackgroundJob.pm b/Koha/BackgroundJob.pm index 154c00e1e8c..1deff3bb8c8 100644 --- a/Koha/BackgroundJob.pm +++ b/Koha/BackgroundJob.pm @@ -18,8 +18,7 @@ package Koha::BackgroundJob; use Modern::Perl; use Encode qw(); use JSON; -use Carp qw( croak ); -use Net::Stomp; +use Carp qw( croak ); use Try::Tiny qw( catch try ); use C4::Context; @@ -55,49 +54,6 @@ See also C for a full example =head2 Class methods -=head3 connect - -Connect to the message broker using default guest/guest credential - -=cut - -sub connect { - my ($self); - - my $notification_method = C4::Context->preference('JobsNotificationMethod') // 'STOMP'; - - return - unless $notification_method eq 'STOMP'; - - my $hostname = 'localhost'; - my $port = '61613'; - - my $config = C4::Context->config('message_broker'); - my $credentials = { - login => 'guest', - passcode => 'guest', - }; - if ($config) { - $hostname = $config->{hostname} if $config->{hostname}; - $port = $config->{port} if $config->{port}; - $credentials->{login} = $config->{username} if $config->{username}; - $credentials->{passcode} = $config->{password} if $config->{password}; - $credentials->{host} = $config->{vhost} if $config->{vhost}; - } - - my $stomp; - - try { - $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } ); - $stomp->connect($credentials); - } catch { - warn "Cannot connect to broker " . $_; - $stomp = undef; - }; - - return $stomp; -} - =head3 enqueue Enqueue a new job. It will insert a new row in the DB table and notify the broker that a new job has been enqueued. @@ -137,31 +93,6 @@ sub enqueue { } )->store; - $job_args->{job_id} = $self->id; - - my $conn = $self->connect; - return $self->id unless $conn; - - $json_args = $json->encode($job_args); - try { - # This namespace is wrong, it must be a vhost instead. - # But to do so it needs to be created on the server => much more work when a new Koha instance is created. - # Also, here we just want the Koha instance's name, but it's not in the config... - # Picking a random id (memcached_namespace) from the config - my $namespace = C4::Context->config('memcached_namespace'); - my $encoded_args = Encode::encode_utf8($json_args); # FIXME We should better leave this to Net::Stomp? - my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue ); - $conn->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } ) - or Koha::Exceptions::BackgroundJob->throw('Job has not been enqueued'); - } catch { - $self->status('failed')->store; - if ( ref($_) eq 'Koha::Exceptions::BackgroundJob' ) { - $_->rethrow; - } else { - warn sprintf "The job has not been sent to the message broker: (%s)", $_; - } - }; - return $self->id; } @@ -172,7 +103,7 @@ Process the job! =cut sub process { - my ( $self, $args ) = @_; + my ($self) = @_; return {} if ref($self) ne 'Koha::BackgroundJob'; @@ -184,8 +115,6 @@ sub process { my $derived_class = $self->_derived_class; - $args ||= {}; - if ( $self->context ) { my $context = $self->json->decode( $self->context ); C4::Context->interface( $context->{interface} ); @@ -202,7 +131,7 @@ sub process { Koha::Logger->get->warn( "A background job didn't have context defined (" . $self->id . ")" ); } - return $derived_class->process($args); + return $derived_class->process(); } =head3 start diff --git a/Koha/Worker.pm b/Koha/Worker.pm new file mode 100644 index 00000000000..408655c05f3 --- /dev/null +++ b/Koha/Worker.pm @@ -0,0 +1,122 @@ +package Koha::Worker; + +# 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 C4::Context; +use Koha::BackgroundJobs; +use Koha::Logger; + +use Parallel::ForkManager; +use Try::Tiny qw(catch try); + +=head1 NAME + +Koha::Worker - Koha bakground jobs worker class + +=head1 API + +=head2 Class methods + +=head3 new + + my $worker = Koha::Worker->new( + { + [ queues => [ 'default', 'long_tasks', ... ], ] + [ max_processes => N, ] + } + ); + +Constructor for the I class. + +=cut + +sub new { + my ( $class, $params ) = @_; + + my $max_processes = $params->{max_processes}; + $max_processes //= C4::Context->config('background_jobs_worker') // 1; + + my $queues = $params->{queues} // ['default']; + + my $self = { + max_processes => $max_processes, + queues => $queues, + }; + + bless $self, $class; + return $self; +} + +=head3 run + + $worker->run(); + +Method that triggers the main loop. + +=cut + +sub run { + my ($self) = @_; + + my $pm = Parallel::ForkManager->new( $self->{max_processes} ); + + # Main loop + while (1) { + + # FIXME: we should put a limit on the rows to featch each time. + my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => $self->{queues} } ); + while ( my $job = $jobs->next ) { + $pm->start and next; + srand(); # ensure each child process begins with a new seed + $self->process($job); + $pm->finish; + + } + $pm->reap_finished_children; + sleep 10; + } + + # Work is done + $pm->wait_all_children; +} + +=head3 process + + $self->process($job); + +Method that takes care of running the required job. + +=cut + +sub process { + my ( $self, $job ) = @_; + try { + Koha::Logger->get( { interface => 'worker' } )->info( sprintf "Started '%s' job id=%s", $job->type, $job->id ); + $job->process(); + Koha::Logger->get( { interface => 'worker' } ) + ->info( sprintf "Finished '%s' job id=%s status=%s", $job->type, $job->id, $job->status ); + } catch { + Koha::Logger->get( { interface => 'worker' } ) + ->warn( sprintf "Uncaught exception processing job id=%s: %s", $job->id, $_ ); + $job->status('failed')->store; + }; + + return; +} + +1; diff --git a/misc/workers/background_jobs_worker.pl b/misc/workers/background_jobs_worker.pl index f070c3cf642..7d3b29edba6 100755 --- a/misc/workers/background_jobs_worker.pl +++ b/misc/workers/background_jobs_worker.pl @@ -25,14 +25,17 @@ background_jobs_worker.pl - Worker script that will process background jobs =head1 DESCRIPTION -This script will connect to the Stomp server (RabbitMQ) and subscribe to the queues passed in parameter (or the 'default' queue), -or if a Stomp server is not active it will poll the database every 10s for new jobs in the passed queue. +This script will launch a worker for the specified queues. It will poll the database +every 10s for new jobs in the passed queue. -You can specify some queues only (using --queue, which is repeatable) if you want to run several workers that will handle their own jobs. +You can specify some queues only (using --queue, which is repeatable) if you want to +run several workers that will handle their own jobs. --m --max-processes specifies how many jobs to process simultaneously -Max processes will be set from the command line option, the environment variable MAX_PROCESSES, or the koha-conf file, in that order of precedence. +Max processes will be set from the command line option, the environment variable MAX_PROCESSES, +or the koha-conf file, in that order of precedence. + By default the script will only run one job at a time. =head1 OPTIONS @@ -53,17 +56,12 @@ 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 C4::Context; -use Koha::Logger; -use Koha::BackgroundJobs; -use C4::Context; +use Koha::Worker; $SIG{'PIPE'} = 'IGNORE'; # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu. @@ -73,10 +71,6 @@ 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; GetOptions( 'm|max-processes=i' => \$max_processes, @@ -90,125 +84,9 @@ 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, - } - ); - } -} -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; - - } - $pm->reap_finished_children; - sleep 10; - } -} -$conn->disconnect; -$pm->wait_all_children; - -sub process_job { - my ( $job, $args ) = @_; - try { - $job->process($args); - } catch { - Koha::Logger->get( { interface => 'worker' } ) - ->warn( sprintf "Uncaught exception processing job id=%s: %s", $job->id, $_ ); - $job->status('failed')->store; - }; -} +my $worker = Koha::Worker->new( { max_processes => $max_processes, queues => \@queues } ); + +# main loop +$worker->run(); + +1; -- 2.49.0