Koha/SearchEngine/Elasticsearch/Indexer.pm

   1 package Koha::SearchEngine::Elasticsearch::Indexer;
   2
   3 # Copyright 2013 Catalyst IT
   4 #
   5 # This file is part of Koha.
   6 #
   7 # Koha is free software; you can redistribute it and/or modify it
   8 # under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # Koha is distributed in the hope that it will be useful, but
  13 # WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with Koha; if not, see <http://www.gnu.org/licenses>.
  19
  20 use Carp;
  21 use Modern::Perl;
  22 use Try::Tiny;
  23 use List::Util qw(any);
  24 use base qw(Koha::SearchEngine::Elasticsearch);
  25 use Data::Dumper;
  26
  27 use Koha::Exceptions;
  28 use Koha::SearchEngine::Zebra::Indexer;
  29 use C4::AuthoritiesMarc qw//;
  30 use C4::Biblio;
  31 use C4::Context;
  32
  33 =head1 NAME
  34
  35 Koha::SearchEngine::Elasticsearch::Indexer - handles adding new records to the index
  36
  37 =head1 SYNOPSIS
  38
  39     my $indexer = Koha::SearchEngine::Elasticsearch::Indexer->new(
  40         { index => Koha::SearchEngine::BIBLIOS_INDEX } );
  41     $indexer->drop_index();
  42     $indexer->update_index(\@biblionumbers, \@records);
  43
  44
  45 =head1 CONSTANTS
  46
  47 =over 4
  48
  49 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_OK>
  50
  51 Represents an index state where index is created and in a working state.
  52
  53 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_REINDEX_REQUIRED>
  54
  55 Not currently used, but could be useful later, for example if can detect when new field or mapping added.
  56
  57 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_RECREATE_REQUIRED>
  58
  59 Representings an index state where index needs to be recreated and is not in a working state.
  60
  61 =back
  62
  63 =cut
  64
  65 use constant {
  66     INDEX_STATUS_OK => 0,
  67     INDEX_STATUS_REINDEX_REQUIRED => 1,
  68     INDEX_STATUS_RECREATE_REQUIRED => 2,
  69 };
  70
  71 =head1 FUNCTIONS
  72
  73 =head2 update_index($biblionums, $records)
  74
  75     try {
  76         $self->update_index($biblionums, $records);
  77     } catch {
  78         die("Something went wrong trying to update index:" .  $_[0]);
  79     }
  80
  81 Converts C<MARC::Records> C<$records> to Elasticsearch documents and performs
  82 an update request for these records on the Elasticsearch index.
  83
  84 =over 4
  85
  86 =item C<$biblionums>
  87
  88 Arrayref of biblio numbers for the C<$records>, the order must be the same as
  89 and match up with C<$records>.
  90
  91 =item C<$records>
  92
  93 Arrayref of C<MARC::Record>s.
  94
  95 =back
  96
  97 =cut
  98
  99 sub update_index {
 100     my ($self, $biblionums, $records) = @_;
 101
 102     my $documents = $self->marc_records_to_documents($records);
 103     my @body;
 104     for (my $i = 0; $i < scalar @$biblionums; $i++) {
 105         my $id = $biblionums->[$i];
 106         my $document = $documents->[$i];
 107         push @body, {
 108             index => {
 109                 _id => "$id"
 110             }
 111         };
 112         push @body, $document;
 113     }
 114     my $response;
 115     if (@body) {
 116         my $elasticsearch = $self->get_elasticsearch();
 117         $response = $elasticsearch->bulk(
 118             index => $self->index_name,
 119             type => 'data', # is just hard coded in Indexer.pm?
 120             body => \@body
 121         );
 122         if ($response->{errors}) {
 123             carp "One or more ElasticSearch errors occurred when indexing documents";
 124         }
 125     }
 126     return $response;
 127 }
 128
 129 =head2 set_index_status_ok
 130
 131 Convenience method for setting index status to C<INDEX_STATUS_OK>.
 132
 133 =cut
 134
 135 sub set_index_status_ok {
 136     my ($self) = @_;
 137     $self->index_status(INDEX_STATUS_OK);
 138 }
 139
 140 =head2 is_index_status_ok
 141
 142 Convenience method for checking if index status is C<INDEX_STATUS_OK>.
 143
 144 =cut
 145
 146 sub is_index_status_ok {
 147     my ($self) = @_;
 148     return $self->index_status == INDEX_STATUS_OK;
 149 }
 150
 151 =head2 set_index_status_reindex_required
 152
 153 Convenience method for setting index status to C<INDEX_REINDEX_REQUIRED>.
 154
 155 =cut
 156
 157 sub set_index_status_reindex_required {
 158     my ($self) = @_;
 159     $self->index_status(INDEX_STATUS_REINDEX_REQUIRED);
 160 }
 161
 162 =head2 is_index_status_reindex_required
 163
 164 Convenience method for checking if index status is C<INDEX_STATUS_REINDEX_REQUIRED>.
 165
 166 =cut
 167
 168 sub is_index_status_reindex_required {
 169     my ($self) = @_;
 170     return $self->index_status == INDEX_STATUS_REINDEX_REQUIRED;
 171 }
 172
 173 =head2 set_index_status_recreate_required
 174
 175 Convenience method for setting index status to C<INDEX_STATUS_RECREATE_REQUIRED>.
 176
 177 =cut
 178
 179 sub set_index_status_recreate_required {
 180     my ($self) = @_;
 181     $self->index_status(INDEX_STATUS_RECREATE_REQUIRED);
 182 }
 183
 184 =head2 is_index_status_recreate_required
 185
 186 Convenience method for checking if index status is C<INDEX_STATUS_RECREATE_REQUIRED>.
 187
 188 =cut
 189
 190 sub is_index_status_recreate_required {
 191     my ($self) = @_;
 192     return $self->index_status == INDEX_STATUS_RECREATE_REQUIRED;
 193 }
 194
 195 =head2 index_status($status)
 196
 197 Will either set the current index status to C<$status> and return C<$status>,
 198 or return the current index status if called with no arguments.
 199
 200 =over 4
 201
 202 =item C<$status>
 203
 204 Optional argument. If passed will set current index status to C<$status> if C<$status> is
 205 a valid status. See L</CONSTANTS>.
 206
 207 =back
 208
 209 =cut
 210
 211 sub index_status {
 212     my ($self, $status) = @_;
 213     my $key = 'ElasticsearchIndexStatus_' . $self->index;
 214
 215     if (defined $status) {
 216         unless (any { $status == $_ } (
 217                 INDEX_STATUS_OK,
 218                 INDEX_STATUS_REINDEX_REQUIRED,
 219                 INDEX_STATUS_RECREATE_REQUIRED,
 220             )
 221         ) {
 222             Koha::Exceptions::Exception->throw("Invalid index status: $status");
 223         }
 224         C4::Context->set_preference($key, $status);
 225         return $status;
 226     }
 227     else {
 228         return C4::Context->preference($key);
 229     }
 230 }
 231
 232 =head2 update_mappings
 233
 234 Generate Elasticsearch mappings from mappings stored in database and
 235 perform a request to update Elasticsearch index mappings. Will throw an
 236 error and set index status to C<INDEX_STATUS_RECREATE_REQUIRED> if update
 237 failes.
 238
 239 =cut
 240
 241 sub update_mappings {
 242     my ($self) = @_;
 243     my $elasticsearch = $self->get_elasticsearch();
 244     my $mappings = $self->get_elasticsearch_mappings();
 245
 246     foreach my $type (keys %{$mappings}) {
 247         try {
 248             my $response = $elasticsearch->indices->put_mapping(
 249                 index => $self->index_name,
 250                 type => $type,
 251                 body => {
 252                     $type => $mappings->{$type}
 253                 }
 254             );
 255         } catch {
 256             $self->set_index_status_recreate_required();
 257             my $reason = $_[0]->{vars}->{body}->{error}->{reason};
 258             my $index_name = $self->index_name;
 259             Koha::Exceptions::Exception->throw(
 260                 error => "Unable to update mappings for index \"$index_name\". Reason was: \"$reason\". Index needs to be recreated and reindexed",
 261             );
 262         };
 263     }
 264     $self->set_index_status_ok();
 265 }
 266
 267 =head2 update_index_background($biblionums, $records)
 268
 269 This has exactly the same API as C<update_index> however it'll
 270 return immediately. It'll start a background process that does the adding.
 271
 272 If it fails to add to Elasticsearch then it'll add to a queue that will cause
 273 it to be updated by a regular index cron job in the future.
 274
 275 =cut
 276
 277 # TODO implement in the future - I don't know the best way of doing this yet.
 278 # If fork: make sure process group is changed so apache doesn't wait for us.
 279
 280 sub update_index_background {
 281     my $self = shift;
 282     $self->update_index(@_);
 283 }
 284
 285 =head2 index_records
 286
 287 This function takes an array of record numbers and fetches the records to send to update_index
 288 for actual indexing.
 289
 290 If $records parameter is provided the records will be used as-is, this is only utilized for authorities
 291 at the moment.
 292
 293 The other variables are used for parity with Zebra indexing calls. Currently the calls are passed through
 294 to Zebra as well.
 295
 296 =cut
 297
 298 sub index_records {
 299     my ( $self, $record_numbers, $op, $server, $records ) = @_;
 300     $record_numbers = [$record_numbers] if ref $record_numbers ne 'ARRAY' && defined $record_numbers;
 301     $records = [$records] if ref $records ne 'ARRAY' && defined $records;
 302     if ( $op eq 'specialUpdate' ) {
 303         my $index_record_numbers;
 304         if ($records){
 305             $index_record_numbers = $record_numbers;
 306         } else {
 307             foreach my $record_number ( @$record_numbers ){
 308                 my $record = _get_record( $record_number, $server );
 309                 if( $record ){
 310                     push @$records, $record;
 311                     push @$index_record_numbers, $record_number;
 312                 }
 313             }
 314         }
 315         $self->update_index_background( $index_record_numbers, $records ) if $index_record_numbers && $records;
 316     }
 317     elsif ( $op eq 'recordDelete' ) {
 318         $self->delete_index_background( $record_numbers );
 319     }
 320     #FIXME Current behaviour is to index Zebra when using ES, at some point we should stop
 321     Koha::SearchEngine::Zebra::Indexer::index_records( $self, $record_numbers, $op, $server, undef );
 322 }
 323
 324 sub _get_record {
 325     my ( $id, $server ) = @_;
 326     return $server eq 'biblioserver'
 327         ? C4::Biblio::GetMarcBiblio({ biblionumber => $id, embed_items  => 1 })
 328         : C4::AuthoritiesMarc::GetAuthority($id);
 329 }
 330
 331 =head2 delete_index($biblionums)
 332
 333 C<$biblionums> is an arrayref of biblionumbers to delete from the index.
 334
 335 =cut
 336
 337 sub delete_index {
 338     my ($self, $biblionums) = @_;
 339
 340     my $elasticsearch = $self->get_elasticsearch();
 341     my @body = map { { delete => { _id => "$_" } } } @{$biblionums};
 342     my $result = $elasticsearch->bulk(
 343         index => $self->index_name,
 344         type => 'data',
 345         body => \@body,
 346     );
 347     if ($result->{errors}) {
 348         croak "An Elasticsearch error occurred during bulk delete";
 349     }
 350 }
 351
 352 =head2 delete_index_background($biblionums)
 353
 354 Identical to L</delete_index($biblionums)>
 355
 356 =cut
 357
 358 # TODO: Should be made async
 359 sub delete_index_background {
 360     my $self = shift;
 361     $self->delete_index(@_);
 362 }
 363
 364 =head2 drop_index
 365
 366 Drops the index from the Elasticsearch server.
 367
 368 =cut
 369
 370 sub drop_index {
 371     my ($self) = @_;
 372     if ($self->index_exists) {
 373         my $elasticsearch = $self->get_elasticsearch();
 374         $elasticsearch->indices->delete(index => $self->index_name);
 375         $self->set_index_status_recreate_required();
 376     }
 377 }
 378
 379 =head2 create_index
 380
 381 Creates the index (including mappings) on the Elasticsearch server.
 382
 383 =cut
 384
 385 sub create_index {
 386     my ($self) = @_;
 387     my $settings = $self->get_elasticsearch_settings();
 388     my $elasticsearch = $self->get_elasticsearch();
 389     $elasticsearch->indices->create(
 390         index => $self->index_name,
 391         body => {
 392             settings => $settings
 393         }
 394     );
 395     $self->update_mappings();
 396 }
 397
 398 =head2 index_exists
 399
 400 Checks if index has been created on the Elasticsearch server. Returns C<1> or the
 401 empty string to indicate whether index exists or not.
 402
 403 =cut
 404
 405 sub index_exists {
 406     my ($self) = @_;
 407     my $elasticsearch = $self->get_elasticsearch();
 408     return $elasticsearch->indices->exists(
 409         index => $self->index_name,
 410     );
 411 }
 412
 413 1;
 414
 415 __END__
 416
 417 =head1 AUTHOR
 418
 419 =over 4
 420
 421 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
 422
 423 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
 424
 425 =back