Decouple Stager and CloudController

CC uses StagerClient

Change-Id: I76e60c142af0fece68a34354f8919626e16145b9

cloud_controller/Gemfile

@@ -13,7 +13,7 @@ gem 'cf-uaa-client', '>= 0.0.8'
 
 # For queuing staging tasks
 gem 'em-hiredis'
-gem 'vcap_stager', '~> 0.1.14'
+gem 'stager-client', '~> 0.0.2'
 
 # Databases
 gem 'sqlite3'

cloud_controller/Gemfile.lock

@@ -121,6 +121,10 @@ GEM
       rack (~> 1.1)
       tilt (>= 1.2.2, < 2.0)
     sqlite3 (1.3.3)
+    stager-client (0.0.2)
+      eventmachine
+      nats
+      yajl-ruby
     thin (1.3.1)
       daemons (>= 1.0.9)
       eventmachine (>= 0.12.6)
@@ -138,7 +142,6 @@ GEM
       thin (~> 1.3.1)
       yajl-ruby (~> 0.8.3)
     vcap_logging (1.0.1)
-    vcap_stager (0.1.14)
     vcap_staging (0.1.55)
       nokogiri (>= 1.4.4)
       rake
@@ -175,10 +178,10 @@ DEPENDENCIES
   ruby-hmac (~> 0.4.0)
   sinatra
   sqlite3
+  stager-client (~> 0.0.2)
   thin
   uuidtools (~> 2.1.2)
   vcap_common
   vcap_logging
-  vcap_stager (~> 0.1.14)
   vcap_staging (~> 0.1.55)
   yajl-ruby (~> 0.8.3)

cloud_controller/app/controllers/apps_controller.rb

@@ -1,4 +1,4 @@
-require 'staging_task_manager'
+require "vcap/stager/client"
 
 class AppsController < ApplicationController
   before_filter :require_user, :except => [:download_staged]
@@ -219,25 +219,36 @@ def files
   private
 
   def stage_app(app)
-    task_mgr = StagingTaskManager.new(:logger  => CloudController.logger,
-                                      :timeout => AppConfig[:staging][:max_staging_runtime])
     dl_uri = StagingController.download_app_uri(app)
     ul_hdl = StagingController.create_upload(app)
 
-    result = task_mgr.run_staging_task(app, dl_uri, ul_hdl.upload_uri)
+    client = VCAP::Stager::Client::FiberAware.new(NATS.client,
+                                                  AppConfig[:staging][:queue])
 
-    # Update run count to be consistent with previous staging code
-    if result.was_success?
-      CloudController.logger.debug("Staging task for app_id=#{app.id} succeded.", :tags => [:staging])
-      CloudController.logger.debug1("Details: #{result.task_log}", :tags => [:staging])
+    request = {
+      "app_id"       => app.id,
+      "properties"   => app.staging_task_properties,
+      "download_uri" => dl_uri,
+      "upload_uri"   => ul_hdl.upload_uri,
+    }
+
+    begin
+      result = client.stage(request, AppConfig[:staging][:max_staging_runtime])
+      StagingTaskLog.new(app.id, result["task_log"]).save
+    rescue VCAP::Stager::Client::Error => e
+      result = { "error" => e.to_s }
+    end
+
+    if result["error"]
+      CloudController.logger.warn("Staging for app_id=#{app.id} failed")
+      CloudController.logger.warn("Error: #{result["error"]}")
+      raise CloudError.new(CloudError::APP_STAGING_ERROR, result["error"])
+    else
+      # Update run count to be consistent with previous staging code
+      CloudController.logger.info("Staging for app_id=#{app.id} succeeded")
       app.update_staged_package(ul_hdl.upload_path)
       app.package_state = 'STAGED'
       app.update_run_count()
-    else
-      CloudController.logger.warn("Staging task for app_id=#{app.id} failed: #{result.error}",
-                                  :tags => [:staging])
-      CloudController.logger.debug1("Details: #{result.task_log}", :tags => [:staging])
-      raise CloudError.new(CloudError::APP_STAGING_ERROR, result.error.to_s)
     end
 
   rescue => e

cloud_controller/app/models/app_manager.rb

@@ -1,5 +1,3 @@
-require 'vcap/stager'
-
 class AppManager
   attr_reader :app
 

cloud_controller/app/models/staging_task_log.rb

@@ -17,7 +17,7 @@ def fetch_fibered(app_id, redis=nil, timeout=5)
       get_def = redis.get(key)
       get_def.timeout(timeout)
       get_def.errback do |e|
-        e = VCAP::Stager::StagingTimeoutError.new("Timed out fetching result") if e == nil
+        e = CloudError.new(CloudError::SYSTEM_ERROR, "Timed out fetching result") if e == nil
         logger.error("Failed fetching result for key '#{key}': #{e}")
         logger.error(e)
         f.resume([false, e])

cloud_controller/config/appconfig.rb

@@ -205,6 +205,8 @@
   exit 1
 end
 
