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交互。