Final stager changes

- Rewrite stager to use NATS queue groups instead of Resque.
- Add support for staging uploads to nginx config
- Integrate CC with stager
- Add support to bin/vcap for controlling stager/redis instance

Test plan:
* Ran BVTs locally with and w/out the new stager, and with and w/out nginx support.
* Ran BVTs against my deployment with and without the new stager.
* Ran unit tests for stager and CC.

Change-Id: I3c85d7de58f518d6111ca3dbf6c55fc532350d9d

bin/config/redis-server.conf

@@ -0,0 +1,3 @@
+bind 0.0.0.0
+port 5454
+loglevel debug

bin/stager

@@ -0,0 +1,4 @@
+#!/usr/bin/env ruby
+# Copyright (c) 2009-2011 VMware, Inc.
+
+exec(File.expand_path("../../stager/bin/stager", __FILE__), *ARGV)

bin/vcap

@@ -79,7 +79,7 @@ class Component
   end
 
   def pid_file
-    configuration["pid"] || raise("#{@configuration_path} does not specify location of pid file")
+    configuration["pid"] || configuration['pid_filename'] || raise("#{@configuration_path} does not specify location of pid file")
   end
 
   def log_file?
@@ -244,10 +244,64 @@ class NatsServer
   end
 end
 
+class RedisServer
+  DEFAULT_CONFIG_PATH = File.expand_path('../config/redis-server.conf', __FILE__)
+
+  def initialize(pid_filename, config_path=DEFAULT_CONFIG_PATH)
+    @config_path  = config_path
+    @pid_filename = pid_filename
+    @redis_path   = `which redis-server`.chomp
+    unless $? == 0
+      STDERR.puts "Could not find redis-server, exiting.".red
+      exit 1
+    end
+    @pid = read_pidfile
+  end
+
+  def start
+    return if running?
+    @pid = fork do
+      log_file = File.join(TMP, 'redis-server.log')
+      stdout = File.open(log_file, 'a')
+      STDOUT.reopen(stdout)
+      stderr = File.open(log_file, 'a')
+      STDERR.reopen(stderr)
+      exec("#{@redis_path} #{@config_path}")
+    end
+    Process.detach(@pid)
+    File.open(@pid_filename, 'w+') {|f| f.write(@pid) }
+  end
+
+  def stop
+    return unless running?
+    Process.kill('TERM', @pid)
+    @pid = nil
+  end
+
+  private
+
+  def read_pidfile
+    if File.exist?(@pid_filename)
+      File.read(@pid_filename).chomp.to_i
+    else
+      nil
+    end
+  end
+
+  def running?
+    return false unless @pid
+    File.exist?(File.join('/proc', @pid.to_s))
+  end
+end
+
+
 module Run
+  DEFAULT_REDIS_PIDFILE = File.join(TMP, 'redis-server.pid')
+
   def self.start_init
     nats_server = NatsServer.new
     nats_server.start_server
+    RedisServer.new(DEFAULT_REDIS_PIDFILE).start
   end
 
   def self.start(args)
@@ -259,6 +313,7 @@ module Run
     # Only process this if no one else running..
     running_components = components([]).select {|c| c.running?}.map{|c| c.name }
     return unless running_components.empty?
+    RedisServer.new(DEFAULT_REDIS_PIDFILE).stop
     nats_server = NatsServer.new
     return unless nats_server.is_running?
     nats_server.kill_server
@@ -378,7 +433,7 @@ module Run
   private
 
   def self.core
-    %w(router cloud_controller dea health_manager)
+    %w(router cloud_controller dea health_manager stager)
   end
 
   def self.services

cloud_controller/Gemfile

@@ -10,9 +10,11 @@ gem 'logging', '>= 1.5.0'
 # VCAP common components
 gem 'vcap_common', :require => ['vcap/common', 'vcap/component'], :path => '../common'
 gem 'vcap_logging', :require => ['vcap/logging']
+gem 'vcap_staging', '= 0.1.2'
 
-# XXX - Vendor once working
-gem 'vcap_staging'
+# For queuing staging tasks
+gem 'em-hiredis'
+gem 'vcap_stager', '= 0.1.3'
 
 # Databases
 gem 'sqlite3'
@@ -45,6 +47,7 @@ gem 'bcrypt-ruby', '>= 2.1.4'
 gem 'ruby-hmac', :require => 'hmac-sha1'
 gem 'SystemTimer', :platforms => :mri_18
 gem 'uuidtools'
+gem 'rest-client', '= 1.6.7'
 
 # rspec-rails is outside the 'test' group in order to consistently provide Rake tasks.
 gem 'rspec-rails', '>= 2.4.1'

cloud_controller/Gemfile.lock

@@ -48,6 +48,8 @@ GEM
       builder (>= 2.1.2)
     daemons (1.1.2)
     diff-lcs (1.1.2)
+    em-hiredis (0.1.0)
+      hiredis (~> 0.3.0)
     em-http-request (1.0.0.beta.3)
       addressable (>= 2.2.3)
       em-socksify
@@ -60,6 +62,7 @@ GEM
     erubis (2.6.6)
       abstract (>= 1.0.0)
     eventmachine (0.12.10)
