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

(-)a/Koha/BackgroundJob.pm (-47 / +5 lines)
Lines 23-28 use Net::Stomp; Link Here
23
use Try::Tiny qw( catch try );
23
use Try::Tiny qw( catch try );
24
24
25
use C4::Context;
25
use C4::Context;
26
use Koha::BackgroundJob::Broker;
26
use Koha::DateUtils qw( dt_from_string );
27
use Koha::DateUtils qw( dt_from_string );
27
use Koha::Exceptions;
28
use Koha::Exceptions;
28
use Koha::Exceptions::BackgroundJob;
29
use Koha::Exceptions::BackgroundJob;
Lines 54-87 See also C<misc/background_jobs_worker.pl> for a full example Link Here
54
55
55
=head2 Class methods
56
=head2 Class methods
56
57
57
=head3 connect
58
=head3 enqueue_job
58
59
Connect to the message broker using default guest/guest credential
60
61
=cut
62
63
sub connect {
64
    my ( $self );
65
    my $hostname = 'localhost';
66
    my $port = '61613';
67
    my $config = C4::Context->config('message_broker');
68
    my $credentials = {
69
        login => 'guest',
70
        passcode => 'guest',
71
    };
72
    if ($config){
73
        $hostname = $config->{hostname} if $config->{hostname};
74
        $port = $config->{port} if $config->{port};
75
        $credentials->{login} = $config->{username} if $config->{username};
76
        $credentials->{passcode} = $config->{password} if $config->{password};
77
        $credentials->{host} = $config->{vhost} if $config->{vhost};
78
    }
79
    my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
80
    $stomp->connect( $credentials );
81
    return $stomp;
82
}
83
84
=head3 enqueue
85
59
86
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.
60
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.
87
61
Lines 99-105 sub enqueue { Link Here
99
    my $job_size    = $params->{job_size};
73
    my $job_size    = $params->{job_size};
100
    my $job_args    = $params->{job_args};
74
    my $job_args    = $params->{job_args};
101
    my $job_context = $params->{job_context} // C4::Context->userenv;
75
    my $job_context = $params->{job_context} // C4::Context->userenv;
102
    my $job_queue   = $params->{job_queue}  // 'default';
76
    my $job_queue   = $params->{job_queue} // 'default';
103
    my $json = $self->json;
77
    my $json = $self->json;
104
78
105
    my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
79
    my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
Lines 121-146 sub enqueue { Link Here
121
    )->store;
95
    )->store;
122
96
123
    $job_args->{job_id} = $self->id;
97
    $job_args->{job_id} = $self->id;
124
125
    my $conn;
126
    try {
127
        $conn = $self->connect;
128
    } catch {
129
        warn "Cannot connect to broker " . $_;
130
    };
131
    return $self->id unless $conn;
132
133
    $json_args = $json->encode($job_args);
98
    $json_args = $json->encode($job_args);
99
134
    try {
100
    try {
135
        # This namespace is wrong, it must be a vhost instead.
101
        Koha::BackgroundJob::Broker->new->connect->enqueue_job({queue => $job_queue, body => $json_args});
136
        # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
137
        # Also, here we just want the Koha instance's name, but it's not in the config...
138
        # Picking a random id (memcached_namespace) from the config
139
        my $namespace = C4::Context->config('memcached_namespace');
140
        my $encoded_args = Encode::encode_utf8( $json_args ); # FIXME We should better leave this to Net::Stomp?
141
        my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue );
142
        $conn->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } )
143
          or Koha::Exceptions::Exception->throw('Job has not been enqueued');
