Cloud Foundry中Stager组件的源码分析

本文主要是介绍Cloud Foundry中Stager组件的源码分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Cloud Foundry中有一个组件，名为Stager，它主要负责的工作就是将用户部署进Cloud Foundry的源代码打包成一个DEA可以解压执行的droplet。

关于droplet的制作，Cloud Foundry v1中一个完整的流程为：

用户将应用源代码上传至Cloud Controller；
Cloud Controller通过NATS发送请求至Stager，要求制作dropet；
Stager从Cloud Controller下载压缩后的应用源码，并解压；
Stager将解压后的应用源码，添加运行容器以及启动终止脚本；
压缩成droplet形式上传至Cloud Foundry。

Stager制作droplet总流程实现

现在从制作一个droplet的流程将分析Cloud Foundry中Stager的源码。

从Stager接收到请求，则开始制作droplet。而Stager能接收到Cloud Controller发送来的stage请求，是由于Stager订阅了主题为“staging”的消息，/stager/lib/vcap/server.rb中代码如下：

  def setup_subscriptions@config[:queues].each do |q|@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}")endend

由以上代码可知，订阅了@config[:queues]中的主题后，关于该主题消息的发布，被Stager接收到后，Stager执行execute_request方法执行接收到的消息msg。在这里的实现，还借助了@thread_pool，该变量是创建了一个线程池，将执行结果加入线程池。

以下进入/stager/lib/vcap/server.rb的execute_request方法：

  def execute_request(encoded_request, reply_to)beginrequest = Yajl::Parser.parse(encoded_request)rescue => e……returnendtask = VCAP::Stager::Task.new(request, @task_config)result = nilbegintask.perform……endencoded_result = Yajl::Encoder.encode(result)EM.next_tick { @nats_conn.publish(reply_to, encoded_result) }nilend

在该方法中首先对request请求进行解析，获得request对象，然后通过request对象和@task_config产生一个task对象，Task类为VCAP::Stager::Task。其中@task_config中有属性:ruby_path, :ruby_plugin_path, secure_user_manager。创建为task对象之后，执行task.perform。关于perform方法，是整个Stager执行流中最为重要的部分，以下便进入/stager/lib/vcap/stager/task.rb的perform方法中：

  def performworkspace = VCAP::Stager::Workspace.createapp_path = File.join(workspace.root_dir, "app.zip")download_app(app_path)unpack_app(app_path, workspace.unstaged_dir)stage_app(workspace.unstaged_dir, workspace.staged_dir, @task_logger)droplet_path = File.join(workspace.root_dir, "droplet.tgz")create_droplet(workspace.staged_dir, droplet_path)upload_droplet(droplet_path)nilensureworkspace.destroy if workspaceend

该方法中的流程非常清晰，顺序一次是download_app, upack_app, stage_app, create_droplet, upload_app。

首先先进入download_app方法：

  def download_app(app_path)cfg_file = Tempfile.new("curl_dl_config")write_curl_config(@request["download_uri"], cfg_file.path,"output" => app_path)# Show errors but not progress, fail on non-200res = @runner.run_logged("env -u http_proxy -u https_proxy curl -s -S -f -K #{cfg_file.path}")unless res[:status].success?raise VCAP::Stager::TaskError.new("Failed downloading app")endnilensurecfg_file.unlink if cfg_fileend

该方法中较为重要的部分就是如何write_curl_config，以及如何执行脚本命令。在write_curl_config中有一个参数@request["download_uri"]，该参数的意义是用户上传的应用源码存在Cloud Controller处的位置，也是Stager要去下载的应用源码的未知。该参数的产生的流程为：Cloud Controller中的app_controller.rb中，调用stage_app方法，该方法中调用了download_uri方法，并将其作为request的一部分，通过NATS发给了Stager。关于write_curl_config主要实现的是将curl的配置条件写入指定的路径，以便在执行curl命令的时候，可以通过配置文件的读入来方便实现，实现代码为：res = @runner.run_logged("env -u http_proxy -u https_proxy curl -s -S -f -K #{cfg_file.path}")。

当将应用源码下载至指定路径之后，第二步需要做的是将其解压，unpack_app方法则实现了这一点，实现代码为：res = @runner.run_logged("unzip -q #{packed_app_path} -d #{dst_dir}")。

第三步需要做的，也是整个stager最为重要的部分，就是stage_app方法。实现代码如下：

  def stage_app(src_dir, dst_dir, task_logger)plugin_config = {"source_dir"   => src_dir,"dest_dir"     => dst_dir,"environment"  => @request["properties"]}……plugin_config_file = Tempfile.new("plugin_config")StagingPlugin::Config.to_file(plugin_config, plugin_config_file.path)cmd = [@ruby_path, @run_plugin_path,@request["properties"]["framework_info"]["name"],plugin_config_file.path].join(" ")res = @runner.run_logged(cmd,:max_staging_duration => @max_st./bin/catalina.sh runaging_duration)……ensureplugin_config_file.unlink if plugin_config_filereturn_secure_user(secure_user) if secure_userend

