Cloud Foundry中DEA组件内应用的启动与资源监控

本文主要是介绍Cloud Foundry中DEA组件内应用的启动与资源监控，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

        Cloud Foundry中所有的应用都运行在一个称为DEA的组件中，DEA的全称是Droplet Execution Agent。

        DEA的主要功能可以分为两个部分：运行所有的应用，监控所有的应用。本文主要讲解Cloud Foundry v1版本中DEA如何启动一个应用，以及DEA如何监控应用的资源使用。虽然DEA两个功能的实现远不止这么多，但是笔者认为启动应用和监控应用资源是DEA的精髓所在，很多其他的内容都是在这两个点上进行封装或者强化。

DEA启动应用

        在一般情况下，启动一个应用，首先需要三样东西：环境，源码，应用入口。

        关于环境，Cloud Foundry在DEA节点处，已经部署完多套不同框架应用运行所需要的环境。关于源码，Cloud Foundry中有一个droplet的概念，它是一个可运行的源码包，比用户上传的源码还要多一些Cloud Foundry自定义添加的内容，比如说对于Spring应用，Cloud Foundry会将Tomcat将应用源码打包在一起，并且添加Tomcat中应用的启动，终止脚本等。关于程序入口，刚才已经涉及到，打包过程中，启动脚本已经被加入droplet之中。因此，DEA只需要简单的通过环境来执行启动脚本，就可以成功启动应用。在研究了源码之后，也会发现DEA中最主要的文件名为agent.rb，在启动这方面，也就是充当一个代理的角色，通过Linux底层的脚本命令来完成对应用的操作。

        了解Cloud Foundry中消息机制的开发者，一定会熟悉NATS的订阅/发布机制，而DEA也正是通过这种机制实现接收“启动应用”的请求。订阅代码如下：

NATS.subscribe("dea.#{uuid}.start") { |msg| process_dea_start(msg) }

        可见DEA组件订阅“dea.#{uuid}.start”主题的消息后，只要NATS接受到相同主题消息的发布，就会将消息转发给DEA，也就是上行代码的msg，DEA接受到msg后，通过函数process_dea_start来处理。以下我们详细介绍process_dea_start方法实现。

        在process_dea_start方法中，首先通过JSON类来解析message消息，从中获取众多的属性值，然后将这些属性值赋值给方法内的变量。由于在下文的资源监控部分，会涉及到应用使用的内存，文件数，硬盘等，故在这里以这几个属性为例，介绍解析message。

      message_json = JSON.parse(message)mem     = DEFAULT_APP_MEMnum_fds = DEFAULT_APP_NUM_FDSdisk    = DEFAULT_APP_DISKif limits = message_json['limits']mem = limits['mem'] if limits['mem']num_fds = limits['fds'] if limits['fds']disk = limits['disk'] if limits['disk']end

        解析message后，在message_json中取出键为limits的值，如果存在的话，那就以limits的值来初始化变量mem,num_fds,disk，如果不存在的话，那就是使用DEA默认的值来给这些变量赋值。

        通过一系列资源检测之后，DEA开始为需要启动的应用创建合适的实例，这些实例最终需要保存在DEA所维护的内存块中。同时，DEA还需要为应用创建合适的文件目录来运行该应用。

         在DEA所在节点的文件系统中创建文件目录的代码如下：

instance_dir = File.join(@apps_dir, "#{name}-#{instance_index}-#{instance_id}")

         通过process_dea_start方法内的变量，创建在内从中的内存实例，简化代码如下：

      instance = {:droplet_id => droplet_id,:instance_id => instance_id,……:mem_quota => mem * (1024*1024),:disk_quota => disk  * (1024*1024),:fds_quota => num_fds,:state => :STARTING,……:cc_partition => cc_partition}

        随后，将以上创建的实例，存入DEA维护的Hash类型内存@droplets中，代码如下：

      instances = @droplets[droplet_id] || {}instances[instance_id] = instance@droplets[droplet_id] = instances

        然后，在process_dea_start方法中，紧接着定义了一个lambda函数start_operation。

       在start_operation方法中，DEA首先做了以下的操作：

