Merge branch 'next'

hainesr · hainesr · commit 0250ff8e7fcc · 2014-10-31T13:40:20.000Z
* next:
  Factor out the interactions handler in the worker code.
  Document the new connection configuration option.
  [TAV-522] Configure number of network error retries.
  [TAV-522] Handle network level errors (retry).
  Fix the "fail in post run callback" worker test.
  Factor out run creation in the worker.
  Use the latest t2-server client library.
diff --git a/README.rdoc b/README.rdoc
@@ -313,7 +313,19 @@ that make use of interactions (so you have users watching the workflows
 running) then it should probably be set lower.
 
 There are a number of options for configuring the connection to the Taverna
-Server. These are actually provided by the underlying
+Server. The first specifies how many times an initial connection to Taverna
+Server will be retried if there are any low level network errors.
+
+  config.server_connection_error_retries = 5
+
+These can be events such as timeouts, broken connections, refused connections
+or anything else out of our control. If such an error is detected Taverna
+Player will back off for the amount of time specified by
+<tt>config.server_retry_interval</tt> before retrying. If the maximum number of
+retries is reached and there is still an error then the run will fail; the
+network error message will be recorded as the reason for the run's failure.
+
+The rest of the connection options are actually provided by the underlying
 {t2-server client library}[https://rubygems.org/gems/t2-server] and surfaced
 here for extra control. They are
 {documented in more detail}[http://mygrid.github.io/t2-server-gem/] elsewhere
diff --git a/config/locales/en.yml b/config/locales/en.yml
@@ -17,6 +17,7 @@ en:
       pre-callback: "Running pre-run tasks"
       connect: "Connecting to Taverna Server"
       full: "Server full - please wait; run will start soon"
+      network-error: "Network error - please wait; retrying"
       initialized: "Initializing new workflow run"
       inputs: "Uploading run inputs"
       start: "Starting run"
diff --git a/lib/generators/templates/server_initializer.rb b/lib/generators/templates/server_initializer.rb
@@ -13,6 +13,7 @@
   config.server_retry_interval = 10
 
   # Taverna Server connection configuration.
+  #config.server_connection_error_retries = 5
   #config.server_connection[:verify_peer] = true
   #config.server_connection[:ca_file] = "/etc/certs/my-cert.crt"
   #config.server_connection[:ca_path] = "/etc/certs/"
diff --git a/lib/taverna-player.rb b/lib/taverna-player.rb
@@ -121,6 +121,10 @@ def self.port_renderers
   mattr_accessor :server_retry_interval
   @@server_retry_interval = 10
 
+  # Number of times to retry on a low-level connection error.
+  mattr_accessor :server_connection_error_retries
+  @@server_connection_error_retries = 5
+
   # Taverna Server connection parameters
   mattr_accessor :server_connection
   @@server_connection = T2Server::DefaultConnectionParameters.new
diff --git a/lib/taverna_player/worker.rb b/lib/taverna_player/worker.rb
@@ -50,23 +50,11 @@ def perform
 
       begin
         server = T2Server::Server.new(@server, conn_params)
-        wkf = File.read(@workflow)
+        workflow = File.read(@workflow)
+        run = create_run(server, workflow, credentials)
 
-        # Try and create the run bearing in mind that the server might be at
-        # the limit of runs that it can hold at once.
-        begin
-          run = server.create_run(wkf, credentials)
-        rescue T2Server::ServerAtCapacityError
-          status_message("full")
-
-          if cancelled?
-            cancel
-            return
-          end
-
-          sleep(TavernaPlayer.server_retry_interval)
-          retry
-        end
+        # If run is nil here then we could have failed or have been cancelled.
+        return if run.nil?
 
         status_message("initialized")
 
@@ -121,65 +109,7 @@ def perform
             return
           end
 
-          run.notifications(:requests).each do |note|
-            if @run.has_parent?
-              next if note.has_reply? || note.is_notification?
-              int = Interaction.find_by_run_id_and_serial(@run.parent_id, note.serial)
-              new_int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
-              if new_int.new_record?
-                note.reply(int.feed_reply, int.data)
-                new_int.displayed = true
-                new_int.replied = true
-                new_int.feed_reply = int.feed_reply
-                new_int.data = int.data
-                new_int.save
-              end
-            else
-              int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
-
-              # Need to catch this here in case some other process has replied.
-              if note.has_reply? && !int.replied?
-                int.replied = true
-                int.save
-              end
-
-              unless int.replied?
-                if int.page.blank?
-                  page = server.read(note.uri, "text/html", credentials)
-
-                  INTERACTION_REGEX.match(page) do
-                    page_uri = $1
-
-                    if page_uri.starts_with?(server.uri.to_s)
-                      page = server.read(URI.parse(page_uri), "text/html", credentials)
-                      int.page = page.gsub("#{run.interactions_uri.to_s}/pmrpc.js",
-                        "/assets/taverna_player/application.js")
-                    else
-                      int.page_uri = page_uri
-                    end
-                  end
-                end
-
-                if note.is_notification? && !int.new_record?
-                  int.replied = true
-                end
-
-                if int.data.blank?
-                  int.data = note.input_data.force_encoding("UTF-8")
-                end
-
-                if !int.feed_reply.blank? && !int.data.blank?
-                  note.reply(int.feed_reply, int.data)
-
-                  int.replied = true
-                end
-
-                int.save
-              end
-
-              waiting = true unless int.replied?
-            end
-          end
+          waiting = interactions(run, credentials)
 
           status_message(waiting ? "interact" : "running")
         end
@@ -220,6 +150,108 @@ def server_credentials
       T2Server::HttpBasic.new(user, pass)
     end
 
+    # Try and create the run bearing in mind that the server might be at
+    # the limit of runs that it can hold at once.
+    def create_run(server, workflow, credentials)
+      retries ||= TavernaPlayer.server_connection_error_retries
+      server.create_run(workflow, credentials)
+    rescue T2Server::ServerAtCapacityError
+      status_message("full")
+
+      if cancelled?
+        cancel
+        return
+      end
+
+      sleep(TavernaPlayer.server_retry_interval)
+      retry
+    rescue T2Server::ConnectionError => ce
+      status_message("network-error")
+
+      if cancelled?
+        cancel
+        return
+      end
+
+      sleep(TavernaPlayer.server_retry_interval)
+      unless retries.zero?
+        retries -= 1
+        retry
+      end
+
+      # If we're out of retries, fail the run.
+      failed(ce)
+    end
+
+    def interactions(run, credentials)
+      wait = false
+
+      run.notifications(:requests).each do |note|
+        if @run.has_parent?
+          next if note.has_reply? || note.is_notification?
+
+          int = Interaction.find_by_run_id_and_serial(@run.parent_id, note.serial)
+          new_int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
+
+          if new_int.new_record?
+            note.reply(int.feed_reply, int.data)
+            new_int.displayed = true
+            new_int.replied = true
+            new_int.feed_reply = int.feed_reply
+            new_int.data = int.data
+            new_int.save
+          end
+        else
+          int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
+
+          # Need to catch this here in case some other process has replied.
+          if note.has_reply? && !int.replied?
+            int.replied = true
+            int.save
+          end
+
+          unless int.replied?
+            if int.page.blank?
+              server = run.server
+              page = server.read(note.uri, "text/html", credentials)
+
+              INTERACTION_REGEX.match(page) do
+                page_uri = $1
+
+                if page_uri.starts_with?(server.uri.to_s)
+                  page = server.read(URI.parse(page_uri), "text/html", credentials)
+                  int.page = page.gsub("#{run.interactions_uri.to_s}/pmrpc.js",
+                  "/assets/taverna_player/application.js")
+                else
+                  int.page_uri = page_uri
+                end
+              end
+            end
+
+            # If this is a pure notification that we've already seen then
+            # set it as replied as we don't want it blocking a proper
+            # interaction.
+            int.replied = true if note.is_notification? && !int.new_record?
+
+            if int.data.blank?
+              int.data = note.input_data.force_encoding("UTF-8")
+            end
+
+            if !int.feed_reply.blank? && !int.data.blank?
+              note.reply(int.feed_reply, int.data)
+              int.replied = true
+            end
+
+            int.save
+          end
+
+          wait = true unless int.replied?
+        end
+      end
+
+      wait
+    end
+
     # Run the specified callback and return false on error so that we know to
     # return out of the worker code completely.
     def run_callback(cb, message)
diff --git a/taverna_player.gemspec b/taverna_player.gemspec
@@ -37,7 +37,7 @@ Gem::Specification.new do |s|
   s.add_dependency "jquery-rails", "~> 3.0"
   s.add_dependency "paperclip", "~> 4.1"
   s.add_dependency "taverna-t2flow", "~> 0.5.1"
-  s.add_dependency "t2-server", "~> 1.1"
+  s.add_dependency "t2-server", "~> 1.2"
   s.add_dependency "delayed_job_active_record", "~> 4.0"
   s.add_dependency "daemons", "~> 1.1.9"
   s.add_dependency "rubyzip", "~> 1.1.4"
diff --git a/test/dummy/config/locales/taverna_player.en.yml b/test/dummy/config/locales/taverna_player.en.yml
@@ -17,6 +17,7 @@ en:
       pre-callback: "Running pre-run tasks"
       connect: "Connecting to Taverna Server"
       full: "Server full - please wait; run will start soon"
+      network-error: "Network error - please wait; retrying"
       initialized: "Initializing new workflow run"
       inputs: "Uploading run inputs"
       start: "Starting run"
diff --git a/test/unit/taverna_player/worker_test.rb b/test/unit/taverna_player/worker_test.rb
@@ -28,6 +28,7 @@ class WorkerTest < ActiveSupport::TestCase
       config.server_password = "taverna"
       config.server_poll_interval = 0
       config.server_retry_interval = 0
+      config.server_connection_error_retries = 2
       config.pre_run_callback = @noop_callback
       config.post_run_callback = @noop_callback
       config.run_cancelled_callback = @noop_callback
@@ -38,6 +39,7 @@ class WorkerTest < ActiveSupport::TestCase
 
     # Stuff we can't test yet in TavernaPlayer::Worker.
     flexmock(TavernaPlayer::Worker).new_instances do |w|
+      w.should_receive(:interactions).and_return(false)
       w.should_receive(:download_outputs).and_return_undefined
       w.should_receive(:process_outputs).and_return([])
     end
@@ -101,7 +103,6 @@ class WorkerTest < ActiveSupport::TestCase
       r.should_receive(:name=).once.and_return(true)
       r.should_receive(:start).twice.and_return(false, true)
       r.should_receive(:start_time).and_return(Time.now)
-      r.should_receive(:notifications).and_return([])
       r.should_receive(:finish_time).and_return(Time.now)
       r.should_receive(:log).once.and_return(0)
       r.should_receive(:delete).and_return_undefined
@@ -281,6 +282,25 @@ class WorkerTest < ActiveSupport::TestCase
     # Set a failing post_run callback
     TavernaPlayer.post_run_callback = Proc.new { raise RuntimeError }
 
+    # Stub the creation of a run on a Taverna Server.
+    flexmock(T2Server::Server).new_instances do |s|
+      s.should_receive(:initialize_run).once.
+        and_return(URI.parse("http://localhost/run/01"))
+    end
+
+    # Stub the Taverna Server run calls.
+    flexmock(T2Server::Run).new_instances do |r|
+      r.should_receive(:status).times(3).and_return(:initialized, :running, :finished)
+      r.should_receive(:create_time).and_return(Time.now)
+      r.should_receive(:add_password_credential).and_return(true)
+      r.should_receive(:name=).once.and_return(true)
+      r.should_receive(:start).once.and_return(true)
+      r.should_receive(:start_time).and_return(Time.now)
+      r.should_receive(:finish_time).and_return(Time.now)
+      r.should_receive(:log).once.and_return(0)
+      r.should_receive(:delete).and_return_undefined
+    end
+
     assert_equal :pending, @run.state, "Initial run state not ':pending'"
 
     @worker.perform
@@ -302,4 +322,50 @@ class WorkerTest < ActiveSupport::TestCase
 
     assert_equal :timeout, @run.state, "Final run state not ':timeout'"
   end
+
+  test "network error with recovery" do
+    # Stub the creation of a run on a Taverna Server with a network error
+    # first.
+    flexmock(T2Server::Server).new_instances do |s|
+      s.should_receive(:initialize_run).twice.
+        and_raise(T2Server::ConnectionError, Timeout::Error.new).
+        and_return(URI.parse("http://localhost/run/01"))
+    end
+
+    # Stub the Taverna Server run calls.
+    flexmock(T2Server::Run).new_instances do |r|
+      r.should_receive(:status).times(3).and_return(:initialized, :running, :finished)
+      r.should_receive(:create_time).and_return(Time.now)
+      r.should_receive(:add_password_credential).and_return(true)
+      r.should_receive(:name=).once.and_return(true)
+      r.should_receive(:start).once.and_return(true)
+      r.should_receive(:start_time).and_return(Time.now)
+      r.should_receive(:finish_time).and_return(Time.now)
+      r.should_receive(:log).once.and_return(0)
+      r.should_receive(:delete).and_return_undefined
+    end
+
+    assert_equal :pending, @run.state, "Initial run state not ':pending'"
+
+    @worker.perform
+
+    assert_equal :finished, @run.state, "Final run state not ':finished'"
+  end
+
+  test "network error with no recovery" do
+    # Stub the creation of a run on a Taverna Server with a network error
+    # first.
+    flexmock(T2Server::Server).new_instances do |s|
+      # Connection retries are set to 2 so this should be called 3 times.
+      s.should_receive(:initialize_run).times(3).
+        and_raise(T2Server::ConnectionError, Timeout::Error.new)
+    end
+
+    assert_equal :pending, @run.state, "Initial run state not ':pending'"
+
+    @worker.perform
+
+    assert_equal :failed, @run.state, "Final run state not ':finished'"
+  end
+
 end