144
    } catch {
102
    } catch {
145
        $self->status('failed')->store;
103
        $self->status('failed')->store;
146
        if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
104
        if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
(-)a/Koha/BackgroundJob/Broker.pm (+178 lines)
Line 0 Link Here
1
package Koha::BackgroundJob::Broker;
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
use Encode qw();
20
use JSON;
21
use Net::Stomp;
22
use Try::Tiny qw( catch try );
23
use Time::HiRes;
24
25
use C4::Context;
26
27
my $not_found_retries = {};
28
my $max_retries       = $ENV{MAX_RETRIES} || 10;
29
30
=head1 NAME
31
32
Koha::BackgroundJob::Broker - Class to connect to RabbitMQ
33
34
=head1 API
35
36
=head2 Class methods
37
38
=cut
39
40
sub new {
41
    my $class = shift;
42
    my $self  = {};
43
    return bless $self, $class;
44
}
45
46
=head3 connect
47
48
Connect to the message broker using default guest/guest credential
49
50
=cut
51
52
sub connect {
53
    my ($self)      = @_;
54
    my $hostname    = 'localhost';
55
    my $port        = '61613';
56
    my $config      = C4::Context->config('message_broker');
57
    my $credentials = {
58
        login    => 'guest',
59
        passcode => 'guest',
60
    };
61
    if ($config) {
62
        $hostname                = $config->{hostname} if $config->{hostname};
63
        $port                    = $config->{port}     if $config->{port};
64
        $credentials->{login}    = $config->{username} if $config->{username};
65
        $credentials->{passcode} = $config->{password} if $config->{password};
66
        $credentials->{host}     = $config->{vhost}    if $config->{vhost};
67
    }
68
    my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
69
    $stomp->connect($credentials);
70
    $self->{conn} = $stomp;
71
    return $self;
72
}
73
74
sub subscribe {
75
    my ( $self, @queues ) = @_;
76
77
    # FIXME cf note in Koha::BackgroundJob about $namespace
78
    my $namespace = C4::Context->config('memcached_namespace');
79
    for my $queue (@queues) {
80
        $self->{conn}->subscribe(
81
            {
82
                destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
83
                ack              => 'client',
84
                'prefetch-count' => 1,
85
            }
86
        );
87
    }
88
}
89
90
sub receive_frame {
91
    my ($self) = @_;
92
    return $self->{conn}->receive_frame;
93
}
94
95
=head3 enqueue_job
96
97
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.
98
99
C<job_size> is the size of the job
100
C<job_args> is the arguments of the job. It's a structure that will be JSON encoded.
101
102
Return the job_id of the newly created job.
103
104
=cut
105
106
sub enqueue_job {
107
    my ( $self, $params ) = @_;
108
109
    my $job_queue = $params->{job_queue} // 'default';
110
    my $body      = $params->{body};
111
    my $conn;
112
    try {
113
        $conn = $self->connect;
114
    } catch {
115
        warn "Cannot connect to broker " . $_;
116
    };
117
    return unless $conn;
118
119
    # This namespace is wrong, it must be a vhost instead.
120
    # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
121
    # Also, here we just want the Koha instance's name, but it's not in the config...
122
    # Picking a random id (memcached_namespace) from the config
123
    my $namespace    = C4::Context->config('memcached_namespace');
124
    my $encoded_args = Encode::encode_utf8($body);                   # FIXME We should better leave this to Net::Stomp?
125
    my $destination  = sprintf( "/queue/%s-%s", $namespace, $job_queue );
126
    $self->{conn}->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } )
127
        or Koha::Exceptions::Exception->throw('Job has not been enqueued');
128
}
129
130
sub decode_frame {
131
    my ( $self, $frame ) = @_;
132
133
    return try {
134
        my $body = $frame->body;
135
        decode_json($body);    # TODO Should this be from_json? Check utf8 flag.
136
    } catch {
137
        Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ );
138
        return;
139
    };
