Cloud foundry warden 源码学习(2)

Warden daemon warden进程首先由BOSH在VM上启动:

root 1724 1 0 Jun06 ? 02:21:21 ruby /var/vcap/data/packages/warden/38.1/warden/vendor/bundle/ruby/1.9.1/bin/rake warden:start[/var/vcap/jobs/dea_next/config/warden.yml]

该启动命令的入口是通过rake task完成的

namespace :warden do
  desc "Run Warden server"
  task :start, :config_path do |t, args|
    require "warden/server"

    if args[:config_path]
      config = YAML.load_file(args[:config_path])
    end

    Warden::Server.setup(config || {})
    Warden::Server.run!
  end
end

看一下Warden::Server的setup和run!方法

   

#这里主要通过读取yml文件完成配置设置

def self.setup(config)
      @config = Config.new(config)

      setup_server
      setup_logging
      setup_network
      setup_port
      setup_user
    end

这里主要是建立unix socket端口的侦听，DEA如果有消息发送到该端口(DEA为warden client, Warden daemon进程为 warden server)， 则会执行warden容器的创建和销毁等。

def self.run!
      ::EM.epoll

      old_soft, old_hard = Process.getrlimit(:NOFILE)
      Process.setrlimit(Process::RLIMIT_NOFILE, 32768)
      new_soft, new_hard = Process.getrlimit(:NOFILE)
      logger.debug2("rlimit_nofile: %d => %d" % [old_soft, new_soft])

      # Log configuration
      logger.info("Configuration", config.to_hash)

      ::EM.run {
        f = Fiber.new do
          container_klass.setup(self.config)

          ::EM.error_handler do |error|
            logger.log_exception(error)
          end

          recover_containers

          FileUtils.rm_f(unix_domain_path)

          #ClientConnection是回调方法类，一旦有unix socket连接请求接入，ClientConnection的receive_data(data)方法会被调用，最终ClientConnection的process(Request)方法会被调用。
          server = ::EM.start_unix_domain_server(unix_domain_path, ClientConnection)
          ::EM.start_server("127.0.0.1",
                            config.health_check_server["port"],
                            HealthCheck)

          @drainer = Drainer.new(server, "USR2")
          @drainer.on_complete do
            Fiber.new do
              logger.info("Drain complete")

              # Serialize container state
              container_klass.registry.each { |_, c| c.write_snapshot }
              container_klass.registry.each { |_, c| c.jobs.each_value(&:kill) }

              EM.stop
            end.resume(nil)
          end

          # This is intentionally blocking. We do not want to start accepting
          # connections before permissions have been set on the socket.
          FileUtils.chmod(unix_domain_permissions, unix_domain_path)

          # Let the world know Warden is ready for action.
          logger.info("Listening on #{unix_domain_path}")

          if pidfile = config.server["pidfile"]
            logger.info("Writing pid #{Process.pid} to #{pidfile}")
            PidFile.new(piddir: File.dirname(pidfile), pidfile: File.basename(pidfile))
          end
        end

        f.resume
      }
    end

ClientConnection的process(Request)会将请求转发到对应的wanden container。目前实现是Warden::Container::Linux.

  def process(request)
        case request
        when Protocol::PingRequest
          response = request.create_response
          send_response(response)

        when Protocol::ListRequest
          response = request.create_response
          response.handles = Server.container_klass.registry.keys.map(&:to_s)
          send_response(response)

        when Protocol::EchoRequest
          response = request.create_response
          response.message = request.message
          send_response(response)

        #这里实例化一个Warden::Containe::Linux注册连接并分发请求
        when Protocol::CreateRequest
          container = Server.container_klass.new
          container.register_connection(self)
          response = container.dispatch(request)
          send_response(response)

        else

          #这里找到已经实例化好的warden实例并将请求转发
          if request.respond_to?(:handle)
            container = find_container(request.handle)
            process_container_request(request, container)
          else
            raise WardenError.new("Unknown request: #{request.class.name.split("::").last}")
          end
        end
      rescue WardenError => e
        send_error(e)
      rescue => e
        logger.log_exception(e)
        send_error(e)
      end

可见，Warden Server启动后，建立unix socket侦听，接收DEA或者runner发送来的请求，创建warden container，执行对应的bash命令，其主要职责是管理多个warden container。而具体执行创建，任务执行的工作，是由Warden::Container::Linux来完成的。

实际上，可以通过运行runner来和warden server连接

var/vcap/packages/warden/warden/bin/warden --socket /var/vcap/data/warden/warden.sock  --trace

ENV['BUNDLE_GEMFILE'] ||= File.expand_path("../../Gemfile", __FILE__)
$LOAD_PATH.unshift(File.expand_path("../../lib", __FILE__))

require "bundler"
Bundler.setup

require "warden/repl/repl_v2_runner"

Warden::Repl::ReplRunner.run(ARGV)

Warden::Repl::ReplRunner可以以交互式和非交互式两种方式运行，通过构造针对Warden Server的各种类型的Request，将请求通过/tmp/socket发送给Warden server，收到response后显示到System.out。而DEA也是类似的方式和Warden Server交互。