Enable gru tasks to emit AMQP messages

Start Mojo::IOLoop and let it end when the AMQP plugin has finished
(or failed) publishing the event

Author: perlpunk

lib/OpenQA/Task/Job/Restart.pm

@@ -40,8 +40,11 @@ sub _restart_job ($minion_job, @args) {
     my $openqa_job = $app->schema->resultset('Jobs')->find($openqa_job_id);
     return $minion_job->finish("Job $openqa_job_id does not exist.") unless $openqa_job;
 
+    _init_amqp_plugin($app);
     # duplicate job and finish normally if no error was returned or job can not be cloned
     my ($is_ok, $cloned_job_or_error) = restart_openqa_job($minion_job, $openqa_job);
+    _wait_for_event_publish($app);
+
     return $minion_job->finish(ref $cloned_job_or_error ? undef : $cloned_job_or_error) if $is_ok;
 
     # retry a certain number of times, maybe the transaction failed due to a conflict
@@ -53,4 +56,20 @@ sub _restart_job ($minion_job, @args) {
     $minion_job->retry({delay => restart_delay});
 }
 
+sub _init_amqp_plugin ($app) {
+    return undef unless $app->config->{amqp}->{enabled};
+    $app->plugin('AMQP');    # Needs to be loaded again from forked process
+    Mojo::IOLoop->singleton->one_tick;
+}
+
+sub _wait_for_event_publish ($app) {
+    return undef unless $app->config->{amqp}->{enabled};
+    OpenQA::Events->singleton->once(
+        'amqp_handled',
+        sub {
+            Mojo::IOLoop->singleton->stop;
+        });
+    Mojo::IOLoop->singleton->start;
+}
+
 1;

lib/OpenQA/WebAPI/Plugin/AMQP.pm

@@ -28,6 +28,8 @@ sub new ($class, @args) {
 sub register ($self, $app, @args) {
     $self->{app} = $app;
     $self->{config} = $app->config;
+    my $config = $self->{config}->{amqp};
+    $config->{enabled} = 1;    # Needed for reloading the plugin later in the forked process
     Mojo::IOLoop->singleton->next_tick(
         sub {
             # register for events
@@ -68,18 +70,26 @@ sub publish_amqp ($self, $topic, $event_data, $headers = {}, $remaining_attempts
 
     $remaining_attempts //= $config->{publish_attempts};
     $retry_delay //= $config->{publish_retry_delay};
-    $publisher->publish_p($event_data, $headers, routing_key => $topic)->then(sub { log_debug "$topic published" })
-      ->catch(
+    $publisher->publish_p($event_data, $headers, routing_key => $topic)->then(
+        sub {
+            log_debug "$topic published";
+            OpenQA::Events->singleton->emit('amqp_handled');
+        }
+    )->catch(
         sub ($error) {
             my $left = looks_like_number $remaining_attempts && $remaining_attempts > 1 ? $remaining_attempts - 1 : 0;
             my $delay = $retry_delay * $config->{publish_retry_delay_factor};
             my ($event_id, $job_id) = ($event_data->{id} // 'none', $event_data->{job_id});
             my $additional_info = $job_id ? ", job ID: $job_id" : '';
             my $log_msg = "Publishing $topic failed: $error (event ID: $event_id$additional_info, $left attempts left)";
-            return log_error $log_msg unless $left;
             my $retry_function = sub ($loop) { $self->publish_amqp($topic, $event_data, $headers, $left, $delay) };
-            log_info $log_msg;
-            Mojo::IOLoop->timer($retry_delay => $retry_function) if $left;
+            if ($left) {
+                log_info $log_msg;
+                Mojo::IOLoop->timer($retry_delay => $retry_function);
+                return;
+            }
+            OpenQA::Events->singleton->emit('amqp_handled');
+            return log_error $log_msg;
         })->finally(sub { undef $publisher });
 }