Class: RocketJob::CLI

Inherits:
Object
  • Object
show all
Includes:
SemanticLogger::Loggable
Defined in:
lib/rocket_job/cli.rb

Overview

Command Line Interface parser for Rocket Job

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(argv) ⇒ CLI

Returns a new instance of CLI.



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/rocket_job/cli.rb', line 16

def initialize(argv)
  @server                      = true
  @quiet                       = false
  @environment                 = nil
  @pidfile                     = nil
  @directory                   = "."
  @log_level                   = nil
  @log_file                    = nil
  @mongo_config                = nil
  @symmetric_encryption_config = nil
  @include_filter              = nil
  @exclude_filter              = nil
  @stop_server                 = nil
  @kill_server                 = nil
  @pause_server                = nil
  @resume_server               = nil
  @thread_dump                 = nil
  @list_servers                = nil
  parse(argv)
end

Instance Attribute Details

#directory ⇒ Object

Returns the value of attribute directory.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def directory
  @directory
end

#environment ⇒ Object

Returns the value of attribute environment.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def environment
  @environment
end

#exclude_filter ⇒ Object

Returns the value of attribute exclude_filter.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def exclude_filter
  @exclude_filter
end

#include_filter ⇒ Object

Returns the value of attribute include_filter.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def include_filter
  @include_filter
end

#kill_server ⇒ Object

Returns the value of attribute kill_server.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def kill_server
  @kill_server
end

#list_servers ⇒ Object

Returns the value of attribute list_servers.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def list_servers
  @list_servers
end

#log_file ⇒ Object

Returns the value of attribute log_file.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def log_file
  @log_file
end

#log_level ⇒ Object

Returns the value of attribute log_level.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def log_level
  @log_level
end

#max_workers ⇒ Object

Returns the value of attribute max_workers.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def max_workers
  @max_workers
end

#mongo_config ⇒ Object

Returns the value of attribute mongo_config.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def mongo_config
  @mongo_config
end

#pause_server ⇒ Object

Returns the value of attribute pause_server.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def pause_server
  @pause_server
end

#pidfile ⇒ Object

Returns the value of attribute pidfile.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def pidfile
  @pidfile
end

#quiet ⇒ Object

Returns the value of attribute quiet.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def quiet
  @quiet
end

#refresh ⇒ Object

Returns the value of attribute refresh.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def refresh
  @refresh
end

#resume_server ⇒ Object

Returns the value of attribute resume_server.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def resume_server
  @resume_server
end

#server ⇒ Object

Returns the value of attribute server.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def server
  @server
end

#stop_server ⇒ Object

Returns the value of attribute stop_server.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def stop_server
  @stop_server
end

#symmetric_encryption_config ⇒ Object

Returns the value of attribute symmetric_encryption_config.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def symmetric_encryption_config
  @symmetric_encryption_config
end

#thread_dump ⇒ Object

Returns the value of attribute thread_dump.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def thread_dump
  @thread_dump
end

#where_filter ⇒ Object

Returns the value of attribute where_filter.



11
12
13
 
# File 'lib/rocket_job/cli.rb', line 11

def where_filter
  @where_filter
end

Class Method Details

.eager_load_jobs(job_path = "jobs") ⇒ Object

Eager load files in jobs folder



162
163
164
165
166
167
168
169
 
# File 'lib/rocket_job/cli.rb', line 162

def self.eager_load_jobs(job_path = "jobs")
  Pathname.glob("#{job_path}/**/*.rb").each do |path|
    next if path.directory?

    logger.debug "Loading #{path}"
    require path.expand_path.to_s
  end
end

Instance Method Details

#boot_rails ⇒ Object

Initialize the Rails environment Returns [true|false] whether Rails is present



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/rocket_job/cli.rb', line 70

