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

(-)a/Koha/BackgroundJob.pm (-74 / +3 lines)
Lines 18-25 package Koha::BackgroundJob; Link Here
18
use Modern::Perl;
18
use Modern::Perl;
19
use Encode qw();
19
use Encode qw();
20
use JSON;
20
use JSON;
21
use Carp qw( croak );
21
use Carp      qw( croak );
22
use Net::Stomp;
23
use Try::Tiny qw( catch try );
22
use Try::Tiny qw( catch try );
24
23
25
use C4::Context;
24
use C4::Context;
Lines 55-103 See also C<misc/background_jobs_worker.pl> for a full example Link Here
55
54
56
=head2 Class methods
55
=head2 Class methods
57
56
58
=head3 connect
59
60
Connect to the message broker using default guest/guest credential
61
62
=cut
63
64
sub connect {
65
    my ($self);
66
67
    my $notification_method = C4::Context->preference('JobsNotificationMethod') // 'STOMP';
68
69
    return
70
        unless $notification_method eq 'STOMP';
71
72
    my $hostname = 'localhost';
73
    my $port     = '61613';
74
75
    my $config      = C4::Context->config('message_broker');
76
    my $credentials = {
77
        login    => 'guest',
78
        passcode => 'guest',
79
    };
80
    if ($config) {
81
        $hostname                = $config->{hostname} if $config->{hostname};
82
        $port                    = $config->{port}     if $config->{port};
83
        $credentials->{login}    = $config->{username} if $config->{username};
84
        $credentials->{passcode} = $config->{password} if $config->{password};
85
        $credentials->{host}     = $config->{vhost}    if $config->{vhost};
86
    }
87
88
    my $stomp;
89
90
    try {
91
        $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
92
        $stomp->connect($credentials);
93
    } catch {
94
        warn "Cannot connect to broker " . $_;
95
        $stomp = undef;
96
    };
97
98
    return $stomp;
99
}
100
101
=head3 enqueue
57
=head3 enqueue
102
58
103
Enqueue a new job. It will insert a new row in the DB table and notify the broker that a new job has been enqueued.
59
Enqueue a new job. It will insert a new row in the DB table and notify the broker that a new job has been enqueued.
Lines 137-167 sub enqueue { Link Here
137
        }
93
        }
138
    )->store;
94
    )->store;
139
95
140
    $job_args->{job_id} = $self->id;
141
142
    my $conn = $self->connect;
143
    return $self->id unless $conn;
144
145
    $json_args = $json->encode($job_args);
146
    try {
147
        # This namespace is wrong, it must be a vhost instead.
148
        # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
149
        # Also, here we just want the Koha instance's name, but it's not in the config...
150
        # Picking a random id (memcached_namespace) from the config
151
        my $namespace    = C4::Context->config('memcached_namespace');
152
        my $encoded_args = Encode::encode_utf8($json_args);           # FIXME We should better leave this to Net::Stomp?
153
        my $destination  = sprintf( "/queue/%s-%s", $namespace, $job_queue );
154
        $conn->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } )
155
            or Koha::Exceptions::BackgroundJob->throw('Job has not been enqueued');
156
    } catch {
157
        $self->status('failed')->store;
158
        if ( ref($_) eq 'Koha::Exceptions::BackgroundJob' ) {
159
            $_->rethrow;
160
        } else {
161
            warn sprintf "The job has not been sent to the message broker: (%s)", $_;
162
        }
163
    };
164
165
    return $self->id;
96
    return $self->id;
