View | Details | Raw Unified | Return to bug 32594
Collapse All | Expand All

(-)a/Koha/BackgroundJob/UpdateElasticIndex.pm (+4 lines)
Lines 105-114 sub enqueue { Link Here
105
105
106
    my $record_server = $args->{record_server};
106
    my $record_server = $args->{record_server};
107
    my @record_ids = @{ $args->{record_ids} };
107
    my @record_ids = @{ $args->{record_ids} };
108
    my $queue = $record_server eq 'biblioserver' ? 'elastic_index' : 'default';
109
    # Biblios should be sent to ES daemon for batching.
110
    # Authorities are much less common and can be processed by default
108
111
109
    $self->SUPER::enqueue({
112
    $self->SUPER::enqueue({
110
        job_size => 1,
113
        job_size => 1,
111
        job_args => {record_server => $record_server, record_ids => \@record_ids},
114
        job_args => {record_server => $record_server, record_ids => \@record_ids},
115
        job_queue => $queue
112
    });
116
    });
113
}
117
}
114
118
(-)a/misc/search_tools/es_indexer_daemon.pl (-1 / +136 lines)
Line 0 Link Here
0
- 
1
#!/usr/bin/perl
2
3
# This file is part of Koha.
4
#
5
# Koha is free software; you can redistribute it and/or modify it
6
# under the terms of the GNU General Public License as published by
7
# the Free Software Foundation; either version 3 of the License, or
8
# (at your option) any later version.
9
#
10
# Koha is distributed in the hope that it will be useful, but
11
# WITHOUT ANY WARRANTY; without even the implied warranty of
12
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
# GNU General Public License for more details.
14
#
15
# You should have received a copy of the GNU General Public License
16
# along with Koha; if not, see <http://www.gnu.org/licenses>.
17
18
=head1 NAME
19
20
background_jobs_worker_es.pl - Worker script that will process background Elasticsearch jobs
21
22
=head1 SYNOPSIS
23
24
./background_jobs_worker_es.pl
25
26
=head1 DESCRIPTION
27
28
This script will connect to the Stomp server (RabbitMQ) and subscribe to the Elasticsearch queue, processing batches every second.
29
If a Stomp server is not active it will poll the database every 10s for new jobs in the Elasticsearch queue
30
and process them in batches every second.
31
32
=cut
33
34
use Modern::Perl;
35
use JSON qw( decode_json );
36
use Try::Tiny qw( catch try );
37
use Pod::Usage;
38
use Getopt::Long;
39
40
use C4::Context;
41
use Koha::BackgroundJobs;
42
use Koha::SearchEngine;
43
use Koha::SearchEngine::Indexer;
44
45
my ( $help );
46
GetOptions(
47
    'h|help' => \$help,
48
) || pod2usage(1);
49
50
pod2usage(0) if $help;
51
52
die "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch';
53
54
my $conn;
55
try {
56
    $conn = Koha::BackgroundJob->connect;
57
} catch {
58
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
59
};
60
61
if ( $conn ) {
62
    # FIXME cf note in Koha::BackgroundJob about $namespace
63
    my $namespace = C4::Context->config('memcached_namespace');
64
    $conn->subscribe({ destination => sprintf("/queue/%s-%s", $namespace, 'elastic_index'), ack => 'client' });
65
}
66
my $indexer = Koha::SearchEngine::Indexer->new({ index => $Koha::SearchEngine::BIBLIOS_INDEX });
67
my @jobs = ();
68
my @records = ();
69
my $frame;
70
71
while (1) {
72
    local $SIG{ALRM} = sub { commit_records() if scalar @jobs }; # NB: \n required
73
    alarm 1;
74
    if ( $conn ) {
75
        $frame = $conn->receive_frame;
76
        if ( !defined $frame ) {
77
            # maybe log connection problems
78
            next;    # will reconnect automatically
79
        }
80
81
        my $body = $frame->body;
82
        my $args = decode_json($body); # TODO Should this be from_json? Check utf8 flag.<F9>
83
84
        # FIXME This means we need to have create the DB entry before
85
        # It could work in a first step, but then we will want to handle job that will be created from the message received
86
        my $job = Koha::BackgroundJobs->find($args->{job_id});
87
        next unless defined $job;
88
89
        push @records, @{ $args->{record_ids} };
90
        push @jobs, $job;
91
92
93
    } else {
94
        my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => 'elastic_index' });
95
        while ( my $job = $jobs->next ) {
96
            my $args = $job->json->decode($job->data);
97
            push @records, @{ $args->{record_ids} };
98
            push @jobs, $job;
99
        }
100
        sleep 10;
101
    }
102
}
103
$conn->disconnect;
104
105
sub commit_records {
106
    eval {
107
        $indexer->update_index(\@records);
108
    };
109
    if ( $@ ){
110
        warn $@;
111
    }
112
    foreach my $job (@jobs){
113
        $job->status('finished')->store;
114
    }
115
    @jobs = ();
116
    @records = ();
117
    $conn->ack( { frame => $frame } ) if $frame; # FIXME depending on success?
118
    # Acknowledging the current frame also acknowledges all previously batched frames
119
}
120
121
122
sub process_job {
123
    my ( $job, $args ) = @_;
124
125
    my $pid;
126
    if ( $pid = fork ) {
127
        wait;
128
        return;
129
    }
130
131
    die "fork failed!" unless defined $pid;
132
133
    $job->process( $args );
134
135
    exit;
136
}

Return to bug 32594