Class: ProcessBot::ControlSocket

Inherits:
Object
  • Object
show all
Defined in:
lib/process_bot/control_socket.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options:, process:) ⇒ ControlSocket

Returns a new instance of ControlSocket.



8
9
10
11
12
13
14
 
# File 'lib/process_bot/control_socket.rb', line 8

def initialize(options:, process:)
  @options = options
  @process = process
  @port = options.fetch(:port).to_i
  @clients = []
  @clients_mutex = Mutex.new
end

Instance Attribute Details

#clients ⇒ Object (readonly)

Returns the value of attribute clients.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def clients
  @clients
end

#clients_mutex ⇒ Object (readonly)

Returns the value of attribute clients_mutex.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def clients_mutex
  @clients_mutex
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def options
  @options
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def port
  @port
end

#process ⇒ Object (readonly)

Returns the value of attribute process.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def process
  @process
end

#server ⇒ Object (readonly)

Returns the value of attribute server.



6
7
8
 
# File 'lib/process_bot/control_socket.rb', line 6

def server
  @server
end

Instance Method Details

#actually_start_tcp_server(host, port) ⇒ Object 



112
113
114
 
# File 'lib/process_bot/control_socket.rb', line 112

def actually_start_tcp_server(host, port)
  TCPServer.new(host, port)
end

#add_client(client) ⇒ Object 



190
191
192
193
194
 
# File 'lib/process_bot/control_socket.rb', line 190

def add_client(client)
  clients_mutex.synchronize do
    clients << client
  end
end

#broadcast_log(_event_name, output:, type:) ⇒ Object 



179
180
181
182
183
184
185
186
187
188
 
# File 'lib/process_bot/control_socket.rb', line 179

def broadcast_log(_event_name, output:, type:)
  safe_output = normalize_output(output)
  payload = JSON.generate(type: "log", stream: type.to_s, output: safe_output)

  clients_snapshot.each do |client|
    client.puts(payload)
  rescue IOError, Errno::EPIPE, Errno::ECONNRESET
    remove_client(client)
  end
end

#clients_snapshot ⇒ Object 



202
203
204
205
206
 
# File 'lib/process_bot/control_socket.rb', line 202

def clients_snapshot
  clients_mutex.synchronize do
    clients.dup
  end
end

#duplicate_id_entry_matches?(entry, id, basename) ⇒ Boolean

Returns:

  • (Boolean) 


81
82
83
84
85
86
 
# File 'lib/process_bot/control_socket.rb', line 81

def duplicate_id_entry_matches?(entry, id, basename)
  return false unless entry[:id] == id
  return true if basename && entry[:application_basename] == basename

  entry[:application_basename].nil? && same_control_port?(entry)
end

#duplicate_id_error_message(id, duplicates) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/process_bot/control_socket.rb', line 94

def duplicate_id_error_message(id, duplicates)
  details = duplicates.map { |entry| "PID #{entry[:pid]} on port #{entry[:port]}" }.join(", ")
  example_port = duplicates.first[:port]
  handler = options.fetch(:handler, "custom")
  release_path = options.fetch(:release_path, "/")

  "Another process_bot with id=#{id.inspect} is already running for this application (#{details}). " \
    "Stop it (e.g. `process_bot --command stop --port #{example_port} --id #{id} " \
    "--handler #{handler} --release-path #{release_path}`) " \
    "or kill that PID before starting a new instance."
end

#ensure_no_duplicate_id! ⇒ Object

Prevent a second process_bot with the same ‘–id` under the same application from starting while the first is still alive. The `start_tcp_server` loop silently drifts to a free port when the requested one is in use; drift is intentional when unrelated process_bots share a host, but it’s a bug when a Capistrano deploy’s stop failed to clean up the previous release’s process_bot and the new release’s start drifts around the zombie. Scope the match by ‘application_basename` (derived from `release_path`) so that two unrelated apps on the same host can reuse a generic id like `sidekiq-main` without falsely blocking each other.



65
66
67
68
69
70
71
72
73
 
# File 'lib/process_bot/control_socket.rb', line 65

def ensure_no_duplicate_id!
  id = options[:id]
  return if id.nil? || id.to_s.strip.empty?

  duplicates = find_duplicate_id_entries(id.to_s, safe_application_basename)
  return if duplicates.empty?

  raise duplicate_id_error_message(id, duplicates)
end

#find_duplicate_id_entries(id, basename) ⇒ Object 



75
76
77
78
79
 
# File 'lib/process_bot/control_socket.rb', line 75

def find_duplicate_id_entries(id, basename)
  running_process_bot_entries.select do |entry|
    duplicate_id_entry_matches?(entry, id, basename)
  end
end

#handle_client(client) ⇒ Object

rubocop:disable Metrics/AbcSize



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/process_bot/control_socket.rb', line 136

