Class: Falcon::Service::Supervisor

Inherits:
Generic
  • Object
show all
Defined in:
lib/falcon/service/supervisor.rb

Overview

Implements a host supervisor which can restart the host services and provide various metrics about the running processes.

Instance Method Summary collapse

Methods inherited from Generic

#include?, #logger, #name, wrap

Constructor Details

#initialize(environment) ⇒ Supervisor

Initialize the supervisor using the given environment.



18
19
20
21
22
 
# File 'lib/falcon/service/supervisor.rb', line 18

def initialize(environment)
	super
	
	@bound_endpoint = nil
end

Instance Method Details

#do_metrics(message) ⇒ Object

Capture process metrics relating to the process group that the supervisor belongs to.



43
44
45
 
# File 'lib/falcon/service/supervisor.rb', line 43

def do_metrics(message)
	Process::Metrics::General.capture(pid: Process.ppid, ppid: Process.ppid)
end

#do_restart(message) ⇒ Object

Restart the process group that the supervisor belongs to.



31
32
33
34
35
36
37
38
39
40
 
# File 'lib/falcon/service/supervisor.rb', line 31

def do_restart(message)
	# Tell the parent of this process group to spin up a new process group/container.
	# Wait for that to start accepting new connections.
	# Stop accepting connections.
	# Wait for existing connnections to drain.
	# Terminate this process group.
	signal = message[:signal] || :INT
	
	Process.kill(signal, Process.ppid)
end

#endpoint ⇒ Object

The endpoint which the supervisor will bind to. Typically a unix pipe in the same directory as the host.



26
27
28
 
# File 'lib/falcon/service/supervisor.rb', line 26

def endpoint
	@evaluator.endpoint
end

#handle(message) ⇒ Object

Handle an incoming request.



49
50
51
52
53
54
55
56
 
# File 'lib/falcon/service/supervisor.rb', line 49

def handle(message)
	case message[:please]
	when 'restart'
		self.do_restart(message)
	when 'metrics'
		self.do_metrics(message)
	end
end

#setup(container) ⇒ Object

Start the supervisor process which accepts connections from the bound endpoint and processes JSON formatted messages.



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/falcon/service/supervisor.rb', line 71

def setup(container)
	container.run(name: self.name, restart: true, count: 1) do |instance|
		Async do
			@bound_endpoint.accept do |peer|
				stream = Async::IO::Stream.new(peer)
				
				while message = stream.gets("\0")
					response = handle(JSON.parse(message, symbolize_names: true))
					stream.puts(response.to_json, separator: "\0")
				end
			end
			
			instance.ready!
		end
	end
	
	super
end

#start ⇒ Object

Bind the supervisor to the specified endpoint.



59
60
61
62
63
64
65
66
67
 
# File 'lib/falcon/service/supervisor.rb', line 59

def start
	Console.logger.info(self) {"Binding to #{self.endpoint}..."}
	
	@bound_endpoint = Async::Reactor.run do
		Async::IO::SharedEndpoint.bound(self.endpoint)
	end.wait
	
	super
end

#stop ⇒ Object

Release the bound endpoint.



91
92
93
94
95
96
 
# File 'lib/falcon/service/supervisor.rb', line 91

def stop
	@bound_endpoint&.close
	@bound_endpoint = nil
	
	super
end