From 0d5aab8f98cb10c19759b4ce3d6ab03044a0be72 Mon Sep 17 00:00:00 2001 From: Jonathan Druart Date: Tue, 10 Jan 2023 17:05:00 +0100 Subject: [PATCH] Bug 32594: Improve the ES daemon indexer --- misc/background_jobs_worker.pl | 13 ++-- misc/search_tools/es_indexer_daemon.pl | 99 +++++++++++++------------- 2 files changed, 59 insertions(+), 53 deletions(-) diff --git a/misc/background_jobs_worker.pl b/misc/background_jobs_worker.pl index 6bbc82bc08e..7ae99659369 100755 --- a/misc/background_jobs_worker.pl +++ b/misc/background_jobs_worker.pl @@ -53,6 +53,7 @@ use Try::Tiny; use Pod::Usage; use Getopt::Long; +use C4::Context; use Koha::Logger; use Koha::BackgroundJobs; @@ -79,11 +80,13 @@ 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, - }); + $conn->subscribe( + { + destination => sprintf( "/queue/%s-%s", $namespace, $queue ), + ack => 'client', + 'prefetch-count' => 1, + } + ); } } while (1) { diff --git a/misc/search_tools/es_indexer_daemon.pl b/misc/search_tools/es_indexer_daemon.pl index a57fec8b510..68f52fcea9c 100755 --- a/misc/search_tools/es_indexer_daemon.pl +++ b/misc/search_tools/es_indexer_daemon.pl @@ -33,15 +33,18 @@ and process them in batches every second. use Modern::Perl; use JSON qw( decode_json ); -use Try::Tiny qw( catch try ); +use Try::Tiny; use Pod::Usage; use Getopt::Long; use C4::Context; +use Koha::Logger; use Koha::BackgroundJobs; use Koha::SearchEngine; use Koha::SearchEngine::Indexer; +use constant BATCH_SIZE => 10; # FIXME Must be in the koha config and/or options + my ( $help ); GetOptions( 'h|help' => \$help, @@ -61,76 +64,76 @@ try { 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' }); + $conn->subscribe( + { + destination => sprintf( "/queue/%s-%s", $namespace, 'elastic_index' ), + ack => 'client', + 'prefetch-count' => 1, + } + ); } my $indexer = Koha::SearchEngine::Indexer->new({ index => $Koha::SearchEngine::BIBLIOS_INDEX }); my @jobs = (); -my @records = (); -my $frame; while (1) { - local $SIG{ALRM} = sub { commit_records() if scalar @jobs }; # NB: \n required - alarm 1; + if ( $conn ) { - $frame = $conn->receive_frame; + my $frame = $conn->receive_frame; if ( !defined $frame ) { # maybe log connection problems next; # will reconnect automatically } - my $body = $frame->body; - my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. + my $job; + try { + my $body = $frame->body; + my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag. - # FIXME This means we need to have create the DB entry before - # It could work in a first step, but then we will want to handle job that will be created from the message received - my $job = Koha::BackgroundJobs->find($args->{job_id}); - next unless defined $job; + # FIXME ack early if it's what we decide on 32573 + $conn->ack( { frame => $frame } ); - push @records, @{ $args->{record_ids} }; - push @jobs, $job; + # FIXME This means we need to have create the DB entry before + # It could work in a first step, but then we will want to handle job that will be created from the message received + my $job = Koha::BackgroundJobs->find($args->{job_id}); + next unless defined $job; + push @jobs, $job; + if ( @jobs >= BATCH_SIZE || !$conn->can_read ) { + commit(@jobs); + @jobs = (); + } + } catch { + Koha::Logger->get->warn(sprintf "Job and/or frame not processed - %s", $_); + } finally { + $job->status('failed')->store if $job && @_; + }; } else { - my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => 'elastic_index' }); - while ( my $job = $jobs->next ) { - my $args = $job->json->decode($job->data); - push @records, @{ $args->{record_ids} }; - push @jobs, $job; - } + @jobs = Koha::BackgroundJobs->search( + { status => 'new', queue => 'elastic_index' } )->as_list; + commit(@jobs); + @jobs = (); sleep 10; } -} -$conn->disconnect; -sub commit_records { - eval { - $indexer->update_index(\@records); - }; - if ( $@ ){ - warn $@; - } - foreach my $job (@jobs){ - $job->status('finished')->store; - } - @jobs = (); - @records = (); - $conn->ack( { frame => $frame } ) if $frame; # FIXME depending on success? - # Acknowledging the current frame also acknowledges all previously batched frames } +$conn->disconnect; +sub commit { + my ( @jobs ) = @_; -sub process_job { - my ( $job, $args ) = @_; - - my $pid; - if ( $pid = fork ) { - wait; - return; + my @records; + for my $job ( @jobs ) { + my $args = $job->json->decode($job->data); + push @records, @{ $args->{record_ids} }; } - die "fork failed!" unless defined $pid; - - $job->process( $args ); + try { + $indexer->update_index(\@records); + } catch { + Koha::Logger->get->warn(sprintf "Update of elastic index failed with: %s", $_); + } - exit; + Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] } ) + ->update( { status => 'finished' }, { no_triggers => 1 } ); } -- 2.25.1