From 5bab5242ceb9f013ea5f313b5256b6fddabb415c Mon Sep 17 00:00:00 2001 From: Jonathan Druart Date: Tue, 5 Jan 2021 14:42:20 +0100 Subject: [PATCH] Bug 27344: Implement Elastic's update_index_background using Koha::BackgroundJob This patch adds a background job submodule, UpdateElasticIndex, to deal with async ES index update (not the deletion). Using NYTProf (on a checkin): Without 618ms, executing 35676 statements and 26355 subroutine calls in 266 source files and 83 string evals. With 521ms, executing 13282 statements and 7979 subroutine calls in 195 source files and 26 string evals. However there are some problems with this patch: 1. We don't want *all* the index update to be in the background_jobs table (we could add a filter on the list view) 2. We don't track the "progress" of the job as we are sending all the records to Elastic. It is okish in my opinion but it must be noted. --- Koha/BackgroundJob.pm | 3 + Koha/BackgroundJob/UpdateElasticIndex.pm | 132 ++++++++++++++++++ Koha/SearchEngine/Elasticsearch/Indexer.pm | 52 ++++--- .../prog/en/modules/admin/background_jobs.tt | 16 +++ misc/background_jobs_worker.pl | 6 +- 5 files changed, 186 insertions(+), 23 deletions(-) create mode 100644 Koha/BackgroundJob/UpdateElasticIndex.pm diff --git a/Koha/BackgroundJob.pm b/Koha/BackgroundJob.pm index d375dfa0d5..0abf468b72 100644 --- a/Koha/BackgroundJob.pm +++ b/Koha/BackgroundJob.pm @@ -26,6 +26,7 @@ use Koha::DateUtils qw( dt_from_string ); use Koha::Exceptions; use Koha::BackgroundJob::BatchUpdateBiblio; use Koha::BackgroundJob::BatchUpdateAuthority; +use Koha::BackgroundJob::UpdateElasticIndex; use base qw( Koha::Object ); @@ -141,6 +142,8 @@ sub process { ? Koha::BackgroundJob::BatchUpdateBiblio->process($args) : $job_type eq 'batch_authority_record_modification' ? Koha::BackgroundJob::BatchUpdateAuthority->process($args) + : $job_type eq 'update_elastic_index' + ? Koha::BackgroundJob::UpdateElasticIndex->process($args) : Koha::Exceptions::Exception->throw('->process called without valid job_type'); } diff --git a/Koha/BackgroundJob/UpdateElasticIndex.pm b/Koha/BackgroundJob/UpdateElasticIndex.pm new file mode 100644 index 0000000000..6b98aefc9c --- /dev/null +++ b/Koha/BackgroundJob/UpdateElasticIndex.pm @@ -0,0 +1,132 @@ +package Koha::BackgroundJob::UpdateElasticIndex; + +# 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 JSON qw( encode_json decode_json ); + +use Koha::BackgroundJobs; +use Koha::DateUtils qw( dt_from_string ); +use C4::Biblio; +use C4::MarcModificationTemplates; + +use base 'Koha::BackgroundJob'; + +=head1 NAME + +Koha::BackgroundJob::UpdateElasticIndex - Update Elastic index + +This is a subclass of Koha::BackgroundJob. + +=head1 API + +=head2 Class methods + +=head3 job_type + +Define the job type of this job: update_elastic_index + +=cut + +sub job_type { + return 'update_elastic_index'; +} + +=head3 process + +Process the modification. + +=cut + +sub process { + my ( $self, $args ) = @_; + + my $job = Koha::BackgroundJobs->find( $args->{job_id} ); + + if ( !exists $args->{job_id} || !$job || $job->status eq 'cancelled' ) { + return; + } + + # FIXME If the job has already been started, but started again (worker has been restart for instance) + # Then we will start from scratch and so double process the same records + + my $job_progress = 0; + $job->started_on(dt_from_string) + ->progress($job_progress) + ->status('started') + ->store; + + my @record_ids = @{ $args->{record_ids} }; + my $record_server = $args->{record_server}; + + my $report = { + total_records => scalar @record_ids, + total_success => 0, + }; + + my @messages; + eval { + my $es_index = + $record_server eq "authorityserver" + ? $Koha::SearchEngine::AUTHORITIES_INDEX + : $Koha::SearchEngine::BIBLIOS_INDEX; + my $indexer = Koha::SearchEngine::Indexer->new({ index => $es_index }); + $indexer->update_index(\@record_ids); + }; + if ( $@ ) { + push @messages, { + type => 'error', + code => 'index_error', + error => $@, + + } + } else { + # FIXME This is not correct if some record_ids have been skipped + $report->{total_success} = scalar @record_ids; + } + + my $job_data = decode_json $job->data; + $job_data->{messages} = \@messages; + $job_data->{report} = $report; + + $job->ended_on(dt_from_string) + ->data(encode_json $job_data); + $job->status('finished') if $job->status ne 'cancelled'; + $job->store; +} + +=head3 enqueue + +Enqueue the new job + +=cut + +sub enqueue { + my ( $self, $args) = @_; + + return unless exists $args->{record_server}; + return unless exists $args->{record_ids}; + + my $record_server = $args->{record_server}; + my @record_ids = @{ $args->{record_ids} }; + + $self->SUPER::enqueue({ + job_size => scalar @record_ids, + job_args => {record_server => $record_server, record_ids => \@record_ids,} + }); +} + +1; diff --git a/Koha/SearchEngine/Elasticsearch/Indexer.pm b/Koha/SearchEngine/Elasticsearch/Indexer.pm index e5f6474fb3..6ec770a4fa 100644 --- a/Koha/SearchEngine/Elasticsearch/Indexer.pm +++ b/Koha/SearchEngine/Elasticsearch/Indexer.pm @@ -26,6 +26,7 @@ use Data::Dumper; use Koha::Exceptions; use Koha::SearchEngine::Zebra::Indexer; +use Koha::BackgroundJob::UpdateElasticIndex; use C4::AuthoritiesMarc qw//; use C4::Biblio; use C4::Context; @@ -97,12 +98,28 @@ Arrayref of Cs. =cut sub update_index { - my ($self, $biblionums, $records) = @_; + my ($self, $record_ids, $records) = @_; + + my $index_record_ids; + unless ( $records && @$records ) { + for my $record_id ( sort { $a <=> $b } @$record_ids ) { + + next unless $record_id; + + my $record = $self->_get_record( $record_id ); + if( $record ){ + push @$records, $record; + push @$index_record_ids, $record_id; + } + } + } else { + $index_record_ids = $record_ids; + } my $documents = $self->marc_records_to_documents($records); my @body; - for (my $i = 0; $i < scalar @$biblionums; $i++) { - my $id = $biblionums->[$i]; + for (my $i = 0; $i < scalar @$index_record_ids; $i++) { + my $id = $index_record_ids->[$i]; my $document = $documents->[$i]; push @body, { index => { @@ -264,7 +281,7 @@ sub update_mappings { $self->set_index_status_ok(); } -=head2 update_index_background($biblionums, $records) +=head2 update_index_background($record_numbers, $server) This has exactly the same API as C however it'll return immediately. It'll start a background process that does the adding. @@ -274,12 +291,10 @@ it to be updated by a regular index cron job in the future. =cut -# TODO implement in the future - I don't know the best way of doing this yet. -# If fork: make sure process group is changed so apache doesn't wait for us. - sub update_index_background { - my $self = shift; - $self->update_index(@_); + my ( $self, $record_numbers, $server ) = @_; + + Koha::BackgroundJob::UpdateElasticIndex->new->enqueue({ record_ids => $record_numbers, record_server => $server }); } =head2 index_records @@ -302,17 +317,10 @@ sub index_records { if ( $op eq 'specialUpdate' ) { my $index_record_numbers; if ($records){ - $index_record_numbers = $record_numbers; + $self->update_index( $record_numbers, $records ); } else { - foreach my $record_number ( @$record_numbers ){ - my $record = _get_record( $record_number, $server ); - if( $record ){ - push @$records, $record; - push @$index_record_numbers, $record_number; - } - } + $self->update_index_background( $record_numbers, $server ); } - $self->update_index_background( $index_record_numbers, $records ) if $index_record_numbers && $records; } elsif ( $op eq 'recordDelete' ) { $self->delete_index_background( $record_numbers ); @@ -322,10 +330,10 @@ sub index_records { } sub _get_record { - my ( $id, $server ) = @_; - return $server eq 'biblioserver' - ? C4::Biblio::GetMarcBiblio({ biblionumber => $id, embed_items => 1 }) - : C4::AuthoritiesMarc::GetAuthority($id); + my ( $self, $record_id ) = @_; + return $self->index eq $Koha::SearchEngine::BIBLIOS_INDEX + ? C4::Biblio::GetMarcBiblio({ biblionumber => $record_id, embed_items => 1 }) + : C4::AuthoritiesMarc::GetAuthority($record_id); } =head2 delete_index($biblionums) diff --git a/koha-tmpl/intranet-tmpl/prog/en/modules/admin/background_jobs.tt b/koha-tmpl/intranet-tmpl/prog/en/modules/admin/background_jobs.tt index 8731486026..20f5cc73e1 100644 --- a/koha-tmpl/intranet-tmpl/prog/en/modules/admin/background_jobs.tt +++ b/koha-tmpl/intranet-tmpl/prog/en/modules/admin/background_jobs.tt @@ -85,6 +85,20 @@ [% END %] [% END %] + [% CASE 'update_elastic_index' %] + [% SET report = job.report %] + [% IF report %] + [% IF report.total_records == 1 %] + [% IF report.total_success == 1 %] +
The records have successfully been reindexed!
+ [% END %] + [% ELSE %] +
+ [% report.total_success | html %] / [% report.total_records | html %] records have successfully been reindexed. Some errors occurred. + [% IF job.status == 'cancelled' %]The job has been cancelled before it finished.[% END %] +
+ [% END %] + [% END %] [% CASE %]Job type "[% job.type | html %]" not handled in the template [% END %] @@ -126,6 +140,7 @@ [% END %] [% END %] + [% CASE 'update_elastic_index' %] [% CASE %]Job type "[% job.type | html %]" not handled in the template [% END %] @@ -165,6 +180,7 @@ [% SWITCH job.type %] [% CASE 'batch_biblio_record_modification' %]Batch bibliographic record modification [% CASE 'batch_authority_record_modification' %]Batch authority record modification + [% CASE 'update_elastic_index' %]Update Elasticsearch index [% CASE %][% job.type | html %] [% END %] diff --git a/misc/background_jobs_worker.pl b/misc/background_jobs_worker.pl index 0520189fa5..681eed04c0 100755 --- a/misc/background_jobs_worker.pl +++ b/misc/background_jobs_worker.pl @@ -28,7 +28,11 @@ try { warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; }; -my @job_types = qw( batch_biblio_record_modification batch_authority_record_modification ); +my @job_types = qw( + batch_biblio_record_modification + batch_authority_record_modification + update_elastic_index +); if ( $conn ) { # FIXME cf note in Koha::BackgroundJob about $namespace -- 2.20.1