在该方法中，首先创建plugin_config这个Hash对象，随后创建plugin_config_file, 又创建了cmd对象，最后通过res = @runner.run_logged(cmd, :max_staging_duration => @max_staging_duration)实现了执行了cmd。现在分析cmd对象：

     cmd = [@ruby_path, @run_plugin_path,@request["properties"]["framework_info"]["name"],plugin_config_file.path].join(" ")

该对象中@ruby_path为Stager组件所在节点出ruby的可执行文件路径；@ruby_plugin_path为运行plugin可执行文件的路径，具体为：/stager/bin/run_plugin；@request["properties"]["framework_info"]["name"]为所需要执行stage操作的应用源码的框架名称，比如，Java_web，spring，Play， Rails3等框架；而 plugin_config_file.path则是做plugin时所需配置文件的路径。

关于cmd对象的执行，将作为本文的一个重要模块，稍后再讲，现在先将讲述一个完整dropet制作流程的实现。

当昨晚stage_app工作之后，也就相当于一个将源码放入了一个server容器中，也做到了实现添加启动终止脚本等，但是这些都是一个完成的文件目录结构存在与文件系统中，为方便管理以及节省空间，Stager会将stage做完后的内容进行压缩，create_droplet则是实现了这一部分的内容，代码如下：

  def create_droplet(staged_dir, droplet_path)cmd = ["cd", staged_dir, "&&", "COPYFILE_DISABLE=true", "tar", "-czf", droplet_path, "*"].join(" ")res = @runner.run_logged(cmd)unless res[:status].success?raise VCAP::Stager::TaskError.new("Failed creating droplet")endend

当创建完droplet这个压缩包之后，Stager还会将这个dropet上传至Cloud Controller的某个路径下，以便之后DEA在启动这个应用的时候，可以从Cloud Controller的文件系统中下载到droplet，并解压启动。以下是upload_app的代码实现：

  def upload_droplet(droplet_path)cfg_file = Tempfile.new("curl_ul_config")write_curl_config(@request["upload_uri"], cfg_file.path, "form" => "upload[droplet]=@#{droplet_path}")res = @runner.run_logged("env -u http_proxy -u https_proxy curl -s -S -f -K #{cfg_file.path}")……nilensurecfg_file.unlink if cfg_fileend

至此的话，Stager所做工作的大致流程，已经全部完成，只是在实现的同时，采用的详细技术，本文并没有一一讲述。

虽然Stager的实现流程已经讲述完毕，但是关于Stager最重要的模块stage_app方法的具体实现，本文还没有进行讲述，本文以下内容会详细讲解这部分内容，并以Java_web和standalone这两种不同框架的应用为案例进行分析。

上文中以及提到，实现stage的功能的代码部分为：

    cmd = [@ruby_path, @run_plugin_path,@request["properties"]["framework_info"]["name"],plugin_config_file.path].join(" ")./bin/catalina.sh runres = @runner.run_logged(cmd,:max_staging_duration => @max_staging_duration)

关于cmd中的各个参数，上文已经分析过，现在更深入的了解@run_plugin_path，该对象指向/stager/bin/run_plugin可执行文件，现在进入该文件中：

ENV['BUNDLE_GEMFILE'] ||= File.expand_path('../../Gemfile', __FILE__)
require 'rubygems'
require 'bundler/setup'
$LOAD_PATH.unshift(File.expand_path('../../lib', __FILE__))
require 'vcap/staging/plugin/common'
unless ARGV.length == 2puts "Usage: run_staging_plugin [plugin name] [plugin config file]"exit 1
end
plugin_name, config_path = ARGV
klass  = StagingPlugin.load_plugin_for(plugin_name)
plugin = klass.from_file(config_path)
plugin.stage_application

该部分的源码主要实现了，提取出cmd命令中的后两个参数，一个为plugin_name，另一个为config_name，并通过这两个参数实现加载plugin以及真正的stage。

首先进入load_plugin_for方法，源码位置为/vcap_staging/lib/vcap/staging/plugin/common.rb：

  def self.load_plugin_for(framework)./bin/catalina.sh runframework = framework.to_splugin_path = File.join(staging_root, framework, 'plugin.rb')require plugin_pathObject.const_get("#{camelize(framework)}Plugin")end

Java_web框架应用的stage流程

