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.

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



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

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 = nil
  @thread = nil
  @running = false
  @mutex = Mutex.new
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.

Parameters:

  • client (UNIXSocket) 


113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/raptor/control_server.rb', line 113

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
  @mutex.synchronize { @client = nil }
end

#remove_stale_socket ⇒ void

This method returns an undefined value.



86
87
88
89
90
91
92
93
94
95
 
# File 'lib/raptor/control_server.rb', line 86

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.



98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/raptor/control_server.rb', line 98

def serve
  while @running
    readable, = IO.select([@server], nil, nil, 1)
    next unless readable

    @mutex.synchronize do
      client = @server.accept_nonblock(exception: false)
      @client = client if client.is_a?(UNIXSocket)
    end
    handle(@client) if @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
 
# File 'lib/raptor/control_server.rb', line 73

def shutdown
  @running = false
  @mutex.synchronize do
    @server&.close
    @client&.close
  end
  @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
66
 
# File 'lib/raptor/control_server.rb', line 56

def start
  @running = true
  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