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

(-)a/Koha/BackgroundWorker.pm (-4 / +13 lines)
Lines 50-55 Koha::BackgroundWorker - Centralized code for background worker scripts Link Here
50
50
51
    Main method to start the worker.
51
    Main method to start the worker.
52
    Available parameters:
52
    Available parameters:
53
    - batch_size
53
    - callback
54
    - callback
54
    - queues
55
    - queues
55
    - max_processes
56
    - max_processes
Lines 90-95 sub run { Link Here
90
    }
91
    }
91
92
92
    ### The main loop
93
    ### The main loop
94
    my @jobs;
95
    my $lastseen = time;
93
    while (1) {
96
    while (1) {
94
        if ($conn) {
97
        if ($conn) {
95
            my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } );
98
            my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } );
Lines 147-156 sub run { Link Here
147
            }
150
            }
148
            $conn->ack( { frame => $frame } );
151
            $conn->ack( { frame => $frame } );
149
152
150
            $pm->start and next;
153
            push @jobs, [ $job, $args ];
151
            srand();    # ensure each child process begins with a new seed
154
            if ( @jobs >= $self->{batch_size} || time > $lastseen ) {
152
            &$callback( $job, $args );
155
                my @commits = @jobs; @jobs = ();
153
            $pm->finish;
156
                $lastseen    = time;
157
                $pm->start and next;
158
                srand();    # ensure each child process begins with a new seed
159
                &$callback(@commits);
160
                $pm->finish;
161
            }
154
162
155
        } else {
163
        } else {
156
            my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } );
164
            my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } );
