Class: Fluent::Supervisor

Inherits:

Object

• Object
• Fluent::Supervisor

Defined in:

lib/fluent/supervisor.rb

Class Method Summary

• .cleanup_socketmanager_path ⇒ Object
• .default_options ⇒ Object
• .serverengine_config(params = {}) ⇒ Object

Instance Method Summary

• #configure(supervisor: false) ⇒ Object
• #initialize(cl_opt) ⇒ Supervisor constructor

  A new instance of Supervisor.

• #options ⇒ Object
• #run_supervisor(dry_run: false) ⇒ Object
• #run_worker ⇒ Object

Constructor Details

#initialize(cl_opt) ⇒ Supervisor

Returns a new instance of Supervisor.

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

def initialize(cl_opt)
  @cl_opt = cl_opt
  opt = self.class.default_options.merge(cl_opt)

  @config_file_type = opt[:config_file_type]
  @daemonize = opt[:daemonize]
  @standalone_worker= opt[:standalone_worker]
  @config_path = opt[:config_path]
  @inline_config = opt[:inline_config]
  @use_v1_config = opt[:use_v1_config]
  @conf_encoding = opt[:conf_encoding]
  @show_plugin_config = opt[:show_plugin_config]
  @libs = opt[:libs]
  @plugin_dirs = opt[:plugin_dirs]
  @chgroup = opt[:chgroup]
  @chuser = opt[:chuser]
  @chumask = opt[:chumask]
  @signame = opt[:signame]

  # TODO: `@log_path`, `@log_rotate_age` and `@log_rotate_size` should be removed
  # since it should be merged with SystemConfig in `build_system_config()`.
  # We should always use `system_config.log.path`, `system_config.log.rotate_age`
  # and `system_config.log.rotate_size`.
  # However, currently, there is a bug that `system_config.log` parameters
  # are not in `Fluent::SystemConfig::SYSTEM_CONFIG_PARAMETERS`, and these
  # parameters are not merged in `build_system_config()`.
  # Until we fix the bug of `Fluent::SystemConfig`, we need to use these instance variables.
  @log_path = opt[:log_path]
  @log_rotate_age = opt[:log_rotate_age]
  @log_rotate_size = opt[:log_rotate_size]

  @finished = false
end

Class Method Details

.cleanup_socketmanager_path ⇒ Object

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

def self.cleanup_socketmanager_path
  return if Fluent.windows?
  return unless ENV.key?('SERVERENGINE_SOCKETMANAGER_PATH')

  FileUtils.rm_f(ENV['SERVERENGINE_SOCKETMANAGER_PATH'])
end

.default_options ⇒ Object

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

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,
    chumask: "0",
    root_dir: nil,
    suppress_interval: 0,
    suppress_repeated_stacktrace: true,
    ignore_repeated_log_interval: nil,
    without_source: nil,
    with_source_only: nil,
    enable_input_metrics: true,
    enable_size_metrics: nil,
    use_v1_config: true,
    strict_config_value: nil,
    supervise: true,
    standalone_worker: false,
    signame: nil,
    conf_encoding: 'utf-8',
    disable_shared_socket: nil,
    config_file_type: :guess,
  }
end

.serverengine_config(params = {}) ⇒ Object

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

