koha-dev.rot13.org Git - srvgit/blob - misc/search_tools/rebuild_elasticsearch.pl

   1 #!/usr/bin/perl
   2
   3 # This inserts records from a Koha database into elastic search
   4
   5 # Copyright 2014 Catalyst IT
   6 #
   7 # This file is part of Koha.
   8 #
   9 # Koha is free software; you can redistribute it and/or modify it
  10 # under the terms of the GNU General Public License as published by
  11 # the Free Software Foundation; either version 3 of the License, or
  12 # (at your option) any later version.
  13 #
  14 # Koha is distributed in the hope that it will be useful, but
  15 # WITHOUT ANY WARRANTY; without even the implied warranty of
  16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17 # GNU General Public License for more details.
  18 #
  19 # You should have received a copy of the GNU General Public License
  20 # along with Koha; if not, see <http://www.gnu.org/licenses>.
  21
  22 =head1 NAME
  23
  24 rebuild_elasticsearch.pl - inserts records from a Koha database into Elasticsearch
  25
  26 =head1 SYNOPSIS
  27
  28 B<rebuild_elasticsearch.pl>
  29 [B<-c|--commit>=C<count>]
  30 [B<-d|--delete>]
  31 [B<-r|--reset>]
  32 [B<-a|--authorities>]
  33 [B<-b|--biblios>]
  34 [B<--desc>]
  35 [B<-bn|--bnumber>]
  36 [B<-ai|--authid>]
  37 [B<-p|--processes>]
  38 [B<-v|--verbose>]
  39 [B<-h|--help>]
  40 [B<--man>]
  41
  42 =head1 DESCRIPTION
  43
  44 Inserts records from a Koha database into Elasticsearch.
  45
  46 =head1 OPTIONS
  47
  48 =over
  49
  50 =item B<-c|--commit>=C<count>
  51
  52 Specify how many records will be batched up before they're added to Elasticsearch.
  53 Higher should be faster, but will cause more RAM usage. Default is 5000.
  54
  55 =item B<-d|--delete>
  56
  57 Delete the index and recreate it before indexing.
  58
  59 =item B<-r|--reset>
  60
  61 Reload mappings from files (specified in koha-conf.xml) before indexing.
  62 Implies --delete.
  63
  64 =item B<-a|--authorities>
  65
  66 Index the authorities only. Combining this with B<-b> is the same as
  67 specifying neither and so both get indexed.
  68
  69 =item B<-b|--biblios>
  70
  71 Index the biblios only. Combining this with B<-a> is the same as
  72 specifying neither and so both get indexed.
  73
  74 =item B<--desc>
  75
  76 Index the records in descending id order. Intended to index newer record before older records.
  77 Default is to index in ascending order.
  78 Does not work with --bnumber or --authid
  79
  80 =item B<-bn|--bnumber>
  81
  82 Only index the supplied biblionumber, mostly for testing purposes. May be
  83 repeated.
  84
  85 =item B<-ai|--authid>
  86
  87 Only index the supplied authority id, mostly for testing purposes. May be
  88 repeated.
  89
  90 =item B<-p|--processes>
  91
  92 Number of processes to use for indexing. This can be used to do more indexing
  93 work in parallel on multicore systems. By default, a single process is used.
  94
  95 =item B<-v|--verbose>
  96
  97 By default, this program only emits warnings and errors. This makes it talk
  98 more. Add more to make it even more wordy, in particular when debugging.
  99
 100 =item B<-h|--help>
 101
 102 Help!
 103
 104 =item B<--man>
 105
 106 Full documentation.
 107
 108 =back
 109
 110 =head1 IMPLEMENTATION
 111
 112 =cut
 113
 114 use autodie;
 115 use Getopt::Long qw( GetOptions );
 116 use Koha::Script;
 117 use C4::Context;
 118 use Koha::MetadataRecord::Authority;
 119 use Koha::BiblioUtils;
 120 use Koha::SearchEngine::Elasticsearch;
 121 use Koha::SearchEngine::Elasticsearch::Indexer;
 122 use MARC::Field;
 123 use Modern::Perl;
 124 use Pod::Usage qw( pod2usage );
 125 use Try::Tiny qw( catch try );
 126
 127 my $verbose = 0;
 128 my $commit = 5000;
 129 my ($delete, $reset, $help, $man, $processes);
 130 my ($index_biblios, $index_authorities);
 131 my (@biblionumbers,@authids);
 132 my $desc;
 133
 134 $|=1; # flushes output
 135
 136 GetOptions(
 137     'c|commit=i'    => \$commit,
 138     'd|delete'      => \$delete,
 139     'r|reset'       => \$reset,
 140     'a|authorities' => \$index_authorities,
 141     'b|biblios'     => \$index_biblios,
 142     'desc'          => \$desc,
 143     'bn|bnumber=i'  => \@biblionumbers,
 144     'ai|authid=i'   => \@authids,
 145     'p|processes=i' => \$processes,
 146     'v|verbose+'    => \$verbose,
 147     'h|help'        => \$help,
 148     'man'           => \$man,
 149 );
 150
 151 # Default is to do both
 152 unless ($index_authorities || $index_biblios) {
 153     $index_authorities = $index_biblios = 1;
 154 }
 155
 156 if ($processes && ( @biblionumbers || @authids) ) {
 157     die "Argument p|processes cannot be combined with bn|bnumber or ai|authid";
 158 }
 159
 160 pod2usage(1) if $help;
 161 pod2usage( -exitstatus => 0, -verbose => 2 ) if $man;
 162
 163 _sanity_check();
 164
 165 if ($reset){
 166     Koha::SearchEngine::Elasticsearch->reset_elasticsearch_mappings;
 167     $delete = 1;
 168 }
 169
 170 _verify_index_state($Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX, $delete) if ($index_biblios);
 171 _verify_index_state($Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX, $delete) if ($index_authorities);
 172
 173 my $slice_index = 0;
 174 my $slice_count = ( $processes //= 1 );
 175 my %iterator_options;
 176
 177 if ($slice_count > 1) {
 178     # Fire up child processes for processing slices from 2 on. This main process will handle slice 1.
 179     $slice_index = 0;
 180     for (my $proc = 1; $proc < $slice_count; $proc++) {
 181         my $pid = fork();
 182         die "Failed to fork a child process\n" unless defined $pid;
 183         if ($pid == 0) {
 184             # Child process, give it a slice to process
 185             $slice_index = $proc;
 186             last;
 187         }
 188     }
 189     # Fudge the commit count a bit to spread out the Elasticsearch commits
 190     $commit *= 1 + 0.10 * $slice_index;
 191     $commit = int( $commit );
 192     _log(1, "Processing slice @{[$slice_index + 1]} of $slice_count\n");
 193     $iterator_options{slice} = { index => $slice_index, count => $slice_count };
 194 }
 195
 196 if( $desc ){
 197     $iterator_options{desc} = 1;
 198 }
 199
 200 my $next;
 201 if ($index_biblios) {
 202     _log(1, "Indexing biblios\n");
 203     if (@biblionumbers) {
 204         $next = sub {
 205             my $r = shift @biblionumbers;
 206             return () unless defined $r;
 207             return ($r, Koha::BiblioUtils->get_from_biblionumber($r, item_data => 1 ));
 208         };
 209     } else {
 210         my $records = Koha::BiblioUtils->get_all_biblios_iterator(%iterator_options);
 211         $next = sub {
 212             $records->next();
 213         }
 214     }
 215     _do_reindex($next, $Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX);
 216 }
 217 if ($index_authorities) {
 218     _log(1, "Indexing authorities\n");
 219     if (@authids) {
 220         $next = sub {
 221             my $r = shift @authids;
 222             return () unless defined $r;
 223             my $a = Koha::MetadataRecord::Authority->get_from_authid($r);
 224             return ($r, $a);
 225         };
 226     } else {
 227         my $records = Koha::MetadataRecord::Authority->get_all_authorities_iterator(%iterator_options);
 228         $next = sub {
 229             $records->next();
 230         }
 231     }
 232     _do_reindex($next, $Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX);
 233 }
 234
 235 if ($slice_index == 0) {
 236     # Main process, wait for children
 237     for (my $proc = 1; $proc < $processes; $proc++) {
 238         wait();
 239     }
 240 }
 241
 242 =head1 INTERNAL METHODS
 243
 244 =head2 _verify_index_state
 245
 246     _verify_index_state($Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX, 1);
 247
 248 Checks the index state and recreates it if requested.
 249
 250 =cut
 251
 252 sub _verify_index_state {
 253     my ( $index_name, $recreate ) = @_;
 254
 255     _log(1, "Checking state of $index_name index\n");
 256     my $indexer = Koha::SearchEngine::Elasticsearch::Indexer->new( { index => $index_name } );
 257
 258     if ($recreate) {
 259         _log(1, "Dropping and recreating $index_name index\n");
 260         $indexer->drop_index() if $indexer->index_exists();
 261         $indexer->create_index();
 262     }
 263     elsif (!$indexer->index_exists) {
 264         # Create index if does not exist
 265         $indexer->create_index();
 266     } elsif ($indexer->is_index_status_ok) {
 267         # Update mapping unless index is some kind of problematic state
 268         $indexer->update_mappings();
 269     } elsif ($indexer->is_index_status_recreate_required) {
 270         warn qq/Index "$index_name" has status "recreate required", suggesting it should be recreated/;
 271     }
 272 }
 273
 274 =head2 _do_reindex
 275
 276     _do_reindex($callback, $Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX);
 277
 278 Does the actual reindexing. $callback is a function that always returns the next record.
 279 For each index we iterate through the records, committing at specified count
 280
 281 =cut
 282
 283 sub _do_reindex {
 284     my ( $next, $index_name ) = @_;
 285
 286     my $indexer = Koha::SearchEngine::Elasticsearch::Indexer->new( { index => $index_name } );
 287
 288     my $count        = 0;
 289     my $commit_count = $commit;
 290     my ( @id_buffer, @commit_buffer );
 291     while ( my $record = $next->() ) {
 292         my $id     = $record->id // $record->authid;
 293         my $record = $record->record;
 294         $count++;
 295         if ( $verbose == 1 ) {
 296             _log( 1, "$count records processed\n" ) if ( $count % 1000 == 0);
 297         } else {
 298             _log( 2, "$id\n" );
 299         }
 300
 301         push @id_buffer,     $id;
 302         push @commit_buffer, $record;
 303         if ( !( --$commit_count ) ) {
 304             _log( 1, "Committing $commit records...\n" );
 305             my $response;
 306             try{
 307                 $response = $indexer->update_index( \@id_buffer, \@commit_buffer );
 308                 _handle_response($response);
 309                 _log( 1, "Commit complete\n" );
 310             } catch {
 311                 _log(1,"Elasticsearch exception thrown: ".$_->type."\n");
 312                 _log(2,"Details: ".$_->details."\n");
 313             };
 314             $commit_count  = $commit;
 315             @id_buffer     = ();
 316             @commit_buffer = ();
 317         }
 318     }
 319
 320     # There are probably uncommitted records
 321     _log( 1, "Committing final records...\n" );
 322     my $response = $indexer->update_index( \@id_buffer, \@commit_buffer );
 323     _handle_response($response);
 324     _log( 1, "Total $count records indexed\n" );
 325 }
 326
 327 =head2 _sanity_check
 328
 329     _sanity_check();
 330
 331 Checks some basic stuff to ensure that it's sane before we start.
 332
 333 =cut
 334
 335 sub _sanity_check {
 336     # Do we have an elasticsearch block defined?
 337     my $conf = C4::Context->config('elasticsearch');
 338     die "No 'elasticsearch' block is defined in koha-conf.xml.\n" if ( !$conf );
 339 }
 340
 341 =head2 _handle_response
 342
 343 Parse the return from update_index and display errors depending on verbosity of the script
 344
 345 =cut
 346
 347 sub _handle_response {
 348     my ($response) = @_;
 349     if( $response->{errors} eq 'true' ){
 350         _log( 1, "There were errors during indexing\n" );
 351         if ( $verbose > 1 ){
 352             foreach my $item (@{$response->{items}}){
 353                 next unless defined $item->{index}->{error};
 354                 print "Record #" . $item->{index}->{_id} . " " .
 355                       $item->{index}->{error}->{reason} . " (" . $item->{index}->{error}->{type} . ") : " .
 356                       $item->{index}->{error}->{caused_by}->{type} . " (" . $item->{index}->{error}->{caused_by}->{reason} . ")\n";
 357             }
 358         }
 359     }
 360 }
 361
 362 =head2 _log
 363
 364     _log($level, "Message\n");
 365
 366 Output progress information.
 367
 368 Will output the message if verbosity level is set to $level or more. Will not
 369 include a trailing newline automatically.
 370
 371 =cut
 372
 373 sub _log {
 374     my ($level, $msg) = @_;
 375
 376     print "[$$] $msg" if ($verbose >= $level);
 377 }