Lines 183-188 sub run { Link Here
183
191
184
sub _init {
192
sub _init {
185
    my ($self) = @_;
193
    my ($self) = @_;
194
    $self->{batch_size}    //= 1;
186
    $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes}
195
    $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes}
187
        if C4::Context->config('background_jobs_worker');
196
        if C4::Context->config('background_jobs_worker');
188
    $self->{max_processes} ||= 1;
197
    $self->{max_processes} ||= 1;
(-)a/misc/workers/background_jobs_worker.pl (-6 / +9 lines)
Lines 82-91 Koha::BackgroundWorker->run( Link Here
82
);
82
);
83
83
84
sub process_job {
84
sub process_job {
85
    my ( $job, $args ) = @_;
85
    my ( @jobs ) = @_;
86
    try {
86
    foreach my $elt (@jobs) {
87
        $job->process( $args );
87
        my ( $job, $args ) = @$elt;
88
    } catch {
88
        try {
89
        $job->status('failed')->store;
89
            $job->process( $args );
90
    };
90
        } catch {
91
            $job->status('failed')->store;
92
        };
93
    }
91
}
94
}
(-)a/misc/workers/es_indexer_daemon.pl (-107 / +18 lines)
Lines 56-191 use Try::Tiny; Link Here
56
use Pod::Usage;
56
use Pod::Usage;
57
use Getopt::Long;
57
use Getopt::Long;
58
use List::MoreUtils qw( natatime );
58
use List::MoreUtils qw( natatime );
59
use Time::HiRes;
60
59
61
use C4::Context;
60
use C4::Context;
62
use Koha::Logger;
63
use Koha::BackgroundJobs;
61
use Koha::BackgroundJobs;
62
use Koha::Logger;
64
use Koha::SearchEngine;
63
use Koha::SearchEngine;
65
use Koha::SearchEngine::Indexer;
64
use Koha::SearchEngine::Indexer;
66
65
67
66
my ( $max_processes, $help, $batch_size );
68
my ( $help, $batch_size );
69
70
my $not_found_retries = {};
71
my $max_retries = $ENV{MAX_RETRIES} || 10;
72
73
GetOptions(
67
GetOptions(
74
    'h|help' => \$help,
68
    'm|max-processes=i' => \$max_processes,
75
    'b|batch_size=s' => \$batch_size
69
    'h|help'            => \$help,
70
    'b|batch_size=s'    => \$batch_size
76
) || pod2usage(1);
71
) || pod2usage(1);
77
78
pod2usage(0) if $help;
72
pod2usage(0) if $help;
79
80
$batch_size //= 10;
73
$batch_size //= 10;
81
74
82
warn "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch';
75
warn "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch';
83
84
my $logger = Koha::Logger->get({ interface =>  'worker' });
85
86
my $conn;
87
try {
88
    $conn = Koha::BackgroundJob->connect;
89
} catch {
90
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
91
};
92
93
if ( $conn ) {
94
    # FIXME cf note in Koha::BackgroundJob about $namespace
95
    my $namespace = C4::Context->config('memcached_namespace');
96
    $conn->subscribe(
97
        {
98
            destination      => sprintf( "/queue/%s-%s", $namespace, 'elastic_index' ),
99
            ack              => 'client',
100
            'prefetch-count' => 1,
101
        }
102
    );
103
}
104
my $biblio_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::BIBLIOS_INDEX } );
76
my $biblio_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::BIBLIOS_INDEX } );
105
my $auth_indexer   = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::AUTHORITIES_INDEX } );
77
my $auth_indexer   = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::AUTHORITIES_INDEX } );
106
my $config         = $biblio_indexer->get_elasticsearch_params;
78
my $config         = $biblio_indexer->get_elasticsearch_params;
107
my $at_a_time      = $config->{chunk_size} // 5000;
79
my $at_a_time      = $config->{chunk_size} // 5000;
108
80
my $lastseen       = time;
109
my @jobs = ();
81
110
82
Koha::BackgroundWorker->run(
111
while (1) {
83
    {
112
84
        max_processes => $max_processes // $ENV{MAX_PROCESSES},
113
    if ( $conn ) {
85
        mq_timeout    => $ENV{MQ_TIMEOUT},
114
        my $frame = $conn->receive_frame;
86
        max_retries   => $ENV{MAX_RETRIES},
115
        if ( !defined $frame ) {
87
        callback      => \&commit,
116
            # maybe log connection problems
88
        queues        => [ 'elastic_index' ],
117
            next;    # will reconnect automatically
118
        }
119
120
        my $args = try {
121
            my $body = $frame->body;
122
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
123
        } catch {
124
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_);
125
            return;
126
        };
127
128
        unless ( $args ) {
129
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it");
130
            $conn->nack( { frame => $frame, requeue => 'false' } );
131
            next;
132
        }
133
134
        my $job = Koha::BackgroundJobs->find( $args->{job_id} );
135
136
        if ( $job && $job->status ne 'new' ) {
137
            Koha::Logger->get( { interface => 'worker' } )
138
                ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
139
140
            # nack without requeue, we do not want to process this frame again
141
            $conn->nack( { frame => $frame, requeue => 'false' } );
142
            next;
143
        }
144
145
        unless ($job) {
146
            $not_found_retries->{ $args->{job_id} } //= 0;
147
            if ( ++$not_found_retries->{ $args->{job_id} } >= $max_retries ) {
148
                Koha::Logger->get( { interface => 'worker' } )
149
                    ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
150
151
                # nack without requeue, we do not want to process this frame again
152
                $conn->nack( { frame => $frame, requeue => 'false' } );
153
                next;
154
            }
155
156
            Koha::Logger->get( { interface => 'worker' } )
157
                ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
158
159
            # nack to force requeue
160
            $conn->nack( { frame => $frame, requeue => 'true' } );
161
            Time::HiRes::sleep(0.5);
162
            next;
163
        }
164
        $conn->ack( { frame => $frame } );
165
166
        push @jobs, $job;
167
        if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) {
168
            commit(@jobs);
169
            @jobs = ();
170
        }
171
172
    } else {
173
        @jobs = Koha::BackgroundJobs->search(
174
            { status => 'new', queue => 'elastic_index' } )->as_list;
175
        commit(@jobs);
176
        @jobs = ();
177
        sleep 10;
178
    }
89
    }
179
90
);
180
}
181
$conn->disconnect;
182
91
183
sub commit {
92
sub commit {
184
    my (@jobs) = @_;
93
    my (@jobs) = @_;
94
    # Extract jobs only from passed arrayrefs
95
    @jobs = map { $_->[0] } @jobs;
185
96
186
    my @bib_records;
97
    my @bib_records;
187
    my @auth_records;
98
    my @auth_records;
188
99
100
    my $logger = Koha::Logger->get({ interface =>  'worker' });
189
    my $jobs = Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] });
101
    my $jobs = Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] });
190
    # Start
102
    # Start
191
    $jobs->update({
103
    $jobs->update({
192
- 

Return to bug 35920