def handle_client(client) # rubocop:disable Metrics/AbcSize
  add_client(client)

  loop do
    data = client.gets
    break if data.nil? # Client disconnected

    command = JSON.parse(data)
    command_type = command.fetch("command")

    if command_type == "graceful" || command_type == "graceful_no_wait" || command_type == "restart" || command_type == "stop"
      begin
        unless process.accept_control_commands?
          client.puts(JSON.generate(type: "error", message: "ProcessBot is shutting down", backtrace: Thread.current.backtrace))
          break
        end

        command_options = if command["options"]
          symbolize_keys(command.fetch("options"))
        else
          {}
        end

        process.with_control_command do
          run_command(command_type, command_options, client)
        end
      rescue => e # rubocop:disable Style/RescueStandardError
        logger.error e.message
        logger.error e.backtrace

        client.puts(JSON.generate(type: "error", message: e.message, backtrace: e.backtrace))

        raise e
      end
    else
      client.puts(JSON.generate(type: "error", message: "Unknown command: #{command_type}", backtrace: Thread.current.backtrace))
    end
  end
ensure
  remove_client(client)
  client.close unless client.closed?
end

#logger ⇒ Object 



16
17
18
 
# File 'lib/process_bot/control_socket.rb', line 16

def logger
  @logger ||= ProcessBot::Logger.new(options: options)
end

#normalize_output(output) ⇒ Object 



208
209
210
 
# File 'lib/process_bot/control_socket.rb', line 208

def normalize_output(output)
  output.to_s.encode("UTF-8", invalid: :replace, undef: :replace, replace: "?")
end

#process_bot_entry_from_process_data(process, process_data) ⇒ Object 



257
258
259
260
261
262
263
264
265
 
# File 'lib/process_bot/control_socket.rb', line 257

def process_bot_entry_from_process_data(process, process_data)
  {
    application: process_data["application"],
    application_basename: process_data["application_basename"],
    id: process_data["id"]&.to_s,
    pid: process.data["pid"] || process.pid,
    port: process_data["port"]&.to_i
  }
end

#remove_client(client) ⇒ Object 



196
197
198
199
200
 
# File 'lib/process_bot/control_socket.rb', line 196

def remove_client(client)
  clients_mutex.synchronize do
    clients.delete(client)
  end
end

#run_client_loop ⇒ Object 



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

def run_client_loop
  Thread.new do
    loop do
      begin
        client = server.accept
      rescue IOError, Errno::EBADF
        break
      end

      Thread.new do
        handle_client(client)
      end
    end
  end
end

#run_command(command_type, command_options, client) ⇒ Object 



223
224
225
226
227
228
 
# File 'lib/process_bot/control_socket.rb', line 223

def run_command(command_type, command_options, client)
  logger.logs "Command #{command_type} with options #{command_options}"

  process.__send__(command_type, **command_options)
  client.puts(JSON.generate(type: "success"))
end

#running_process_bot_entries ⇒ Object

Parsed ‘id:, pid:, port:` entries for every running process_bot visible to `ps`, extracted from each instance’s JSON process title.



237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/process_bot/control_socket.rb', line 237

def running_process_bot_entries
  entries = []

  Knj::Unix_proc.list("grep" => "ProcessBot") do |process|
    process_command = process.data.fetch("cmd")
    match = process_command.match(/ProcessBot (\{.+\})/)
    next unless match

    begin
      process_data = JSON.parse(match[1])
    rescue JSON::ParserError
      next
    end

    entries << process_bot_entry_from_process_data(process, process_data)
  end

  entries
end

#safe_application_basename ⇒ Object 



106
107
108
109
110
 
# File 'lib/process_bot/control_socket.rb', line 106

def safe_application_basename
  options.application_basename
rescue KeyError
  nil
end

#same_control_port?(entry) ⇒ Boolean

Returns:

  • (Boolean) 


88
89
90
91
92
 
# File 'lib/process_bot/control_socket.rb', line 88

def same_control_port?(entry)
  return false unless entry[:port] && options[:port]

  entry[:port].to_i == options[:port].to_i
end

#start ⇒ Object 



20
21
22
23
24
25
26
27
28
 
# File 'lib/process_bot/control_socket.rb', line 20

def start
  start_tcp_server
  run_client_loop
  logger.logs "TCPServer started"
  options.events.call(:on_socket_opened, port: @port)
  options.events.connect(:on_log) do |event_name, output:, type:|
    broadcast_log(event_name, output: output, type: type)
  end
end

#start_tcp_server ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/process_bot/control_socket.rb', line 30

def start_tcp_server
  ensure_no_duplicate_id!

  used_ports = used_process_bot_ports
  attempts = 0

  loop do
    if used_ports.include?(@port)
      @port += 1
      next
    end

    attempts += 1
    @server = actually_start_tcp_server("localhost", @port)
    break
  rescue Errno::EADDRINUSE, Errno::EADDRNOTAVAIL => e
    if attempts <= 100
      @port += 1
      next
    else
      raise e
    end
  end
end

#stop ⇒ Object 



116
117
118
 
# File 'lib/process_bot/control_socket.rb', line 116

def stop
  server&.close
end

#symbolize_keys(hash) ⇒ Object 



212
213
214
215
216
217
218
219
220
221
 
# File 'lib/process_bot/control_socket.rb', line 212

def symbolize_keys(hash)
  new_hash = {}
  hash.each do |key, value|
    next if key == "port"

    new_hash[key.to_sym] = value
  end

  new_hash
end

#used_process_bot_ports ⇒ Object 



230
231
232
 
# File 'lib/process_bot/control_socket.rb', line 230

def used_process_bot_ports
  running_process_bot_entries.filter_map { |entry| entry[:port] }.uniq
end