From a0e1dcf9d293df7103267f2da2d860433fab57a9 Mon Sep 17 00:00:00 2001 From: Jonathan Druart Date: Fri, 26 Jan 2024 16:13:16 +0100 Subject: [PATCH] Bug 35920: Centralize code from workers --- Koha/BackgroundJob.pm | 52 +------- Koha/BackgroundJob/Broker.pm | 178 +++++++++++++++++++++++++ misc/workers/background_jobs_worker.pl | 77 +++-------- 3 files changed, 199 insertions(+), 108 deletions(-) create mode 100644 Koha/BackgroundJob/Broker.pm diff --git a/Koha/BackgroundJob.pm b/Koha/BackgroundJob.pm index 7c2f735ada4..f5b7c1b5832 100644 --- a/Koha/BackgroundJob.pm +++ b/Koha/BackgroundJob.pm @@ -23,6 +23,7 @@ use Net::Stomp; use Try::Tiny qw( catch try ); use C4::Context; +use Koha::BackgroundJob::Broker; use Koha::DateUtils qw( dt_from_string ); use Koha::Exceptions; use Koha::Exceptions::BackgroundJob; @@ -54,34 +55,7 @@ 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 $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 = Net::Stomp->new( { hostname => $hostname, port => $port } ); - $stomp->connect( $credentials ); - return $stomp; -} - -=head3 enqueue +=head3 enqueue_job 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. @@ -99,7 +73,7 @@ sub enqueue { my $job_size = $params->{job_size}; my $job_args = $params->{job_args}; my $job_context = $params->{job_context} // C4::Context->userenv; - my $job_queue = $params->{job_queue} // 'default'; + my $job_queue = $params->{job_queue} // 'default'; my $json = $self->json; my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef; @@ -121,26 +95,10 @@ sub enqueue { )->store; $job_args->{job_id} = $self->id; - - my $conn; - try { - $conn = $self->connect; - } catch { - warn "Cannot connect to broker " . $_; - }; - 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::Exception->throw('Job has not been enqueued'); + Koha::BackgroundJob::Broker->new->connect->enqueue_job({queue => $job_queue, body => $json_args}); } catch { $self->status('failed')->store; if ( ref($_) eq 'Koha::Exceptions::Exception' ) { diff --git a/Koha/BackgroundJob/Broker.pm b/Koha/BackgroundJob/Broker.pm new file mode 100644 index 00000000000..a40f4131d5d --- /dev/null +++ b/Koha/BackgroundJob/Broker.pm @@ -0,0 +1,178 @@ +package Koha::BackgroundJob::Broker; + +# 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 Encode qw(); +use JSON; +use Net::Stomp; +use Try::Tiny qw( catch try ); +use Time::HiRes; + +use C4::Context; + +my $not_found_retries = {}; +my $max_retries = $ENV{MAX_RETRIES} || 10; + +=head1 NAME + +Koha::BackgroundJob::Broker - Class to connect to RabbitMQ + +=head1 API + +=head2 Class methods + +=cut + +sub new { + my $class = shift; + my $self = {}; + return bless $self, $class; +} + +=head3 connect + +Connect to the message broker using default guest/guest credential + +=cut + +sub connect { + my ($self) = @_; + 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 = Net::Stomp->new( { hostname => $hostname, port => $port } ); + $stomp->connect($credentials); + $self->{conn} = $stomp; + return $self; +} + +sub subscribe { + my ( $self, @queues ) = @_; + + # FIXME cf note in Koha::BackgroundJob about $namespace + my $namespace = C4::Context->config('memcached_namespace'); + for my $queue (@queues) { + $self->{conn}->subscribe( + { + destination => sprintf( "/queue/%s-%s", $namespace, $queue ), + ack => 'client', + 'prefetch-count' => 1, + } + ); + } +} + +sub receive_frame { + my ($self) = @_; + return $self->{conn}->receive_frame; +} + +=head3 enqueue_job + +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. + +C is the size of the job +C is the arguments of the job. It's a structure that will be JSON encoded. + +Return the job_id of the newly created job. + +=cut + +sub enqueue_job { + my ( $self, $params ) = @_; + + my $job_queue = $params->{job_queue} // 'default'; + my $body = $params->{body}; + my $conn; + try { + $conn = $self->connect; + } catch { + warn "Cannot connect to broker " . $_; + }; + return unless $conn; + + # 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($body); # FIXME We should better leave this to Net::Stomp? + my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue ); + $self->{conn}->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } ) + or Koha::Exceptions::Exception->throw('Job has not been enqueued'); +} + +sub decode_frame { + my ( $self, $frame ) = @_; + + return 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; + }; +} + +sub get_job { + my ( $self, $params ) = @_; + + my $job_id = $params->{job_id}; + my $frame = $params->{frame}; + + my $job = Koha::BackgroundJobs->find($job_id); + + if ( $job && $job->status ne 'new' ) { + Koha::Logger->get( { interface => 'worker' } ) + ->warn( sprintf "Job %s has wrong status %s", $job_id, $job->status ); + + # nack without requeue, we do not want to process this frame again + $self->{conn}->nack( { frame => $frame, requeue => 'false' } ); + return; + } + + unless ($job) { + if ( ++$not_found_retries->{$job_id} >= $max_retries ) { + Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Job %s not found, no more retry", $job_id ); + + # nack without requeue, we do not want to process this frame again + $self->{conn}->nack( { frame => $frame, requeue => 'false' } ); + return; + } + + Koha::Logger->get( { interface => 'worker' } )->debug( sprintf "Job %s not found, will retry later", $job_id ); + + # nack to force requeue + $self->{conn}->nack( { frame => $frame, requeue => 'true' } ); + Time::HiRes::sleep(0.5); + return; + } + +} + +1; diff --git a/misc/workers/background_jobs_worker.pl b/misc/workers/background_jobs_worker.pl index 94c458268c3..ded4bce9ca0 100755 --- a/misc/workers/background_jobs_worker.pl +++ b/misc/workers/background_jobs_worker.pl @@ -58,12 +58,11 @@ use Try::Tiny; use Pod::Usage; use Getopt::Long; use Parallel::ForkManager; -use Time::HiRes; use C4::Context; use Koha::Logger; +use Koha::BackgroundJob::Broker; use Koha::BackgroundJobs; -use C4::Context; $SIG{'PIPE'} = 'IGNORE'; # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu. @@ -73,9 +72,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 $not_found_retries = {}; -my $max_retries = $ENV{MAX_RETRIES} || 10; - GetOptions( 'm|max-processes=i' => \$max_processes, 'h|help' => \$help, @@ -89,80 +85,39 @@ unless (@queues) { push @queues, 'default'; } -my $conn; +my $rabbit; try { - $conn = Koha::BackgroundJob->connect; + $rabbit = Koha::BackgroundJob::Broker->new->connect; + $rabbit->subscribe(@queues); } 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; + if ( $rabbit->{conn} ) { + my $frame = $rabbit->receive_frame; + if ( !defined $frame ) { + # maybe log connection problems 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 $args = $rabbit->decode_frame($frame); - my $job = Koha::BackgroundJobs->find( $args->{job_id} ); - - if ( $job && $job->status ne 'new' ) { + unless ($args) { 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' } ); + ->warn( sprintf "Frame does not have correct args, ignoring it" ); + $rabbit->{conn}->nack( { frame => $frame, requeue => 'false' } ); next; } - unless ($job) { - 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} ); + my $job = $rabbit->get_job( { job_id => $args->{job_id}, frame => $frame } ); + next unless $job; - # 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 } ); + $rabbit->{conn}->ack( { frame => $frame } ); $pm->start and next; srand(); # ensure each child process begins with a new seed @@ -191,7 +146,7 @@ while (1) { sleep 10; } } -$conn->disconnect; +$rabbit->{conn}->disconnect; $pm->wait_all_children; sub process_job { -- 2.34.1