• 为应用程序分配一个普通端口，供应用最终的用户进行访问
• 若需要配置debug端口，DEA为应用分配debug端口
• 若需要配置console端口，DEA为应用分配console端口

        做完端口分配之后，DEA开始作必要的环境设置操作。

        按次序的话，首先创建一个manifest变量，存储droplet.yaml文件中的状态值，通过该状态值，来检测应用程序是否准备就绪等先决准备工作。

        接着，创建变量prepare_script，，从源代码的某路径中将内容拷贝入prepare_script，以供执行启动操作命令的时候使用。

        接着，通过DEA运行平台的系统类型，选择不同的shell命令。

        接着，对当前应用目录进行一些权限设置，比如：使用chown命令，对应用的用户进行修改；使用chgrp命令，对用户组进行修改；使用chmod命令，对应用的读写权限进行修改。

        接着，通过setup_instance_env方法实现，对应用环境变量的设置。实现过程中，该方法将所有的环境变量都加入一个数组，最后将数组内所有的内容export到应用的环境变量中。

        紧接着，一个很重要的部分是，DEA定义了一个proc代码块，来实现对DEA中要启动应用程序启动时的资源限制，需要注意的是，这仅仅是代码块的定义，并不是在process_dea_start方法中，在这个代码块所在的位置被调用，代码如下：

        exec_operation = proc do |process|process.send_data("cd #{instance_dir}\n")if @secure || @enforce_ulimitprocess.send_data("renice 0 $$\n")process.send_data("ulimit -m #{mem_kbytes} 2> /dev/null\n")  # ulimit -m takes kb, soft enforceprocess.send_data("ulimit -v 3000000 2> /dev/null\n") # virtual memory at 3G, this will be enforcedprocess.send_data("ulimit -n #{num_fds} 2> /dev/null\n")process.send_data("ulimit -u 512 2> /dev/null\n") # processes/threadsprocess.send_data("ulimit -f #{disk_limit} 2> /dev/null\n") # File size to complete disk usageprocess.send_data("umask 077\n")endapp_env.each { |env| process.send_data("export #{env}\n") }if instance[:port]process.send_data("./startup -p #{instance[:port]}\n")elseprocess.send_data("./startup\n")endprocess.send_data("exit\n")end

        该执行块，首先进入应用相应的目录下，重新设置完程序运行的优先级之后，对该进程启动时所需要的一些资源进行设置，设置过程中使用的是Linux操作系统中的ulimit命令，该命令可以做到：限制shell启动进程是所占用的资源数量，具体的资源类型，有：物理内存，虚拟内存，所需打开的文件数，需要创建的线程数，磁盘容量等。需要注意的是，这样的限制是临时性的限制，一旦shell会话结束之后，这样的限制也就失效了。这样也比较可以理解，否则DEA中应用对于资源的隔离和控制也就不难实现了。资源的限制做完之后，随即实现的是env环境变量的导入，使用export命令。倒数第二实现的是，执行启动脚本，从而真正的实现应用的启动。最后，需要将该shell会话关闭，使用exit命令。

        执行代码块如下，退出代码块如下：

        exit_operation = proc do |_, status|@logger.info("#{name} completed running with status = #{status}.")@logger.info("#{name} uptime was #{Time.now - instance[:start]}.")stop_droplet(instance)end

        该部分代码较为简单，只是调用了stop_droplet方法。

        接着会有一个非常重要的模块，负责实现应用的真正启动：

Bundler.with_clean_env { EM.system("#{@dea_ruby} -- #{prepare_script} true #{sh_command}", exec_operation, exit_operation) }

        该代码的含义是：使用一个全新环境变量的env进行操作括号内的内容，EM.system中的内容即为最终执行的ruby启动脚本。

        接着会有一些检测应用准备情况或者pid信息的操作，为一些必要的辅助操作。

        最后，DEA定义了一个纤程Fiber，纤程中首先通过stage_app_dir来实现droplet中所有的内容的准备工作。一旦成功，则马上调用代码块start_operation。

        最后的最后，使用f.resume方法实现，纤程的唤醒。

DEA监控应用资源

        DEA完成对所在节点应用的监控主要通过周期性监控来完成的，代码如下：