+AppConfig[:staging][:queue] ||= "staging"
+
 if AppConfig[:bootstrap_users]
   unless AppConfig[:bootstrap_users].kind_of?(Array)
     $stderr.puts "List of bootstrap users must be an array"

cloud_controller/spec/models/staging_task_log_spec.rb

@@ -58,7 +58,7 @@
       Fiber.new do
         expect do
           res = StagingTaskLog.fetch_fibered(@task_id, @redis_mock)
-        end.to raise_error(VCAP::Stager::TaskError)
+        end.to raise_error(CloudError)
       end.resume
       @deferrable_mock.fail(nil)
     end

stager/Gemfile

@@ -9,6 +9,7 @@ gem 'vcap_common', '~> 1.0.8'
 gem 'vcap_logging', '>= 0.1.3'
 gem 'vcap_staging', '~> 0.1.55'
 gem 'vcap-concurrency', '~> 0.0.1'
+gem 'stager-client', '~> 0.0.2'
 
 group :test do
   gem 'rspec'

stager/Gemfile.lock

@@ -30,6 +30,10 @@ GEM
     sinatra (1.2.6)
       rack (~> 1.1)
       tilt (>= 1.2.2, < 2.0)
+    stager-client (0.0.2)
+      eventmachine
+      nats
+      yajl-ruby
     thin (1.3.1)
       daemons (>= 1.0.9)
       eventmachine (>= 0.12.6)
@@ -67,6 +71,7 @@ DEPENDENCIES
   rake
   rspec
   sinatra
+  stager-client (~> 0.0.2)
   vcap-concurrency (~> 0.0.1)
   vcap_common (~> 1.0.8)
   vcap_logging (>= 0.1.3)

stager/lib/vcap/stager/server.rb

@@ -61,7 +61,10 @@ def shutdown
     @shutdown_thread = Thread.new do
       # Blocks until all threads have finished
       @thread_pool.shutdown
-      EM.next_tick { EM.stop }
+      EM.next_tick do
+        EM.stop
+        @logger.info("Shutdown complete")
+      end
     end
   end
 
@@ -88,10 +91,13 @@ def install_signal_handlers
 
   def setup_subscriptions
     @config[:queues].each do |q|
-      sq = "vcap.stager.#{q}"
-      @sids << @nats_conn.subscribe(sq, :queue => sq) do |msg, reply_to|
-        add_task(msg)
+      @sids << @nats_conn.subscribe(q, :queue => q) do |msg, reply_to|
+        @thread_pool.enqueue { execute_request(msg, reply_to) }
+
+        @logger.info("Enqueued request #{msg}")
       end
+
+      @logger.info("Subscribed to #{q}")
     end
   end
 
@@ -101,46 +107,47 @@ def teardown_subscriptions
     @sids = []
   end
 
-  def add_task(task_msg)
+  def execute_request(encoded_request, reply_to)
     begin
-      @logger.debug("Decoding task '#{task_msg}'")
+      @logger.debug("Decoding request '#{encoded_request}'")
 
-      request = Yajl::Parser.parse(task_msg)
+      request = Yajl::Parser.parse(encoded_request)
     rescue => e
-      @logger.warn("Failed decoding '#{task_msg}': #{e}")
+      @logger.warn("Failed decoding '#{encoded_request}': #{e}")
       @logger.warn(e)
       return
     end
 
-    @thread_pool.enqueue { execute_request(request) }
-
-    @logger.info("Enqueued request #{request}")
-
-    nil
-  end
-
-  def execute_request(request)
     task = VCAP::Stager::Task.new(request, @task_config)
 
     result = nil
     begin
       task.perform
 
-      result = VCAP::Stager::TaskResult.new(task.task_id, task.log)
+      result = {
+        "task_id"  => task.task_id,
+        "task_log" => task.log,
+      }
 
       @logger.info("Task #{task.task_id} succeeded")
     rescue VCAP::Stager::TaskError => te
       @logger.warn("Task #{task.task_id} failed: #{te}")
 
-      result = VCAP::Stager::TaskResult.new(task.task_id, task.log, te)
+      result = {
+        "task_id"  => task.task_id,
+        "task_log" => task.log,
+        "error"    => te.to_s,
+      }
     rescue Exception => e
       @logger.error("Unexpected exception: #{e}")
       @logger.error(e)
 
       raise e
     end
 
-    EM.next_tick { @nats_conn.publish(request["notify_subj"], result.encode) }
+    encoded_result = Yajl::Encoder.encode(result)
+
+    EM.next_tick { @nats_conn.publish(reply_to, encoded_result) }
 
     nil
   end

stager/lib/vcap/stager/task.rb

@@ -142,7 +142,7 @@ def stage_app(src_dir, dst_dir, task_logger)
       if res[:timed_out]
         emsg = "Staging timed out after #{@max_staging_duration} seconds."
       else
-        emsg = "Staging plugin failed: #{res[:stdout]}"
+        emsg = "Staging plugin failed: #{res[:stderr]}"
       end
 
       task_logger.warn(emsg)