166
}
97
}
167
98
Lines 172-178 Process the job! Link Here
172
=cut
103
=cut
173
104
174
sub process {
105
sub process {
175
    my ( $self, $args ) = @_;
106
    my ($self) = @_;
176
107
177
    return {} if ref($self) ne 'Koha::BackgroundJob';
108
    return {} if ref($self) ne 'Koha::BackgroundJob';
178
109
Lines 184-191 sub process { Link Here
184
115
185
    my $derived_class = $self->_derived_class;
116
    my $derived_class = $self->_derived_class;
186
117
187
    $args ||= {};
188
189
    if ( $self->context ) {
118
    if ( $self->context ) {
190
        my $context = $self->json->decode( $self->context );
119
        my $context = $self->json->decode( $self->context );
191
        C4::Context->interface( $context->{interface} );
120
        C4::Context->interface( $context->{interface} );
Lines 202-208 sub process { Link Here
202
        Koha::Logger->get->warn( "A background job didn't have context defined (" . $self->id . ")" );
131
        Koha::Logger->get->warn( "A background job didn't have context defined (" . $self->id . ")" );
203
    }
132
    }
204
133
205
    return $derived_class->process($args);
134
    return $derived_class->process();
206
}
135
}
207
136
208
=head3 start
137
=head3 start
(-)a/Koha/Worker.pm (+122 lines)
Line 0 Link Here
1
package Koha::Worker;
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
use Modern::Perl;
19
20
use C4::Context;
21
use Koha::BackgroundJobs;
22
use Koha::Logger;
23
24
use Parallel::ForkManager;
25
use Try::Tiny qw(catch try);
26
27
=head1 NAME
28
29
Koha::Worker - Koha bakground jobs worker class
30
31
=head1 API
32
33
=head2 Class methods
34
35
=head3 new
36
37
    my $worker = Koha::Worker->new(
38
        {
39
          [ queues        => [ 'default', 'long_tasks', ... ], ]
40
          [ max_processes => N, ]
41
        }
42
    );
43
44
Constructor for the I<Koha::Worker> class.
45
46
=cut
47
48
sub new {
49
    my ( $class, $params ) = @_;
50
51
    my $max_processes = $params->{max_processes};
52
    $max_processes //= C4::Context->config('background_jobs_worker') // 1;
53
54
    my $queues = $params->{queues} // ['default'];
55
56
    my $self = {
57
        max_processes => $max_processes,
58
        queues        => $queues,
59
    };
60
61
    bless $self, $class;
62
    return $self;
63
}
64
65
=head3 run
66
67
    $worker->run();
68
69
Method that triggers the main loop.
70
71
=cut
72
73
sub run {
74
    my ($self) = @_;
75
76
    my $pm = Parallel::ForkManager->new( $self->{max_processes} );
77
78
    # Main loop
79
    while (1) {
80
81
        # FIXME: we should put a limit on the rows to featch each time.
82
        my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => $self->{queues} } );
83
        while ( my $job = $jobs->next ) {
84
            $pm->start and next;
85
            srand();    # ensure each child process begins with a new seed
86
            $self->process($job);
87
            $pm->finish;
88
89
        }
90
        $pm->reap_finished_children;
91
        sleep 10;
92
    }
93
94
    # Work is done
95
    $pm->wait_all_children;
96
}
97
98
=head3 process
99
100
    $self->process($job);
101
102
Method that takes care of running the required job.
103
104
=cut
105
106
sub process {
107
    my ( $self, $job ) = @_;
108
    try {
109
        Koha::Logger->get( { interface => 'worker' } )->info( sprintf "Started '%s' job id=%s", $job->type, $job->id );
110
        $job->process();
111
        Koha::Logger->get( { interface => 'worker' } )
112
            ->info( sprintf "Finished '%s' job id=%s status=%s", $job->type, $job->id, $job->status );
113
    } catch {
114
        Koha::Logger->get( { interface => 'worker' } )
115
            ->warn( sprintf "Uncaught exception processing job id=%s: %s", $job->id, $_ );
116
        $job->status('failed')->store;
117
    };
118
119
    return;
120
}
121
122
1;
(-)a/misc/workers/background_jobs_worker.pl (-138 / +15 lines)
Lines 25-38 background_jobs_worker.pl - Worker script that will process background jobs Link Here
25
25
26
=head1 DESCRIPTION
26
=head1 DESCRIPTION
27
27
28
This script will connect to the Stomp server (RabbitMQ) and subscribe to the queues passed in parameter (or the 'default' queue),
28
This script will launch a worker for the specified queues. It will poll the database
29
or if a Stomp server is not active it will poll the database every 10s for new jobs in the passed queue.
29
every 10s for new jobs in the passed queue.
30
30
31
You can specify some queues only (using --queue, which is repeatable) if you want to run several workers that will handle their own jobs.
31
You can specify some queues only (using --queue, which is repeatable) if you want to
32
run several workers that will handle their own jobs.
32
33
33
--m --max-processes specifies how many jobs to process simultaneously
34
--m --max-processes specifies how many jobs to process simultaneously
34
35
35
Max processes will be set from the command line option, the environment variable MAX_PROCESSES, or the koha-conf file, in that order of precedence.
36
Max processes will be set from the command line option, the environment variable MAX_PROCESSES,
37
or the koha-conf file, in that order of precedence.
38
36
By default the script will only run one job at a time.
39
By default the script will only run one job at a time.
37
40
38
=head1 OPTIONS
41
=head1 OPTIONS
Lines 53-69 The different values available are: Link Here
53
=cut
56
=cut
54
57
55
use Modern::Perl;
58
use Modern::Perl;
56
use JSON qw( decode_json );
59
57
use Try::Tiny;
60
use Try::Tiny;
58
use Pod::Usage;
61
use Pod::Usage;
59
use Getopt::Long;
62
use Getopt::Long;
60
use Parallel::ForkManager;
61
use Time::HiRes;
62
63
63
use C4::Context;
64
use Koha::Worker;
64
use Koha::Logger;
65
use Koha::BackgroundJobs;
66
use C4::Context;
67
65
68
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
66
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
69
67
Lines 73-82 my $max_processes = $ENV{MAX_PROCESSES}; Link Here
73
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes}
71
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes}
74
    if C4::Context->config('background_jobs_worker');
