Class: Fluent::Supervisor

Inherits:

Object

• Object
• Fluent::Supervisor

Defined in:

lib/fluent/supervisor.rb

Defined Under Namespace

Classes: LoggerInitializer

Class Method Summary

• .cleanup_resources ⇒ Object
• .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 408

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

  @workers = opt[:workers]
  @root_dir = opt[:root_dir]
  @log_level = opt[:log_level]
  @log_rotate_age = opt[:log_rotate_age]
  @log_rotate_size = opt[:log_rotate_size]
  @suppress_interval = opt[:suppress_interval]
  @suppress_config_dump = opt[:suppress_config_dump]
  @log_event_verbose = opt[:log_event_verbose]
  @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,
    log_rotate_age: @log_rotate_age,
    log_rotate_size: @log_rotate_size
  )
  @finished = false
end

Class Method Details

.cleanup_resources ⇒ Object

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

def self.cleanup_resources
  unless Fluent.windows?
    if ENV.has_key?('SERVERENGINE_SOCKETMANAGER_PATH')
      FileUtils.rm_f(ENV['SERVERENGINE_SOCKETMANAGER_PATH'])
    end
  end
end

.default_options ⇒ Object

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

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,
    root_dir: nil,
    suppress_interval: 0,
    suppress_repeated_stacktrace: true,
    without_source: nil,
    use_v1_config: true,
    supervise: true,
    standalone_worker: false,
    signame: nil,
  }
end

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

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

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)
  # Assume fluent.conf encoding is UTF-8
  config_data = File.open(path, "r:utf-8:utf-8") {|f| f.read }
  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)

  # these params must NOT be configured via system config here.
  # these may be overridden by command line params.
  workers = params['workers']
  root_dir = params['root_dir']
  log_level = params['log_level']
  suppress_repeated_stacktrace = params['suppress_repeated_stacktrace']

  log_path = params['log_path']
  chuser = params['chuser']
  chgroup = params['chgroup']
  log_rotate_age = params['log_rotate_age']
  log_rotate_size = params['log_rotate_size']
  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,
    log_rotate_age: log_rotate_age,
    log_rotate_size: log_rotate_size
  )
  # this #init sets initialized logger to $log
  logger_initializer.init(:supervisor, 0)
  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: workers,
      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,
      root_dir: root_dir,
      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 477

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

#run_supervisor ⇒ Object

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

def run_supervisor
  @log.init(:supervisor, 0)
  show_plugin_config if @show_plugin_config
  read_config
  set_system_config
  @log.apply_options(format: @system_config.log.format, time_format: @system_config.log.time_format)

  $log.info :supervisor, "parsing config file is succeeded", path: @config_path

  if @workers < 1
    raise Fluent::ConfigError, "invalid number of workers (must be > 0):#{@workers}"
  end

  if @root_dir
    if File.exist?(@root_dir)
      unless Dir.exist?(@root_dir)
        raise Fluent::InvalidRootDirectory, "non directory entry exists:#{@root_dir}"
      end
    else
      begin
        FileUtils.mkdir_p(@root_dir)
      rescue => e
        raise Fluent::InvalidRootDirectory, "failed to create root directory:#{@root_dir}, #{e.inspect}"
      end
    end
  end

  dry_run_cmd if @dry_run
  supervise
end

#run_worker ⇒ Object

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

def run_worker
  begin
    require 'sigdump/setup'
  rescue Exception
    # ignore LoadError and others (related with signals): it may raise these errors in Windows
  end
  worker_id = ENV['SERVERENGINE_WORKER_ID'].to_i
  process_type = case
                 when @standalone_worker then :standalone
                 when worker_id == 0 then :worker0
                 else :workers
                 end
  @log.init(process_type, worker_id)
  show_plugin_config if @show_plugin_config
  read_config
  set_system_config
  @log.apply_options(format: @system_config.log.format, time_format: @system_config.log.time_format)

  Process.setproctitle("worker:#{@process_name}") if @process_name

  if @standalone_worker && @workers != 1
    raise Fluent::ConfigError, "invalid number of workers (must be 1 or unspecified) with --no-supervisor: #{@workers}"
  end

  install_main_process_signal_handlers

  # This is the only log messsage for @standalone_worker
  $log.info "starting fluentd-#{Fluent::VERSION} without supervision", pid: Process.pid if @standalone_worker

  main_process do
    create_socket_manager if @standalone_worker
    change_privilege
    init_engine
    run_configure
    run_engine
    self.class.cleanup_resources if @standalone_worker
    exit 0
  end
end