def boot_rails
  logger.info "Loading Rails environment: #{environment}"
  RocketJob.rails!

  require "rails"
  require "rocket_job/railtie"
  boot_file = Pathname.new(directory).join("config/environment.rb").expand_path
  require(boot_file.to_s)

  begin
    require "rails_semantic_logger"
  rescue LoadError
    raise "Add the following line to your Gemfile when running rails:\n gem 'rails_semantic_logger'"
  end

  # Override Rails log level if command line option was supplied
  SemanticLogger.default_level = log_level.to_sym if log_level

  return unless Rails.configuration.eager_load

  logger.measure_info("Eager loaded Rails and all Engines") do
    Rails.application.eager_load!
    Rails::Engine.subclasses.each(&:eager_load!)
    self.class.eager_load_jobs(File.expand_path("jobs", File.dirname(__FILE__)))
  end
end

#boot_standalone ⇒ Object

In a standalone environment, explicitly load config files



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/rocket_job/cli.rb', line 98

def boot_standalone
  # Try to load bundler if present
  begin
    require "bundler/setup"
    Bundler.require(environment)
  rescue LoadError
    nil
  end

  require "rocketjob"

  # Log to file except when booting rails, when it will add the log file path
  path = log_file ? Pathname.new(log_file) : Pathname.pwd.join("log/#{environment}.log")
  path.dirname.mkpath
  SemanticLogger.add_appender(file_name: path.to_s, formatter: :color)

  logger.info "Rails not detected. Running standalone: #{environment}"
  RocketJob::Config.load!(environment, mongo_config, symmetric_encryption_config)
  self.class.eager_load_jobs(File.expand_path("jobs", File.dirname(__FILE__)))
  self.class.eager_load_jobs
end

#list_the_servers(filter) ⇒ Object 



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/rocket_job/cli.rb', line 181