+    hiredis (0.3.2)
     http_parser.rb (0.5.1)
     i18n (0.5.0)
     json_pure (1.5.1)
@@ -103,6 +106,8 @@ GEM
       thor (~> 0.14.4)
     rake (0.8.7)
     rcov (0.9.9)
+    rest-client (1.6.7)
+      mime-types (>= 1.16)
     rspec (2.5.0)
       rspec-core (~> 2.5.0)
       rspec-expectations (~> 2.5.0)
@@ -132,7 +137,13 @@ GEM
     tzinfo (0.3.26)
     uuidtools (2.1.2)
     vcap_logging (0.1.0)
-    vcap_staging (0.1.0)
+    vcap_stager (0.1.3)
+    vcap_staging (0.1.2)
+      nokogiri (>= 1.4.4)
+      rake
+      rspec
+      vcap_common
+      yajl-ruby (>= 0.7.9)
     yajl-ruby (0.8.2)
 
 PLATFORMS
@@ -142,6 +153,7 @@ DEPENDENCIES
   SystemTimer
   bcrypt-ruby (>= 2.1.4)
   ci_reporter
+  em-hiredis
   em-http-request (~> 1.0.0.beta.3)
   em-redis
   eventmachine (~> 0.12.10)
@@ -154,6 +166,7 @@ DEPENDENCIES
   rack-fiber_pool
   rails (~> 3.0.5)
   rcov
+  rest-client (= 1.6.7)
   rspec (>= 2.4.0)
   rspec-rails (>= 2.4.1)
   ruby-hmac
@@ -163,5 +176,6 @@ DEPENDENCIES
   uuidtools
   vcap_common!
   vcap_logging
-  vcap_staging
+  vcap_stager (= 0.1.3)
+  vcap_staging (= 0.1.2)
   yajl-ruby (>= 0.7.9)

cloud_controller/app/controllers/apps_controller.rb

@@ -1,3 +1,5 @@
+require 'staging_task_manager'
+
 class AppsController < ApplicationController
   before_filter :require_user, :except => [:download_staged]
   before_filter :find_app_by_name, :except => [:create, :list, :download_staged]
@@ -83,7 +85,7 @@ def upload
   end
 
   def download
-    path = @app.package_path
+    path = @app.unstaged_package_path
     if path && File.exists?(path)
       send_file path
     else
@@ -124,7 +126,13 @@ def start_update
     error_on_lock_mismatch(@app)
     @app.lock_version += 1
     manager = AppManager.new(@app)
-    manager.stage if @app.needs_staging?
+    if @app.needs_staging?
+      if user.uses_new_stager?
+        stage_app(@app)
+      else
+        manager.stage
+      end
+    end
     manager.stop_all
     manager.started
     render :nothing => true, :status => 204
@@ -164,6 +172,18 @@ def check_update
 
   # GET /apps/:name/instances/:instance_id/files/:path'
   def files
+    # XXX - Yuck. This will have to do until we update VMC with a real
+    #       way to fetch staging logs.
+    if user.uses_new_stager? && (params[:path] == 'logs/staging.log')
+      log = StagingTaskLog.fetch_fibered(@app.id)
+      if log
+        render :text => log.task_log
+      else
+        render :nothing => true, :status => 404
+      end
+      return
+    end
+
     # will Fiber.yield
     url, auth = AppManager.new(@app).get_file_url(params[:instance_id], params[:path])
     raise CloudError.new(CloudError::APP_FILE_ERROR, params[:path] || '/') unless url
@@ -191,6 +211,43 @@ 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)
+
+    # 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])
+      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
+    # It may be the case that upload from the stager will happen sometime in the future.
+    # Mark the upload as completed so that any upload that occurs in the future will fail.
+    if ul_hdl
+      StagingController.complete_upload(ul_hdl)
+      FileUtils.rm_f(ul_hdl.upload_path)
+    end
+    app.package_state = 'FAILED'
+    app.update_run_count()
+    raise e
+
+  ensure
+    app.save!
+  end
+
   def find_app_by_name
     # XXX - What do we want semantics to be like for multiple apps w/ same name (possible w/ contribs)
     @app = user.apps_owned.find_by_name(params[:name])
@@ -208,7 +265,6 @@ def find_app_by_name
   # App from the request params and makes the necessary AppManager calls.
   def update_app_from_params(app)
     CloudController.logger.debug "app: #{app.id || "nil"} update_from_parms"
-
     error_on_lock_mismatch(app)
     app.lock_version += 1
 
@@ -247,7 +303,14 @@ def update_app_from_params(app)
 
     # Process any changes that require action on out part here.
     manager = AppManager.new(app)
-    manager.stage if app.needs_staging?
+
+    if app.needs_staging?
+      if user.uses_new_stager?
+        stage_app(app)
+      else
+        manager.stage
+      end
+    end
 
     if changed.include?('state')
       if app.stopped?
@@ -405,4 +468,6 @@ def check_has_capacity_for?(app, previous_state)
       raise CloudError.new(CloudError::ACCOUNT_NOT_ENOUGH_MEMORY, "#{mem_quota}M")
     end
   end
+
+
 end