EM.add_timer(MONITOR_INTERVAL) { monitor_apps }

        其中默认的监控间隔为2s。所有的实现在方法中monitor_apps中实现，简化代码如下：

    def monitor_apps(startup_check = false)……process_statuses = `ps axo pid=,ppid=,pcpu=,rss=,user=`.split("\n")……process_statuses.each do |process_status|parts = process_status.lstrip.split(/\s+/)pid = parts[PID_INDEX].to_ipid_info[pid] = parts(user_info[parts[USER_INDEX]] ||= []) << parts if (@secure && parts[USER_INDEX] =~ SECURE_USER)end……if startup_checkdu_all_out = `cd #{@apps_dir}; du -sk * 2> /dev/null`monitor_apps_helper(startup_check, start, du_start, du_all_out, pid_info, user_info)elsedu_proc = proc do |p|p.send_data("cd #{@apps_dir}\n")p.send_data("du -sk * 2> /dev/null\n")p.send_data("exit\n")endcont_proc = proc do |output, status|monitor_apps_helper(startup_check, start, du_start, output, pid_info, user_info)endEM.system('/bin/sh', du_proc, cont_proc)endend

        其中，首先看一下代码：process_statuses = `ps axo pid=,ppid=,pcpu=,rss=,user=`.split("\n")，其实现的是：通过操作系统层，将所有进程运行的信息，存入process_status数组中，然后将这个数组中按pid信息分类，存入Hash类型pid_info中，同时又通过user信息分类，存入Hash类型user_info中，以备后续资源监控使用。

        然后通过startup_check参数，进行不同的磁盘检测模块，代码为：

      du_entries = du_all_out.split("\n")du_entries.each do |du_entry|size, dir = du_entry.split("\t")size = size.to_i * 1024 # Convert to bytesdu_hash[dir] = sizeenddu_all_out = `cd #{@apps_dir}; du -sk * 2> /dev/null`

        该命令实现进入应用的根目录，执行du命令，统计应用目录所占用的磁盘空间，随后使用monitor_app_helper方法对方才收集到的数据进行处理，得到最终所需要的监控数据。判断语句的另外一部分，只是通过proc代码的方法来实现相同的功能。

        以下分析monitor_app_helper方法。以下是通过目录分离磁盘占用空间的实现：

      du_entries = du_all_out.split("\n")du_entries.each do |du_entry|size, dir = du_entry.split("\t")size = size.to_i * 1024 # Convert to bytesdu_hash[dir] = sizeend

        do_all_out显示的是@droplets中所有应用的磁盘使用情况，通过split方法或者每个应用的信息，然后存入Hash类型du_hash中。

        以下是最重要的信息的汇总：

      @droplets.each_value do |instances|instances.each_value do |instance|if instance[:pid] && pid_info[instance[:pid]]pid = instance[:pid]mem = cpu = 0disk = du_hash[File.basename(instance[:dir])] || 0# For secure mode, gather all stats for secure_user so we can process forks, etc.if @secure && user_info[instance[:secure_user]]user_info[instance[:secure_user]].each do |part|mem += part[MEM_INDEX].to_fcpu += part[CPU_INDEX].to_f# disabled for now, LSOF is too slow to run per app/user# deleted_disk = grab_deleted_file_usage(instance[:secure_user])# disk += deleted_diskendelsemem = pid_info[pid][MEM_INDEX].to_fcpu = pid_info[pid][CPU_INDEX].to_fendusage = @usage[pid] ||= []cur_usage = { :time => Time.now, :cpu => cpu, :mem => mem, :disk => disk }usage << cur_usageusage.shift if usage.length > MAX_USAGE_SAMPLEScheck_usage(instance, cur_usage, usage) if @secure#@logger.debug("Droplet Stats are = #{JSON.pretty_generate(usage)}")@mem_usage += memmetrics.each do |key, value|metric = value[instance[key]] ||= {:used_memory => 0, :reserved_memory => 0,:used_disk => 0, :used_cpu => 0}metric[:used_memory] += memmetric[:reserved_memory] += instance[:mem_quota] / 1024metric[:used_disk] += diskmetric[:used_cpu] += cpuend# Track running apps for varz trackingi2 = instance.dupi2[:usage] = cur_usage # Snapshotrunning_apps << i2# Re-register with router on startup since these are orphaned and may have been dropped.register_instance_with_router(instance) if startup_checkelse……endendend

        遍历@droplets,再遍历@droplets中的每一个对象的instance，首先获取实例的pid，并初始化mem，cpu，disk。随后，找到该instance的：secure_user，通过在user_info中找到所有instance[:secure_user]的值，然后，将其累加进入mem和cpu。

        将获得指定instance的资源使用情况后，需要将这些信息存入DEA维护的内存@usage中，代码如下：

            usage = @usage[pid] ||= []cur_usage = { :time => Time.now, :cpu => cpu, :mem => mem, :disk => disk }usage << cur_usage

        第一行需要注意操作符||=，代码含义为若@usage[pid]不为空的话，将[ ]赋值给@usage[pid]，并最终赋值给usage。第二行代码初始化了一个hash变量，第三行代码表示将cur_usage加入usage数组，而usage数组指向的是@usage[pid]的地址，所以也就相当于在@usage[pid]中添加cur_usage。

       而metrics变量则是记录整个DEA所有运行的应用加起来的应用资源，以便之后更新varz信息。

       以上便是对于Cloud Foundry的DEA中的应用启动与应用资源监控。

评价

        由以上的分析可知，Cloud Foundry v1版本中的DEA在实现资源的监控时，通过linux底层的ps来完成，而对于资源的控制，则并没有做的很好，使用的方式为启动时限制与超额时停止的策略。该方法显然不是最佳的，而Cloud Foundry也推出了warden的容器，实现对DEA中应用资源使用时的控制与隔离。如果笔者这脑袋能理解或者读懂的话，期待不久给大家送上warden的实现，以及cgroup的机制。

关于作者：

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

转载请注明出处。

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

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

这篇关于Cloud Foundry中DEA组件内应用的启动与资源监控的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---