Class: Raptor::ControlServer

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/control_server.rb,
sig/generated/raptor/control_server.rbs

Overview

Serves cluster statistics over a Unix socket.

Constant Summary collapse

SHUTDOWN =

Returns:

  • (::Symbol) 
:shutdown

Instance Method Summary collapse

Constructor Details

#initialize(url) { ... } ⇒ ControlServer

Creates a new ControlServer for url without binding it.

Parameters:

  • url (String) —

    unix:// URL to listen on

Yields:

  • for each stats request

Yield Returns:

  • (Hash) —

    cluster statistics

Raises:

  • (ArgumentError) —

    if the URL is not a Unix socket



31
32
33
34
35
36
37
38
39
40
 
# File 'lib/raptor/control_server.rb', line 31

def initialize(url, &stats)
  uri = URI(url)
  raise ArgumentError, "control_url must use unix://" unless uri.scheme == "unix" && !uri.path.empty?

  @path = uri.path
  @stats = stats
  @server = nil
  @client = Atom.new(nil)
  @thread = nil
end

Instance Method Details

#bind ⇒ void

This method returns an undefined value.

Binds the Unix socket.



47
48
49
50
 
# File 'lib/raptor/control_server.rb', line 47

def bind
  remove_stale_socket
  @server = UNIXServer.new(@path)
end

#handle(client) ⇒ void

This method returns an undefined value.

Writes the response for one control-socket request.

Parameters:

  • client (UNIXSocket) —

    connected control client



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/raptor/control_server.rb', line 133

def handle(client)
  request_line = client.gets
  while (line = client.gets)
    break if line == "\r\n"
  end

  if request_line&.start_with?("GET /stats ")
    body = JSON.generate(@stats.call)
    client.write("HTTP/1.0 200 OK\r\nContent-Type: application/json\r\nContent-Length: #{body.bytesize}\r\n\r\n#{body}")
  else
    client.write("HTTP/1.0 404 Not Found\r\nContent-Length: 0\r\n\r\n")
  end
rescue IOError, SystemCallError
ensure
  client.close rescue nil
  @client.swap { |current| current.equal?(client) ? nil : current }
end

#remove_stale_socket ⇒ void

This method returns an undefined value.

Removes a stale socket while refusing to replace an active server.

Raises:

  • (RuntimeError) —

    if another server is listening on the socket



93
94
95
96
97
98
99
100
101
102
 
# File 'lib/raptor/control_server.rb', line 93

def remove_stale_socket
  return unless File.exist?(@path)

  begin
    UNIXSocket.new(@path).close
    raise "Socket #{@path.inspect} is already in use"
  rescue Errno::ECONNREFUSED
    File.delete(@path)
  end
end

#serve ⇒ void

This method returns an undefined value.

Accepts and handles control requests until shutdown begins.



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/raptor/control_server.rb', line 109

def serve
  until @client.value == SHUTDOWN
    readable, = IO.select([@server], nil, nil, 1)
    next unless readable

    client = @server.accept_nonblock(exception: false)
    next unless client.is_a?(UNIXSocket)

    if @client.swap { |current| current == SHUTDOWN ? current : client } == SHUTDOWN
      client.close
      return
    end

    handle(client)
  end
rescue IOError, Errno::EBADF
end

#shutdown ⇒ void

This method returns an undefined value.

Stops serving and removes the socket.



73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/raptor/control_server.rb', line 73

def shutdown
  client = nil
  @client.swap do |current|
    client = current
    SHUTDOWN
  end
  @server&.close
  client.close if client.is_a?(UNIXSocket)
  @thread&.join
  File.delete(@path) rescue nil
end

#start ⇒ void

This method returns an undefined value.

Starts serving requests in a background thread.



57
58
59
60
61
62
63
64
65
66
 
# File 'lib/raptor/control_server.rb', line 57

def start
  owner_pid = Process.pid
  at_exit { File.delete(@path) rescue nil if Process.pid == owner_pid }

  @thread = Thread.new do
    Thread.current.name = "Control Server"

    serve
  end
end