Bugzilla – Attachment 161544 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 code from workers
Bug-35920-Centralize-code-from-workers.patch (text/plain), 13.86 KB, created by
Jonathan Druart
on 2024-01-26 15:17:26 UTC
(
hide
)
Description:
Bug 35920: Centralize code from workers
Filename:
MIME Type:
Creator:
Jonathan Druart
Created:
2024-01-26 15:17:26 UTC
Size:
13.86 KB
patch
obsolete
>From a0e1dcf9d293df7103267f2da2d860433fab57a9 Mon Sep 17 00:00:00 2001 >From: Jonathan Druart <jonathan.druart@bugs.koha-community.org> >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<misc/background_jobs_worker.pl> 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 <http://www.gnu.org/licenses>. >+ >+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<job_size> is the size of the job >+C<job_args> 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
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