首先以Java_web为例在load_plugin_for方法中，首先提取出该应用源码所设置的框架，并通过框架来创建plugin_path对象，然后再实现require该框架目录下的plugin.rb文件，Java_web的应用则需要require的文件为：/vcap-staging/lib/vcap/staging/plugin/java_web/plugin.rb。

通过require了子类的plugin.rb之后，当执行plugin.stage_application的时候，直接进入/vcap-staging/lib/vcap/staging/plugin/java_web/plugin.rb中的stage_application方法，该方法的源码实现为：

  def stage_applicationDir.chdir(destination_directory) docreate_app_directorieswebapp_root = Tomcat.prepare(destination_directory)copy_source_files(webapp_root)web_config_file = File.join(webapp_root, 'WEB-INF/web.xml')unless File.exist? web_config_fileraise "Web application staging failed: web.xml not found"endservices = environment[:services] if environmentcopy_service_drivers(File.join(webapp_root,'../../lib'), services)Tomcat.prepare_insight destination_directory, environment, insight_agent if Tomcat.insight_bound? servicesconfigure_webapp(webapp_root, self.autostaging_template, environment) unless self.skip_staging(webapp_root)create_startup_scriptcreate_stop_scriptendend

在stage_application中，实现步骤为：1.创建相应的目录；2.拷贝源码和相应的所需文件；3.设置配置文件；4.添加启动和终止脚本。

创建相应的目录，包括create_app_directory，在应用的的目标目录下创建log文件夹以及tmp文件夹。

在实现代码 webapp_root = Tomcat.prepare(destination_directory) 时，进入/vcap-staging/lib/vcap/staging/plugin/java_web/tomcat.rb的prepare方法：

  def self.prepare(dir)FileUtils.cp_r(resource_dir, dir)output = %x[cd #{dir}; unzip -q resources/tomcat.zip]raise "Could not unpack Tomcat: #{output}" unless $? == 0webapp_path = File.join(dir, "tomcat", "webapps", "ROOT")server_xml = File.join(dir, "tomcat", "conf", "server.xml")FileUtils.rm_f(server_xml)FileUtils.rm(File.join(dir, "resources", "tomcat.zip"))FileUtils.mv(File.join(dir, "resources", "droplet.yaml"), dir)FileUtils.mkdir_p(webapp_path)webapp_pathend

该方法中，首先从resource文件夹拷贝至dir目录下，随即将resource文件夹中的tomcat.zip文件解压，随后在dir目录下进行一系列的文件操作。最后返回web_app路径。

返回至stage_application方法中的，copy_source_fiiles从一些source文件全部拷贝至webapp_path下，代码为：copy_source_files(webapp_root)；随后检查WEB-INF/web.xml配置文件是否存在，若不存在的话，抛出异常，代码为“

unless File.exist? web_config_fileraise "Web application staging failed: web.xml not found"end

接着将应用所需的一些服务驱动拷贝至指定的lib目录下，代码为：copy_service_drivers(File.join(webapp_root,'../../lib'), services)；随即又实现了对Tomcat的配置，主要是代理的配置： Tomcat.prepare_insight destination_directory, environment, insight_agent if Tomcat.insight_bound? services ；又然后对webapp路木进行了配置，代码为：configure_webapp(webapp_root, self.autostaging_template, environment) unless self.skip_staging(webapp_root)；最后终于到了实现对启动脚本和终止脚本的生成，代码为：

      create_startup_scriptcreate_stop_script

首先来看一下Java_web框架应用的的启动脚本创建，在/vcap-staging/lib/vcap/staging/plugin/common.rb中的create_startup_script方法：

  def create_startup_scriptpath = File.join(destination_directory, 'startup')File.open(path, 'wb') do |f|f.puts startup_scriptendFileUtils.chmod(0500, path)end

在该方法的时候，首先在目标文件中添加一个startup文件，然后再打开该文件，并通过startup_script产生的内容写入startup文件，最后对startup文件进行权限配置。现在进入startup_script方法中，位置为/vcap-staging/lib/vcap/staging/plugin/java_web/plugin.rb：

  def startup_scriptvars = {}vars['CATALINA_OPTS'] = configure_catalina_optsgenerate_startup_script(vars) do<<-JAVA
export CATALINA_OPTS="$CATALINA_OPTS `ruby resources/set_environm./bin/catalina.sh run./bin/catalina.sh run./bin/catalina.sh runent`"
env > env.log./bin/catalina.sh run
PORT=-1
while getopts ":p:" opt; docase $opt inp)PORT=$OPTARG;;esac
done
if [ $PORT -lt 0 ] ; thenecho "Missing or invalid port (-p)"exit 1
fi
ruby resources/generate_server_xml $PORTJAVAendend

