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 control server for url without binding it.

Parameters:

  • url (String) —

    unix:// URL to listen on

Yields:

Yield Returns:

  • (Hash) —

    cluster statistics

Raises:

  • (ArgumentError) —

    if the URL is not a Unix socket



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

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.



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

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



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

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



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

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.



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

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.



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

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.



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

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