140
}
141
142
sub get_job {
143
    my ( $self, $params ) = @_;
144
145
    my $job_id = $params->{job_id};
146
    my $frame  = $params->{frame};
147
148
    my $job = Koha::BackgroundJobs->find($job_id);
149
150
    if ( $job && $job->status ne 'new' ) {
151
        Koha::Logger->get( { interface => 'worker' } )
152
            ->warn( sprintf "Job %s has wrong status %s", $job_id, $job->status );
153
154
        # nack without requeue, we do not want to process this frame again
155
        $self->{conn}->nack( { frame => $frame, requeue => 'false' } );
156
        return;
157
    }
158
159
    unless ($job) {
160
        if ( ++$not_found_retries->{$job_id} >= $max_retries ) {
161
            Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Job %s not found, no more retry", $job_id );
162
163
            # nack without requeue, we do not want to process this frame again
164
            $self->{conn}->nack( { frame => $frame, requeue => 'false' } );
165
            return;
166
        }
167
168
        Koha::Logger->get( { interface => 'worker' } )->debug( sprintf "Job %s not found, will retry later", $job_id );
169
170
        # nack to force requeue
171
        $self->{conn}->nack( { frame => $frame, requeue => 'true' } );
172
        Time::HiRes::sleep(0.5);
173
        return;
174
    }
175
176
}
177
178
1;
(-)a/misc/workers/background_jobs_worker.pl (-62 / +16 lines)
Lines 58-69 use Try::Tiny; Link Here
58
use Pod::Usage;
58
use Pod::Usage;
59
use Getopt::Long;
59
use Getopt::Long;
60
use Parallel::ForkManager;
60
use Parallel::ForkManager;
61
use Time::HiRes;
62
61
63
use C4::Context;
62
use C4::Context;
64
use Koha::Logger;
63
use Koha::Logger;
64
use Koha::BackgroundJob::Broker;
65
use Koha::BackgroundJobs;
65
use Koha::BackgroundJobs;
66
use C4::Context;
67
66
68
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
67
$SIG{'PIPE'} = 'IGNORE';    # See BZ 35111; added to ignore PIPE error when connection lost on Ubuntu.
69
68
Lines 73-81 my $max_processes = $ENV{MAX_PROCESSES}; Link Here
73
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker');
72
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes} if C4::Context->config('background_jobs_worker');
74
$max_processes ||= 1;
73
$max_processes ||= 1;
75
74
76
my $not_found_retries = {};
77
my $max_retries = $ENV{MAX_RETRIES} || 10;
78
79
GetOptions(
75
GetOptions(
80
    'm|max-processes=i' => \$max_processes,
76
    'm|max-processes=i' => \$max_processes,
81
    'h|help' => \$help,
77
    'h|help' => \$help,
Lines 89-168 unless (@queues) { Link Here
89
    push @queues, 'default';
85
    push @queues, 'default';
90
}
86
}
91
87
92
my $conn;
88
my $rabbit;
93
try {
89
try {
94
    $conn = Koha::BackgroundJob->connect;
90
    $rabbit = Koha::BackgroundJob::Broker->new->connect;
91
    $rabbit->subscribe(@queues);
95
} catch {
92
} catch {
96
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
93
    warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_;
97
};
94
};
98
95
99
my $pm = Parallel::ForkManager->new($max_processes);
96
my $pm = Parallel::ForkManager->new($max_processes);
100
97
101
if ( $conn ) {
102
    # FIXME cf note in Koha::BackgroundJob about $namespace
103
    my $namespace = C4::Context->config('memcached_namespace');
104
    for my $queue (@queues) {
105
        $conn->subscribe(
106
            {
107
                destination      => sprintf( "/queue/%s-%s", $namespace, $queue ),
108
                ack              => 'client',
109
                'prefetch-count' => 1,
110
            }
111
        );
112
    }
113
}
114
while (1) {
98
while (1) {
115
    if ( $conn ) {
99
    if ( $rabbit->{conn} ) {
116
        my $frame = $conn->receive_frame;
100
        my $frame = $rabbit->receive_frame;
101
117
        if ( !defined $frame ) {
102
        if ( !defined $frame ) {
103
118
            # maybe log connection problems
104
            # maybe log connection problems
119
            next;    # will reconnect automatically
105
            next;    # will reconnect automatically
120
        }
106
        }
121
107
122
        my $args = try {
108
        my $args = $rabbit->decode_frame($frame);
123
            my $body = $frame->body;
124
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
125
        } catch {
126
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_);
127
            return;
128
        };
129
130
        unless ( $args ) {
131
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it");
132
            $conn->nack( { frame => $frame, requeue => 'false' } );
133
            next;
134
        }
135
109
136
        my $job = Koha::BackgroundJobs->find( $args->{job_id} );
110
        unless ($args) {
137
138
        if ( $job && $job->status ne 'new' ) {
139
            Koha::Logger->get( { interface => 'worker' } )
111
            Koha::Logger->get( { interface => 'worker' } )
140
                ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
112
                ->warn( sprintf "Frame does not have correct args, ignoring it" );
141
113
            $rabbit->{conn}->nack( { frame => $frame, requeue => 'false' } );
142
            # nack without requeue, we do not want to process this frame again
143
            $conn->nack( { frame => $frame, requeue => 'false' } );
144
            next;
114
            next;
145
        }
115
        }
146
116
147
        unless ($job) {
117
        my $job = $rabbit->get_job( { job_id => $args->{job_id}, frame => $frame } );
148
            if ( ++$not_found_retries->{$args->{job_id}} >= $max_retries ) {
118
        next unless $job;
149
                Koha::Logger->get( { interface => 'worker' } )
150
                    ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
151
119
152
                # nack without requeue, we do not want to process this frame again
120
        $rabbit->{conn}->ack( { frame => $frame } );
153
                $conn->nack( { frame => $frame, requeue => 'false' } );
154
                next;
155
            }
156
157
            Koha::Logger->get( { interface => 'worker' } )
158
                ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
159
160
            # nack to force requeue
161
            $conn->nack( { frame => $frame, requeue => 'true' } );
162
            Time::HiRes::sleep(0.5);
163
            next;
164
        }
165
        $conn->ack( { frame => $frame } );
166
121
167
        $pm->start and next;
122
        $pm->start and next;
168
        srand();    # ensure each child process begins with a new seed
123
        srand();    # ensure each child process begins with a new seed
Lines 191-197 while (1) { Link Here
191
        sleep 10;
146
        sleep 10;
192
    }
147
    }
193
}
148
}
194
$conn->disconnect;
149
$rabbit->{conn}->disconnect;
195
$pm->wait_all_children;
150
$pm->wait_all_children;
196
151
197
sub process_job {
152
sub process_job {
198
- 

Return to bug 35920