From 1a48cbe6bb8bcfd95e5e4a66193f8b5d18d92803 Mon Sep 17 00:00:00 2001 From: Marcel de Rooy Date: Fri, 8 Mar 2024 10:55:36 +0000 Subject: [PATCH] Bug 35920: Use BackgroundWorker in es_indexer_daemon Content-Type: text/plain; charset=utf-8 This needs incorporating the batch_size in the main loop. Test plan: Test both regular worker and es indexer daemon here. --- Koha/BackgroundWorker.pm | 17 +++- misc/workers/background_jobs_worker.pl | 15 +-- misc/workers/es_indexer_daemon.pl | 124 ++++--------------------- 3 files changed, 40 insertions(+), 116 deletions(-) diff --git a/Koha/BackgroundWorker.pm b/Koha/BackgroundWorker.pm index 760ab8fe2a..6e1c132c0e 100644 --- a/Koha/BackgroundWorker.pm +++ b/Koha/BackgroundWorker.pm @@ -50,6 +50,7 @@ Koha::BackgroundWorker - Centralized code for background worker scripts Main method to start the worker. Available parameters: + - batch_size - callback - queues - max_processes @@ -90,6 +91,8 @@ sub run { } ### The main loop + my @jobs; + my $lastseen = time; while (1) { if ($conn) { my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } ); @@ -147,10 +150,15 @@ sub run { } $conn->ack( { frame => $frame } ); - $pm->start and next; - srand(); # ensure each child process begins with a new seed - &$callback( $job, $args ); - $pm->finish; + push @jobs, [ $job, $args ]; + if ( @jobs >= $self->{batch_size} || time > $lastseen ) { + my @commits = @jobs; @jobs = (); + $lastseen = time; + $pm->start and next; + srand(); # ensure each child process begins with a new seed + &$callback(@commits); + $pm->finish; + } } else { my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } ); @@ -183,6 +191,7 @@ sub run { sub _init { my ($self) = @_; + $self->{batch_size} //= 1; $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker'); $self->{max_processes} ||= 1; diff --git a/misc/workers/background_jobs_worker.pl b/misc/workers/background_jobs_worker.pl index fd28dd2b1a..c731c9e199 100755 --- a/misc/workers/background_jobs_worker.pl +++ b/misc/workers/background_jobs_worker.pl @@ -82,10 +82,13 @@ Koha::BackgroundWorker->run( ); sub process_job { - my ( $job, $args ) = @_; - try { - $job->process( $args ); - } catch { - $job->status('failed')->store; - }; + my ( @jobs ) = @_; + foreach my $elt (@jobs) { + my ( $job, $args ) = @$elt; + try { + $job->process( $args ); + } catch { + $job->status('failed')->store; + }; + } } diff --git a/misc/workers/es_indexer_daemon.pl b/misc/workers/es_indexer_daemon.pl index 9e900b7195..6fbef67433 100755 --- a/misc/workers/es_indexer_daemon.pl +++ b/misc/workers/es_indexer_daemon.pl @@ -56,136 +56,48 @@ use Try::Tiny; use Pod::Usage; use Getopt::Long; use List::MoreUtils qw( natatime ); -use Time::HiRes; use C4::Context; -use Koha::Logger; use Koha::BackgroundJobs; +use Koha::Logger; use Koha::SearchEngine; use Koha::SearchEngine::Indexer; - -my ( $help, $batch_size ); - -my $not_found_retries = {}; -my $max_retries = $ENV{MAX_RETRIES} || 10; - +my ( $max_processes, $help, $batch_size ); GetOptions( - 'h|help' => \$help, - 'b|batch_size=s' => \$batch_size + 'm|max-processes=i' => \$max_processes, + 'h|help' => \$help, + 'b|batch_size=s' => \$batch_size ) || pod2usage(1); - pod2usage(0) if $help; - $batch_size //= 10; warn "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch'; - -my $logger = Koha::Logger->get({ interface => 'worker' }); - -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'); - $conn->subscribe( - { - destination => sprintf( "/queue/%s-%s", $namespace, 'elastic_index' ), - ack => 'client', - 'prefetch-count' => 1, - } - ); -} my $biblio_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::BIBLIOS_INDEX } ); my $auth_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::AUTHORITIES_INDEX } ); my $config = $biblio_indexer->get_elasticsearch_params; my $at_a_time = $config->{chunk_size} // 5000; - -my @jobs = (); - -while (1) { - - if ( $conn ) { - my $frame = $conn->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 $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 } ); - - push @jobs, $job; - if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) { - commit(@jobs); - @jobs = (); - } - - } else { - @jobs = Koha::BackgroundJobs->search( - { status => 'new', queue => 'elastic_index' } )->as_list; - commit(@jobs); - @jobs = (); - sleep 10; +my $lastseen = time; + +Koha::BackgroundWorker->run( + { + max_processes => $max_processes // $ENV{MAX_PROCESSES}, + mq_timeout => $ENV{MQ_TIMEOUT}, + max_retries => $ENV{MAX_RETRIES}, + callback => \&commit, + queues => [ 'elastic_index' ], } - -} -$conn->disconnect; +); sub commit { my (@jobs) = @_; + # Extract jobs only from passed arrayrefs + @jobs = map { $_->[0] } @jobs; my @bib_records; my @auth_records; + my $logger = Koha::Logger->get({ interface => 'worker' }); my $jobs = Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] }); # Start $jobs->update({ -- 2.30.2