72
    if C4::Context->config('background_jobs_worker');
75
$max_processes ||= 1;
73
$max_processes ||= 1;
76
my $mq_timeout = $ENV{MQ_TIMEOUT} // 10;
77
78
my $not_found_retries = {};
79
my $max_retries       = $ENV{MAX_RETRIES} || 10;
80
74
81
GetOptions(
75
GetOptions(
82
    'm|max-processes=i' => \$max_processes,
76
    'm|max-processes=i' => \$max_processes,
Lines 90-214 unless (@queues) { Link Here
90
    push @queues, 'default';
84
    push @queues, 'default';
91
}
85
}
92
86
93
my $conn;
87
my $worker = Koha::Worker->new( { max_processes => $max_processes, queues => \@queues } );
94
try {
88
95
    $conn = Koha::BackgroundJob->connect;
89
# main loop
96
} catch {
90
$worker->run();
97
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
91
98
};
92
1;
99
100
my $pm = Parallel::ForkManager->new($max_processes);
101
102
if ($conn) {
103
104
    # FIXME cf note in Koha::BackgroundJob about $namespace
105
    my $namespace = C4::Context->config('memcached_namespace');
106
    for my $queue (@queues) {
107
        $conn->subscribe(
108
            {
109
                destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
110
                ack              => 'client',
111
                'prefetch-count' => 1,
112
            }
113
        );
114
    }
115
}
116
while (1) {
117
    if ($conn) {
118
        my $frame = $conn->receive_frame( { timeout => $mq_timeout } );
119
        if ( !defined $frame ) {
120
121
            # timeout or connection issue?
122
            $pm->reap_finished_children;
123
            next;    # will reconnect automatically
124
        }
125
126
        my $args = try {
127
            my $body = $frame->body;
128
            decode_json($body);    # TODO Should this be from_json? Check utf8 flag.
129
        } catch {
130
            Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ );
131
            return;
132
        };
133
134
        unless ($args) {
135
            Koha::Logger->get( { interface => 'worker' } )
136
                ->warn( sprintf "Frame does not have correct args, ignoring it" );
137
            $conn->nack( { frame => $frame, requeue => 'false' } );
138
            next;
139
        }
140
141
        my $job = Koha::BackgroundJobs->find( $args->{job_id} );
142
143
        if ( $job && $job->status ne 'new' ) {
144
            Koha::Logger->get( { interface => 'worker' } )
145
                ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
146
147
            # nack without requeue, we do not want to process this frame again
148
            $conn->nack( { frame => $frame, requeue => 'false' } );
149
            next;
150
        }
151
152
        unless ($job) {
153
            $not_found_retries->{ $args->{job_id} } //= 0;
154
            if ( ++$not_found_retries->{ $args->{job_id} } >= $max_retries ) {
155
                Koha::Logger->get( { interface => 'worker' } )
156
                    ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
157
158
                # nack without requeue, we do not want to process this frame again
159
                $conn->nack( { frame => $frame, requeue => 'false' } );
160
                next;
161
            }
162
163
            Koha::Logger->get( { interface => 'worker' } )
164
                ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
165
166
            # nack to force requeue
167
            $conn->nack( { frame => $frame, requeue => 'true' } );
168
            Time::HiRes::sleep(0.5);
169
            next;
170
        }
171
        $conn->ack( { frame => $frame } );
172
173
        $pm->start and next;
174
        srand();    # ensure each child process begins with a new seed
175
        process_job( $job, $args );
176
        $pm->finish;
177
178
    } else {
179
        my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } );
180
        while ( my $job = $jobs->next ) {
181
            my $args = try {
182
                $job->json->decode( $job->data );
183
            } catch {
184
                Koha::Logger->get( { interface => 'worker' } )
185
                    ->warn( sprintf "Cannot decode data for job id=%s", $job->id );
186
                $job->status('failed')->store;
187
                return;
188
            };
189
190
            next unless $args;
191
192
            $pm->start and next;
193
            srand();    # ensure each child process begins with a new seed
194
            process_job( $job, { job_id => $job->id, %$args } );
195
            $pm->finish;
196
197
        }
198
        $pm->reap_finished_children;
199
        sleep 10;
200
    }
201
}
202
$conn->disconnect;
203
$pm->wait_all_children;
204
205
sub process_job {
206
    my ( $job, $args ) = @_;
207
    try {
208
        $job->process($args);
209
    } catch {
210
        Koha::Logger->get( { interface => 'worker' } )
211
            ->warn( sprintf "Uncaught exception processing job id=%s: %s", $job->id, $_ );
212
        $job->status('failed')->store;
213
    };
214
}
215
- 

Return to bug 35920