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

(-)a/Koha/BackgroundJob/UpdateElasticIndex.pm (-1 / +3 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
    # elastic_index queue will be handled by the es_indexer_daemon script
108
109
109
    $self->SUPER::enqueue({
110
    $self->SUPER::enqueue({
110
        job_size => 1,
111
        job_size => 1, # Each index is a single job, regardless of the amount of records included
111
        job_args => {record_server => $record_server, record_ids => \@record_ids},
112
        job_args => {record_server => $record_server, record_ids => \@record_ids},
113
        job_queue => 'elastic_index'
112
    });
114
    });
113
}
115
}
114
116
(-)a/misc/background_jobs_worker.pl (-5 / +8 lines)
Lines 53-58 use Try::Tiny; Link Here
53
use Pod::Usage;
53
use Pod::Usage;
54
use Getopt::Long;
54
use Getopt::Long;
55
55
56
use C4::Context;
56
use Koha::Logger;
57
use Koha::Logger;
57
use Koha::BackgroundJobs;
58
use Koha::BackgroundJobs;
58
59
Lines 79-89 if ( $conn ) { Link Here
79
    # FIXME cf note in Koha::BackgroundJob about $namespace
80
    # FIXME cf note in Koha::BackgroundJob about $namespace
80
    my $namespace = C4::Context->config('memcached_namespace');
81
    my $namespace = C4::Context->config('memcached_namespace');
81
    for my $queue (@queues) {
82
    for my $queue (@queues) {
82
        $conn->subscribe({
83
        $conn->subscribe(
83
            destination => sprintf("/queue/%s-%s", $namespace, $queue),
84
            {
84
            ack => 'client',
85
                destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
85
            'prefetch-count' => 1,
86
                ack              => 'client',
86
        });
87
                'prefetch-count' => 1,
88
            }
89
        );
87
    }
90
    }
88
}
91
}
89
while (1) {
92
while (1) {
(-)a/misc/workers/es_indexer_daemon.pl (-1 / +190 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 --batch_size=X
25
26
Options:
27
28
   --help                   brief help message
29
   -b --batch_size          how many jobs to commit
30
31
=head1 OPTIONS
32
33
=over 8
34
35
=item B<--help>
36
37
Print a brief help message and exits.
38
39
=item B<--batch_size>
40
41
How many jobs to commit per batch. Defaults to 10, will commit after .1 seconds if no more jobs incoming.
42
43
=back
44
45
=head1 DESCRIPTION
46
47
This script will connect to the Stomp server (RabbitMQ) and subscribe to the Elasticsearch queue, processing batches every second.
48
If a Stomp server is not active it will poll the database every 10s for new jobs in the Elasticsearch queue
49
and process them in batches every second.
50
51
=cut
52
53
use Modern::Perl;
54
use JSON qw( decode_json );
55
use Try::Tiny;
56
use Pod::Usage;
57
use Getopt::Long;
58
59
use C4::Context;
60
use Koha::Logger;
61
use Koha::BackgroundJobs;
62
use Koha::SearchEngine;
63
use Koha::SearchEngine::Indexer;
64
65
66
my ( $help, $batch_size );
67
GetOptions(
68
    'h|help' => \$help,
69
    'b|batch_size=s' => \$batch_size
70
) || pod2usage(1);
71
72
pod2usage(0) if $help;
73
74
$batch_size //= 10;
75
76
die "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch';
77
78
my $logger = Koha::Logger->get;
79
$logger->debug_to_screen;
80
81
my $conn;
82
try {
83
    $conn = Koha::BackgroundJob->connect;
84
} catch {
85
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
86
};
87
88
if ( $conn ) {
89
    # FIXME cf note in Koha::BackgroundJob about $namespace
90
    my $namespace = C4::Context->config('memcached_namespace');
91
    $conn->subscribe(
92
        {
93
            destination      => sprintf( "/queue/%s-%s", $namespace, 'elastic_index' ),
94
            ack              => 'client',
95
            'prefetch-count' => 1,
96
        }
97
    );
98
}
99
my $biblio_indexer = Koha::SearchEngine::Indexer->new({ index => $Koha::SearchEngine::BIBLIOS_INDEX });
100
my $auth_indexer = Koha::SearchEngine::Indexer->new({ index => $Koha::SearchEngine::AUTHORITIES_INDEX });
101
my @jobs = ();
102
103
while (1) {
104
105
    if ( $conn ) {
106
        my $frame = $conn->receive_frame;
107
        if ( !defined $frame ) {
108
            # maybe log connection problems
109
            next;    # will reconnect automatically
110
        }
111
112
        my $args = try {
113
            my $body = $frame->body;
114
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
115
        } catch {
116
            $logger->debug(sprintf "Frame not processed - %s", $_);
117
            return;
118
        } finally {
119
            $conn->ack( { frame => $frame } );
120
        };
121
122
        next unless $args;
123
124
        # FIXME This means we need to have create the DB entry before
125
        # It could work in a first step, but then we will want to handle job that will be created from the message received
126
        my $job = Koha::BackgroundJobs->find($args->{job_id});
127
128
        unless ( $job ) {
129
            $logger->debug(sprintf "No job found for id=%s", $args->{job_id});
130
            next;
131
        }
132
133
        push @jobs, $job;
134
        if ( @jobs >= $batch_size
135
	    || !$conn->can_read( { timeout => '0.1' } ) )
136
        {
137
	    commit(@jobs);
138
	    @jobs = ();
139
        }
140
141
    } else {
142
        @jobs = Koha::BackgroundJobs->search(
143
            { status => 'new', queue => 'elastic_index' } )->as_list;
144
        commit(@jobs);
145
        @jobs = ();
146
        sleep 10;
147
    }
148
149
}
150
$conn->disconnect;
151
152
sub commit {
153
    my ( @jobs ) = @_;
154
155
    my @bib_records;
156
    my @auth_records;
157
    for my $job ( @jobs ) {
158
        my $args = try {
159
	    $job->json->decode($job->data);
160
        } catch {
161
            $logger->debug(sprintf "Cannot decode data for job id=%s", $job->id);
162
	    $job->status('failed')->store;
163
	    return;
164
        };
165
	next unless $args;
166
	if ( $args->{record_server} eq 'biblioserver' ){
167
            push @bib_records, @{ $args->{record_ids} };
168
        } else {
169
	    push @auth_records, @{ $args->{record_ids} };
170
        }
171
    }
172
173
    if( @auth_records ){
174
        try {
175
	    $auth_indexer->update_index(\@auth_records);
176
        } catch {
177
	    $logger->debug(sprintf "Update of elastic index failed with: %s", $_);
178
        };
179
    }
180
    if( @bib_records ){
181
        try {
182
	    $biblio_indexer->update_index(\@bib_records);
183
        } catch {
184
	    $logger->debug(sprintf "Update of elastic index failed with: %s", $_);
185
        };
186
    }
187
188
    Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] } )
189
      ->update( { status => 'finished', progress => 1 }, { no_triggers => 1 } );
190
}

Return to bug 32594