Class: Fluent::Supervisor

Inherits:

Object

• Object
• Fluent::Supervisor

Defined in:

lib/fluent/supervisor.rb

Defined Under Namespace

Classes: LoggerInitializer

Class Method Summary

• .default_options ⇒ Object
• .load_config(path, params = {}) ⇒ Object

Instance Method Summary

• #initialize(opt) ⇒ Supervisor constructor

  A new instance of Supervisor.

• #options ⇒ Object
• #run_supervisor ⇒ Object
• #run_worker ⇒ Object

Constructor Details

#initialize(opt) ⇒ Supervisor

Returns a new instance of Supervisor.

# File 'lib/fluent/supervisor.rb', line 366

def initialize(opt)
  @daemonize = opt[:daemonize]
  @supervise = opt[:supervise]
  @standalone_worker= opt[:standalone_worker]
  @config_path = opt[:config_path]
  @inline_config = opt[:inline_config]
  @use_v1_config = opt[:use_v1_config]
  @log_path = opt[:log_path]
  @dry_run = opt[:dry_run]
  @show_plugin_config = opt[:show_plugin_config]
  @libs = opt[:libs]
  @plugin_dirs = opt[:plugin_dirs]
  @chgroup = opt[:chgroup]
  @chuser = opt[:chuser]
  @rpc_server = nil
  @process_name = nil

  @log_level = opt[:log_level]
  @suppress_interval = opt[:suppress_interval]
  @suppress_config_dump = opt[:suppress_config_dump]
  @without_source = opt[:without_source]
  @signame = opt[:signame]

  @suppress_repeated_stacktrace = opt[:suppress_repeated_stacktrace]
  log_opts = {suppress_repeated_stacktrace: @suppress_repeated_stacktrace}
  @log = LoggerInitializer.new(@log_path, @log_level, @chuser, @chgroup, log_opts)
  @finished = false
end

Class Method Details

.default_options ⇒ Object

# File 'lib/fluent/supervisor.rb', line 344

def self.default_options
  {
    config_path: Fluent::DEFAULT_CONFIG_PATH,
    plugin_dirs: [Fluent::DEFAULT_PLUGIN_DIR],
    log_level: Fluent::Log::LEVEL_INFO,
    log_path: nil,
    daemonize: nil,
    libs: [],
    setup_path: nil,
    chuser: nil,
    chgroup: nil,
    suppress_interval: 0,
    suppress_repeated_stacktrace: true,
    without_source: false,
    use_v1_config: true,
    supervise: true,
    standalone_worker: false,
    signame: nil,
    winsvcreg: nil,
  }
end

.load_config(path, params = {}) ⇒ Object

# File 'lib/fluent/supervisor.rb', line 203

def self.load_config(path, params = {})

  pre_loadtime = 0
  pre_loadtime = params['pre_loadtime'].to_i if params['pre_loadtime']
  pre_config_mtime = nil
  pre_config_mtime = params['pre_config_mtime'] if params['pre_config_mtime']
  config_mtime = File.mtime(path)

  # reuse previous config if last load time is within 5 seconds and mtime of the config file is not changed
  if Time.now - Time.at(pre_loadtime) < 5 and config_mtime == pre_config_mtime
    return params['pre_conf']
  end

  config_fname = File.basename(path)
  config_basedir = File.dirname(path)
  config_data = File.read(path)
  inline_config = params['inline_config']
  if inline_config == '-'
    config_data << "\n" << STDIN.read
  elsif inline_config
    config_data << "\n" << inline_config.gsub("\\n","\n")
  end
  fluentd_conf = Fluent::Config.parse(config_data, config_fname, config_basedir, params['use_v1_config'])
  system_config = SystemConfig.create(fluentd_conf)

  log_level = system_config.log_level || params['log_level']
  suppress_repeated_stacktrace = system_config.suppress_repeated_stacktrace || params['suppress_repeated_stacktrace']
  log_path = params['log_path']
  chuser = params['chuser']
  chgroup = params['chgroup']
  rpc_endpoint = system_config.rpc_endpoint
  enable_get_dump = system_config.enable_get_dump

  log_opts = {suppress_repeated_stacktrace: suppress_repeated_stacktrace}
  logger_initializer = Supervisor::LoggerInitializer.new(log_path, log_level, chuser, chgroup, log_opts)
  # this #init sets initialized logger to $log
  logger_initializer.init
  logger = $log

  command_sender = Fluent.windows? ? "pipe" : "signal"

  # ServerEngine's "daemonize" option is boolean, and path of pid file is brought by "pid_path"
  pid_path = params['daemonize']
  daemonize = !!params['daemonize']
  main_cmd = params['main_cmd']
  signame = params['signame']

  se_config = {
      worker_type: 'spawn',
      workers: 1,
      log_stdin: false,
      log_stdout: false,
      log_stderr: false,
      enable_heartbeat: true,
      auto_heartbeat: false,
      unrecoverable_exit_codes: [2],
      stop_immediately_at_unrecoverable_exit: true,
      logger: logger,
      log: logger.out,
      log_path: log_path,
      log_level: log_level,
      logger_initializer: logger_initializer,
      chuser: chuser,
      chgroup: chgroup,
      chumask: 0,
      suppress_repeated_stacktrace: suppress_repeated_stacktrace,
      daemonize: daemonize,
      rpc_endpoint: rpc_endpoint,
      enable_get_dump: enable_get_dump,
      windows_daemon_cmdline: [ServerEngine.ruby_bin_path,
                               File.join(File.dirname(__FILE__), 'daemon.rb'),
                               ServerModule.name,
                               WorkerModule.name,
                               path,
                               JSON.dump(params)],
      command_sender: command_sender,
      fluentd_conf: fluentd_conf,
      main_cmd: main_cmd,
      signame: signame,
  }
  if daemonize
    se_config[:pid_path] = pid_path
  end
  pre_params = params.dup
  params['pre_loadtime'] = Time.now.to_i
  params['pre_config_mtime'] = config_mtime
  params['pre_conf'] = se_config
  # prevent pre_conf from being too big by reloading many times.
  pre_params['pre_conf'] = nil
  params['pre_conf'][:windows_daemon_cmdline][5] = JSON.dump(pre_params)

  return se_config
end

Instance Method Details

#options ⇒ Object

# File 'lib/fluent/supervisor.rb', line 405

def options
  {
    'config_path' => @config_path,
    'pid_file' => @daemonize,
    'plugin_dirs' => @plugin_dirs,
    'log_path' => @log_path
  }
end

#run_supervisor ⇒ Object

# File 'lib/fluent/supervisor.rb', line 395

def run_supervisor
  @log.init
  show_plugin_config if @show_plugin_config
  read_config
  set_system_config

  dry_run if @dry_run
  supervise
end

#run_worker ⇒ Object

# File 'lib/fluent/supervisor.rb', line 414

def run_worker
  begin
    require 'sigdump/setup'
  rescue Exception
    # ignore LoadError and others (related with signals): it may raise these errors in Windows
  end
  @log.init
  Process.setproctitle("worker:#{@process_name}") if @process_name

  show_plugin_config if @show_plugin_config
  read_config
  set_system_config

  install_main_process_signal_handlers

  $log.info "starting fluentd-#{Fluent::VERSION} without supervision"

  main_process do
    create_socket_manager if @standalone_worker
    change_privilege
    init_engine
    run_configure
    run_engine
    exit 0
  end
end