def self.serverengine_config(params = {})
  # ServerEngine's "daemonize" option is boolean, and path of pid file is brought by "pid_path"
  pid_path = params['daemonize']
  daemonize = !!params['daemonize']

  se_config = {
    worker_type: 'spawn',
    workers: params['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: params['root_dir'],
    logger: $log,
    log: $log&.out,
    log_level: params['log_level'],
    chuser: params['chuser'],
    chgroup: params['chgroup'],
    chumask: params['chumask'].is_a?(Integer) ? params['chumask'] : params['chumask']&.to_i(8),
    daemonize: daemonize,
    rpc_endpoint: params['rpc_endpoint'],
    counter_server: params['counter_server'],
    enable_get_dump: params['enable_get_dump'],
    windows_daemon_cmdline: [ServerEngine.ruby_bin_path,
                             File.join(File.dirname(__FILE__), 'daemon.rb'),
                             ServerModule.name,
                             WorkerModule.name,
                             JSON.dump(params)],
    command_sender: Fluent.windows? ? "pipe" : "signal",
    config_path: params['fluentd_conf_path'],
    fluentd_conf: params['fluentd_conf'],
    conf_encoding: params['conf_encoding'],
    inline_config: params['inline_config'],
    main_cmd: params['main_cmd'],
    signame: params['signame'],
    disable_shared_socket: params['disable_shared_socket'],
    restart_worker_interval: params['restart_worker_interval'],
  }
  se_config[:pid_path] = pid_path if daemonize

  se_config
end

Instance Method Details

#configure(supervisor: false) ⇒ Object

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

def configure(supervisor: false)
  setup_global_logger(supervisor: supervisor)

  if @show_plugin_config
    show_plugin_config
  end

  if @inline_config == '-'
    $log.warn('the value "-" for `inline_config` is deprecated. See https://github.com/fluent/fluentd/issues/2711')
    @inline_config = STDIN.read
  end
  parsed_files = Set.new
  @conf = Fluent::Config.build(
    config_path: @config_path,
    encoding: @conf_encoding,
    additional_config: @inline_config,
    use_v1_config: @use_v1_config,
    type: @config_file_type,
    on_file_parsed: ->(path) { parsed_files << path },
  )
  @system_config = build_system_config(@conf)

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

  build_additional_configurations(parsed_files) do |additional_conf|
    @conf += additional_conf
  end

  if Fluent.windows?
    @conf.elements.each do |element|
      next unless element.name == 'source'
      if element['@type'] == 'tail' && element['pos_file']
        $log.warn("Recommend adding #{File.dirname(element['pos_file'])} to the exclusion path of your antivirus software on Windows")
      else
        storage = element.elements.find { |v| v.name == 'storage' and v['@type'] == 'local' }
        if storage
          $log.warn("Recommend adding #{File.dirname(storage['path'])} to the exclusion path of your antivirus software on Windows")
        end
      end
    end
  end

  @libs.each do |lib|
    require lib
  end

  @plugin_dirs.each do |dir|
    if Dir.exist?(dir)
      dir = File.expand_path(dir)
      Fluent::Plugin.add_plugin_dir(dir)
    end
  end

  if supervisor
    # plugins / configuration dumps
    Gem::Specification.find_all.select { |x| x.name =~ /^fluent(d|-(plugin|mixin)-.*)$/ }.each do |spec|
      $log.info("gem '#{spec.name}' version '#{spec.version}'")
    end
  end
end

#options ⇒ Object

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

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

#run_supervisor(dry_run: false) ⇒ Object

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

def run_supervisor(dry_run: false)
  if dry_run
    $log.info "starting fluentd-#{Fluent::VERSION} as dry run mode", ruby: RUBY_VERSION
  end

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

  if Fluent.windows? && @system_config.with_source_only
    raise Fluent::ConfigError, "with-source-only is not supported on Windows"
  end

  root_dir = @system_config.root_dir
  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, mode: @system_config.dir_permission || Fluent::DEFAULT_DIR_PERMISSION)
      rescue => e
        raise Fluent::InvalidRootDirectory, "failed to create root directory:#{root_dir}, #{e.inspect}"
      end
    end
  end

  begin
    ServerEngine::Privilege.change(@chuser, @chgroup)
    MessagePackFactory.init(enable_time_support: @system_config.enable_msgpack_time_support)
    Fluent::Engine.init(@system_config, supervisor_mode: true, start_in_parallel: ENV.key?("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD"))
    Fluent::Engine.run_configure(@conf, dry_run: dry_run)
  rescue Fluent::ConfigError => e
    $log.error 'config error', file: @config_path, error: e
    $log.debug_backtrace
    exit!(1)
  rescue ScriptError => e # LoadError, NotImplementedError, SyntaxError
    if e.respond_to?(:path)
      $log.error e.message, path: e.path, error: e
    else
      $log.error e.message, error: e
    end
    $log.debug_backtrace
    exit!(1)
  rescue => e
    $log.error "unexpected error", error: e
    $log.debug_backtrace
    exit!(1)
  end

  if dry_run
    $log.info 'finished dry run mode'
    exit 0
  else
    supervise
  end
end

#run_worker ⇒ Object

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

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

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

  if Fluent.windows? && @system_config.with_source_only
    raise Fluent::ConfigError, "with-source-only is not supported on Windows"
  end

  install_main_process_signal_handlers

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

  main_process do
    create_socket_manager if @standalone_worker
    if @standalone_worker
      ServerEngine::Privilege.change(@chuser, @chgroup)
      File.umask(@chumask.to_i(8))
    end
    MessagePackFactory.init(enable_time_support: @system_config.enable_msgpack_time_support)
    Fluent::Engine.init(@system_config, start_in_parallel: ENV.key?("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD"))
    Fluent::Engine.run_configure(@conf)
    Fluent::Engine.run
    self.class.cleanup_socketmanager_path if @standalone_worker
    exit 0
  end
end