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::Biblio;
  30 use C4::Context;
  31
  32 =head1 NAME
  33
  34 Koha::SearchEngine::Elasticsearch::Indexer - handles adding new records to the index
  35
  36 =head1 SYNOPSIS
  37
  38     my $indexer = Koha::SearchEngine::Elasticsearch::Indexer->new(
  39         { index => Koha::SearchEngine::BIBLIOS_INDEX } );
  40     $indexer->drop_index();
  41     $indexer->update_index(\@biblionumbers, \@records);
  42
  43
  44 =head1 CONSTANTS
  45
  46 =over 4
  47
  48 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_OK>
  49
  50 Represents an index state where index is created and in a working state.
  51
  52 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_REINDEX_REQUIRED>
  53
  54 Not currently used, but could be useful later, for example if can detect when new field or mapping added.
  55
  56 =item C<Koha::SearchEngine::Elasticsearch::Indexer::INDEX_STATUS_RECREATE_REQUIRED>
  57
  58 Representings an index state where index needs to be recreated and is not in a working state.
  59
  60 =back
  61
  62 =cut
  63
  64 use constant {
  65     INDEX_STATUS_OK => 0,
  66     INDEX_STATUS_REINDEX_REQUIRED => 1,
  67     INDEX_STATUS_RECREATE_REQUIRED => 2,
  68 };
  69
  70 =head1 FUNCTIONS
  71
  72 =head2 update_index($biblionums, $records)
  73
  74     try {
  75         $self->update_index($biblionums, $records);
  76     } catch {
  77         die("Something went wrong trying to update index:" .  $_[0]);
  78     }
  79
  80 Converts C<MARC::Records> C<$records> to Elasticsearch documents and performs
  81 an update request for these records on the Elasticsearch index.
  82
  83 =over 4
  84
  85 =item C<$biblionums>
  86
  87 Arrayref of biblio numbers for the C<$records>, the order must be the same as
  88 and match up with C<$records>.
  89
  90 =item C<$records>
  91
  92 Arrayref of C<MARC::Record>s.
  93
  94 =back
  95
  96 =cut
  97
  98 sub update_index {
  99     my ($self, $biblionums, $records) = @_;
 100
 101     my $documents = $self->marc_records_to_documents($records);
 102     my @body;
 103
 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         unless ($records) {
 305             foreach my $record_number ( @$record_numbers ){
 306                 my $record = _get_record( $record_number, $server );
 307                 if( $record ){
 308                     push @$records, $record;
 309                     push @$index_record_numbers, $record_number;
 310                 }
 311             }
 312         }
 313         $self->update_index_background( $index_record_numbers, $records ) if $index_record_numbers && $records;
 314     }
 315     elsif ( $op eq 'recordDelete' ) {
 316         $self->delete_index_background( $record_numbers );
 317     }
 318     #FIXME Current behaviour is to index Zebra when using ES, at some point we should stop
 319     Koha::SearchEngine::Zebra::Indexer::index_records( $self, $record_numbers, $op, $server, undef );
 320 }
 321
 322 sub _get_record {
 323     my ( $id, $server ) = @_;
 324     return $server eq 'biblioserver'
 325         ? C4::Biblio::GetMarcBiblio({ biblionumber => $id, embed_items  => 1 })
 326         : C4::AuthoritiesMarc::GetAuthority($id);
 327 }
 328
 329 =head2 delete_index($biblionums)
 330
 331 C<$biblionums> is an arrayref of biblionumbers to delete from the index.
 332
 333 =cut
 334
 335 sub delete_index {
 336     my ($self, $biblionums) = @_;
 337
 338     my $elasticsearch = $self->get_elasticsearch();
 339     my @body = map { { delete => { _id => "$_" } } } @{$biblionums};
 340     my $result = $elasticsearch->bulk(
 341         index => $self->index_name,
 342         type => 'data',
 343         body => \@body,
 344     );
 345     if ($result->{errors}) {
 346         croak "An Elasticsearch error occurred during bulk delete";
 347     }
 348 }
 349
 350 =head2 delete_index_background($biblionums)
 351
 352 Identical to L</delete_index($biblionums)>
 353
 354 =cut
 355
 356 # TODO: Should be made async
 357 sub delete_index_background {
 358     my $self = shift;
 359     $self->delete_index(@_);
 360 }
 361
 362 =head2 drop_index
 363
 364 Drops the index from the Elasticsearch server.
 365
 366 =cut
 367
 368 sub drop_index {
 369     my ($self) = @_;
 370     if ($self->index_exists) {
 371         my $elasticsearch = $self->get_elasticsearch();
 372         $elasticsearch->indices->delete(index => $self->index_name);
 373         $self->set_index_status_recreate_required();
 374     }
 375 }
 376
 377 =head2 create_index
 378
 379 Creates the index (including mappings) on the Elasticsearch server.
 380
 381 =cut
 382
 383 sub create_index {
 384     my ($self) = @_;
 385     my $settings = $self->get_elasticsearch_settings();
 386     my $elasticsearch = $self->get_elasticsearch();
 387     $elasticsearch->indices->create(
 388         index => $self->index_name,
 389         body => {
 390             settings => $settings
 391         }
 392     );
 393     $self->update_mappings();
 394 }
 395
 396 =head2 index_exists
 397
 398 Checks if index has been created on the Elasticsearch server. Returns C<1> or the
 399 empty string to indicate whether index exists or not.
 400
 401 =cut
 402
 403 sub index_exists {
 404     my ($self) = @_;
 405     my $elasticsearch = $self->get_elasticsearch();
 406     return $elasticsearch->indices->exists(
 407         index => $self->index_name,
 408     );
 409 }
 410
 411 1;
 412
 413 __END__
 414
 415 =head1 AUTHOR
 416
 417 =over 4
 418
 419 =item Chris Cormack C<< <chrisc@catalyst.net.nz> >>
 420
 421 =item Robin Sheat C<< <robin@catalyst.net.nz> >>
 422
 423 =back