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

MAX_REQUEST_SIZE =

Returns:

  • (Object) 
16 * 1024
REQUEST_TIMEOUT =

Returns:

  • (::Integer) 
1

Instance Method Summary collapse

Constructor Details

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

Creates a control server for url without binding it.

RBS:

  • (String url) { () -> Hash[Symbol, untyped] } -> void

Parameters:

  • url (String) —

    unix:// URL to listen on

Yields:

Yield Returns:

  • (Hash) —

    cluster statistics

Raises:

  • (ArgumentError) —

    if the URL is not a Unix socket



33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/raptor/control_server.rb', line 33

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
  @running = AtomicBoolean.new(false)
end

Instance Method Details

#bind ⇒ void

This method returns an undefined value.

Binds the Unix socket.

RBS:

  • () -> void



50
51
52
53
 
# File 'lib/raptor/control_server.rb', line 50

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

#close_client ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



155
156
157
158
159
160
161
162
 
# File 'lib/raptor/control_server.rb', line 155

def close_client
  client = nil
  @client.swap do |current|
    client = current
    nil
  end
  client&.close
end

#handle(client) ⇒ void

This method returns an undefined value.

RBS:

  • (UNIXSocket client) -> void

Parameters:

  • client (UNIXSocket) 


120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/raptor/control_server.rb', line 120

def handle(client)
  request = read_request(client)
  return unless request

  if request.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
  close_client
end

#read_request(client) ⇒ String?

RBS:

  • (UNIXSocket client) -> String?

Parameters:

  • client (UNIXSocket)

Returns:

  • (String, nil) 


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

def read_request(client)
  request = String.new
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + REQUEST_TIMEOUT

  loop do
    return request if request.include?("\r\n\r\n")
    return if request.bytesize >= MAX_REQUEST_SIZE

    timeout = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
    return if timeout <= 0 || !client.wait_readable(timeout)

    chunk = client.read_nonblock(MAX_REQUEST_SIZE - request.bytesize, exception: false)
    return unless chunk.is_a?(String)

    request << chunk
  end
end

#remove_stale_socket ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



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

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.

RBS:

  • () -> void



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/raptor/control_server.rb', line 100

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

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

    @client.swap { client }
    unless @running.true?
      close_client
      break
    end

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

#shutdown ⇒ void

This method returns an undefined value.

Stops serving and removes the socket.

RBS:

  • () -> void



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

def shutdown
  @running.make_false
  @server&.close
  close_client
  @thread&.join
  File.delete(@path) rescue nil
end

#start ⇒ void

This method returns an undefined value.

Starts serving requests in a background thread.

RBS:

  • () -> void



60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/raptor/control_server.rb', line 60

def start
  @running.make_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