def list_the_servers(filter)
  layout = "%50.50s %20.20s %20.20s %20.20s %10.10s"
  puts format(layout, "Server Name", "Workers(Current/Max)", "Started", "Heartbeat", "State")
  header = "=" * 50
  puts format(layout, header, header, header, header, header)
  query = filter == :all ? RocketJob::Server.all : RocketJob::Server.where(name: /#{filter}/)
  query.each do |server|
    workers   = "#{server&.heartbeat&.workers}/#{server.max_workers}"
    duration  = Time.now - (server.started_at || Time.now)
    started   = "#{RocketJob.seconds_as_duration(duration)} ago"
    duration  = Time.now - (server&.heartbeat&.updated_at || Time.now)
    heartbeat = "#{RocketJob.seconds_as_duration(duration)} ago"
    puts format(layout, server.name, workers, started, heartbeat, server.state)
  end
  0
end

#override_config ⇒ Object

Allow the CLI to override the configuration after rails has been loaded.



121
122
123
124
125
126
 
# File 'lib/rocket_job/cli.rb', line 121

def override_config
  Config.max_workers    = max_workers if max_workers
  Config.include_filter = include_filter if include_filter
  Config.exclude_filter = exclude_filter if exclude_filter
  Config.where_filter   = where_filter if where_filter
end

#parse(argv) ⇒ Object

Parse command line options placing results in the corresponding instance variables



235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
 
# File 'lib/rocket_job/cli.rb', line 235

def parse(argv)
  parser        = OptionParser.new do |o|
    o.on("-n", "--name NAME", "Unique Name of this server (Default: host_name:PID)") do |arg|
      Config.name = arg
    end
    o.on("-w", "--workers COUNT", "Number of workers (threads) to start") do |arg|
      @max_workers = arg.to_i
    end
    o.on("--include REGEXP",
         'Limit this server to only those job classes that match this regular expression (case-insensitive). Example: "DirmonJob|WeeklyReportJob"') do |arg|
      @include_filter = Regexp.new(arg, true)
    end
    o.on("-E", "--exclude REGEXP",
         'Prevent this server from working on any job classes that match this regular expression (case-insensitive). Example: "DirmonJob|WeeklyReportJob"') do |arg|
      @exclude_filter = Regexp.new(arg, true)
    end
    o.on("-W", "--where JSON",
         "Limit this server instance to the supplied mongo query filter. Supply as a string in JSON format. Example: '{\"priority\":{\"$lte\":25}}'") do |arg|
      @where_filter = JSON.parse(arg)
    end
    o.on("-q", "--quiet", "Do not write to stdout, only to logfile. Necessary when running as a daemon") do
      @quiet = true
    end
    o.on("-d", "--dir DIR", "Directory containing Rails app, if not current directory") do |arg|
      @directory = arg
    end
    o.on("-e", "--environment ENVIRONMENT",
         "The environment to run the app on (Default: RAILS_ENV || RACK_ENV || development)") do |arg|
      @environment = arg
    end
    o.on("-l", "--log_level trace|debug|info|warn|error|fatal", "The log level to use") do |arg|
      @log_level = arg
    end
    o.on("-f", "--log_file FILE_NAME", "The log file to write to. Default: log/<environment>.log") do |arg|
      @log_file = arg
    end
    o.on("--pidfile PATH", "Use PATH as a pidfile") do |arg|
      @pidfile = arg
    end
    o.on("-m", "--mongo MONGO_CONFIG_FILE_NAME", "Path and filename of config file. Default: config/mongoid.yml") do |arg|
      @mongo_config = arg
    end
    o.on("-s", "--symmetric-encryption SYMMETRIC_ENCRYPTION_CONFIG_FILE_NAME",
         "Path and filename of Symmetric Encryption config file. Default: config/symmetric-encryption.yml") do |arg|
      @symmetric_encryption_config = arg
    end
    o.on("--list [FILTER]",
         "List active servers. Supply either an exact server name or a partial name as a filter.") do |filter|
      @quiet        = true
      @server       = false
      @list_servers = filter || :all
    end
    o.on("--refresh [SECONDS]",
         "When listing active servers, update the list by this number of seconds. Defaults to every 1 second.") do |seconds|
      @refresh = (seconds || 1).to_s.to_f
    end
    o.on("--stop [SERVER_NAME]",
         "Send event to stop a server once all in-process workers have completed. Optionally supply the complete or partial name of the server(s) to stop. Default: All servers.") do |server_name|
      @quiet       = true
      @server      = false
      @stop_server = server_name || :all
    end
    o.on("--kill [SERVER_NAME]",
         "Send event to hard kill a server. Optionally supply the complete or partial name of the server(s) to kill. Default: All servers.") do |server_name|
      @quiet       = true
      @server      = false
      @kill_server = server_name || :all
    end
    o.on("--pause [SERVER_NAME]",
         "Send event to pause a server. Optionally supply the complete or partial name of the server(s) to pause. Default: All servers.") do |server_name|
      @quiet        = true
      @server       = false
      @pause_server = server_name || :all
    end
    o.on("--resume [SERVER_NAME]",
         "Send event to resume a server. Optionally supply the complete or partial name of the server(s) to resume. Default: All servers.") do |server_name|
      @quiet         = true
      @server        = false
      @resume_server = server_name || :all
    end
    o.on("--dump [SERVER_NAME]",
         "Send event for a server to send a worker thread dump to its log file. Optionally supply the complete or partial name of the server(s). Default: All servers.") do |server_name|
      @quiet       = true
      @server      = false
      @thread_dump = server_name || :all
    end
    o.on("-v", "--version", "Print the version information") do
      puts "Rocket Job v#{RocketJob::VERSION}"
      exit 1
    end
  end
  parser.banner = "rocketjob <options>"
  parser.on_tail "-h", "--help", "Show help" do
    puts parser
    exit 1
  end
  parser.parse! argv
end

#perform_list_servers(filter) ⇒ Object 



171
172
173
174
175
176
177
178
179
 
# File 'lib/rocket_job/cli.rb', line 171

def perform_list_servers(filter)
  return list_the_servers(filter) unless refresh

  loop do
    list_the_servers(filter)
    sleep(refresh)
    puts
  end
end

#perform_server_action(server_name, action) ⇒ Object 



198
199
200
201
202
 
# File 'lib/rocket_job/cli.rb', line 198

def perform_server_action(server_name, action)
  server_ids(server_name).each { |server_id| RocketJob::Subscribers::Server.publish(action, server_id: server_id) }
  # RocketJob::Subscribers::Worker.publish(:stop, worker_id: 1, server_id: RocketJob::Server.running.last.id)
  0
end

#rails? ⇒ Boolean

Returns:

  • (Boolean) 


60
61
62
63
64
65
66
 
# File 'lib/rocket_job/cli.rb', line 60

def rails?
  @rails ||=
    begin
      boot_file = Pathname.new(directory).join("config/environment.rb").expand_path
      boot_file.file?
    end
end

#run ⇒ Object

Run a RocketJob::Server from the command line



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/rocket_job/cli.rb', line 38

def run
  Thread.current.name = "rocketjob main"
  RocketJob.server! if server
  setup_environment
  setup_logger
  rails? ? boot_rails : boot_standalone
  override_config
  write_pidfile

  # In case Rails did not load the Mongoid Config
  RocketJob::Config.load!(environment, mongo_config, symmetric_encryption_config) if ::Mongoid::Config.clients.empty?

  return perform_server_action(stop_server, :stop) if stop_server
  return perform_server_action(kill_server, :kill) if kill_server
  return perform_server_action(pause_server, :pause) if pause_server
  return perform_server_action(resume_server, :resume) if resume_server
  return perform_server_action(thread_dump, :thread_dump) if thread_dump
  return perform_list_servers(list_servers) if list_servers

  Supervisor.run
end

#server_ids(server_name) ⇒ Object

Returns server ids for the supplied exact server name, or partial match.

When no ‘:’ is supplied a partial hostname lookup is performed.

Example: Exact server name (hostname and pid) match:

"9cdbe7e995bc:1"

Example: Matches all servers that contain the string ‘.batch.user.org’:

".batch.user.org"

Raises:

  • (ArgumentError) 


213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/rocket_job/cli.rb', line 213

def server_ids(server_name)
  raise(ArgumentError, "Missing server name") unless server_name

  return [nil] if server_name == :all

  hostname, pid = server_name.split(":")
  raise(ArgumentError, "Missing server name in: #{server_name}") unless hostname

  if pid
    server = RocketJob::Server.where(name: server_name).first
    raise(ArgumentError, "No server with exact name: #{server_name} was found.") unless server

    return [server.id]
  end

  server_ids = RocketJob::Server.where(name: /#{hostname}/).collect(&:id)
  raise(ArgumentError, "No server with partial name: #{server_name} was found.") if server_ids.empty?

  server_ids
end

#setup_environment ⇒ Object 



141
142
143
144
145
146
147
148
 
# File 'lib/rocket_job/cli.rb', line 141

def setup_environment
  # Override Env vars when environment is supplied
  if environment
    ENV["RACK_ENV"] = ENV["RAILS_ENV"] = environment
  else
    self.environment = ENV["RAILS_ENV"] || ENV["RACK_ENV"] || "development"
  end
end

#setup_logger ⇒ Object 



150
151
152
153
154
155
156
157
158
159
 
# File 'lib/rocket_job/cli.rb', line 150

def setup_logger
  SemanticLogger.add_appender(io: STDOUT, formatter: :color) unless quiet
  SemanticLogger.default_level = log_level.to_sym if log_level

  # Enable SemanticLogger signal handling for this process
  SemanticLogger.add_signal_handler

  ::Mongoid.logger       = SemanticLogger[::Mongoid]
  ::Mongo::Logger.logger = SemanticLogger[::Mongo]
end

#write_pidfile ⇒ Object

Create a PID file if requested



129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/rocket_job/cli.rb', line 129

def write_pidfile
  return unless pidfile

  pid = $PID
  File.open(pidfile, "w") { |f| f.puts(pid) }

  # Remove pidfile on exit
  at_exit do
    File.delete(pidfile) if pid == $PID
  end
end