在generate_startup_script方法中回天夹start_command，位于/vcap-staging/lib/vcap/staging/plugin/common.rb，运行脚本的添加实现为：

<%= change_directory_for_start %>
<%= start_command %> > ../logs/stdout.log 2> ../logs/stderr.log &

其中change_directory_for_start为： cd tomcat；而start_command为：./bin/catalina.sh run。

至此的话，启动脚本的添加也就完成了，终止的脚本添加的话，也大致一样，主要还是获取应用进程的pid，然后强制杀死该进程，本文就不再具体讲解stop脚本的生成。

讲解到这的话，一个Java_web框架的app应用以及完全stage完成了，在后续会被打包成一个droplet并上传。stage完成之后，自然是需要由DEA来使用的，而DEA正是获取droplet，解压到相应的目录下，最后通过DEA节点提供的运行环境，执行压缩后的droplet中的startup脚本，并最终完全在DEA中启动应用。

Standalone框架应用的stage流程

在Cloud Foundry中，standalone应用被认为是不能被Cloud Foundry识别出框架的所有类型应用。在这种情况下，Cloud Foundry的stager还是会对其进行打包。

在制作standalone应用的droplet时，总的制作流程与Java_web以及其他能被Cloud Foundry识别的框架相比，没有说明区别，但是在stage的时候，会由很大的区别。因为对于standalone的应用，Cloud Foundry的stager不会提供出源码以外的所有运行依赖，所以关于应用的执行的依赖，必须由用户在上传前将其与应用源码捆绑之后在上传。

在识别到需要打包的源码的框架被定义为standalone之后，Stager使用子类StagingPlugin的子类StandalonePlugin来实现stage过程。流程与其类型框架的应用相同，但是具体操作会有一些区别，首先来阅读stage_application的方法：

  def stage_applicationDir.chdir(destination_directory) docreate_app_directoriescopy_source_files#Give everything executable perms, as start command may be a scriptFileUtils.chmod_R(0744, File.join(destination_directory, 'app'))runtime_specific_stagingcreate_startup_scriptcreate_stop_scriptendend

主要的区别在与runtime_specific_staging及之后的方法。runtime_specific_staging方法的功能主要是判断该应用程序的运行环境是否需要ruby，若不需要的话，staging_application继续往下执行；若需要的话，则为应用准备相应的gem包以及安装gem包。

随后在create_startup_script的方法中，首先解析应用的运行时类型，如果是ruby，java，python的话，那就生成相应的启动脚本，如果是其他类型的话，那就直接进入generate_startup_script方法。现在以java运行时为例，分析该过程的实现，代码如下：

  def java_startup_scriptvars = {}java_sys_props = "-Djava.io.tmpdir=$PWD/tmp"vars['JAVA_OPTS'] = "$JAVA_OPTS -Xms#{application_memory}m -Xmx#{application_memory}m #{java_sys_props}"generate_startup_script(vars)end

该方法的实现，创建了vars对象之后，通过将vars参数传递给方法generate_startup_script，来实现启动脚本的生成，代码如下：

  def generate_startup_script(env_vars = {})after_env_before_script = block_given? ? yield : "\n"template = <<-SCRIPT
#!/bin/bash
<%= environment_statements_for(env_vars) %>
<%= after_env_before_script %>
<%= change_directory_for_start %>
<%= start_command %> > ../logs/stdout.log 2> ../logs/stderr.log &
<%= get_launched_process_pid %>
echo "$STARTED" >> ../run.pid
<%= wait_for_launched_process %>SCRIPT# TODO - ERB is pretty irritating when it comes to blank lines, such as when 'after_env_before_script' is nil.# There is probably a better way that doesn't involve making the above Heredoc horrible.ERB.new(template).result(binding).lines.reject {|l| l =~ /^\s*$/}.joinend

到这里，便是stage过程中启动脚本的生成，终止脚本的生成的话，流程也一致，主要是获取应用进程的pid，然后通过kill pid来实现对应用的终止。

以上便是笔者对于Cloud Foundry中Stager组件的简单源码分析。

关于作者：

孙宏亮，DAOCLOUD软件工程师。两年来在云计算方面主要研究PaaS领域的相关知识与技术。坚信轻量级虚拟化容器的技术，会给PaaS领域带来深度影响，甚至决定未来PaaS技术的走向。

转载清注明出处。

这篇文档更多出于我本人的理解，肯定在一些地方存在不足和错误。希望本文能够对接触Cloud Foundry中Stager组件的人有些帮助，如果你对这方面感兴趣，并有更好的想法和建议，也请联系我。

我的邮箱：allen.sun@daocloud.io
新浪微博： @莲子弗如清

这篇关于Cloud Foundry中Stager组件的源码分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！