Bugzilla – Attachment 181684 Details for
Bug 35920
Centralize code from workers
Home
|
New
|
Browse
|
Search
|
[?]
|
Reports
|
Help
|
New Account
|
Log In
[x]
|
Forgot Password
Login:
[x]
[patch]
[Poll-only proposal] Bug 35920: Move worker code to class
Poll-only-proposal-Bug-35920-Move-worker-code-to-c.patch (text/plain), 13.91 KB, created by
Tomás Cohen Arazi (tcohen)
on 2025-04-29 18:33:08 UTC
(
hide
)
Description:
[Poll-only proposal] Bug 35920: Move worker code to class
Filename:
MIME Type:
Creator:
Tomás Cohen Arazi (tcohen)
Created:
2025-04-29 18:33:08 UTC
Size:
13.91 KB
patch
obsolete
>From 66600a6d8d649347fbf38a5a8b36d3af074c7f78 Mon Sep 17 00:00:00 2001 >From: Tomas Cohen Arazi <tomascohen@theke.io> >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 <tomascohen@theke.io> >--- > 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<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 $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 <http://www.gnu.org/licenses>. >+ >+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<Koha::Worker> 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
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
| 181684