From 1cbbd9a81b8576374bcc32454e959c42075bb399 Mon Sep 17 00:00:00 2001 From: Daijiro Fukuda Date: Thu, 10 Oct 2024 23:26:42 +0900 Subject: [PATCH] Restart without downtime Add a new feature: Update/Reload without downtime. 1. The supervisor receives SIGUSR2. 2. Spawn a new supervisor. 3. Take over shared sockets. 4. Launch new workers, and stop old processes in parallel. * Launch new workers with source-only mode * Limit to restart_without_downtime_ready? input plugin * Send SIGTERM to the old supervisor after 10s delay from 3. 5. The old supervisor stops and sends SIGRTMIN(34) to the new one. 6. The new workers run fully. Problem to solve: Updating Fluentd or reloading a config causes downtime. Plugins that receive data as a server, such as `in_udp`, `in_tcp`, and `in_syslog`, cannot receive data during this time. This means that the data sent by a client is lost during this time unless the client has a re-sending feature. This makes updating Fluentd or reloading a config difficult in some cases. Note: need these feature * https://github.com/fluent/fluentd/pull/4661 * https://github.com/treasure-data/serverengine/pull/146 Co-authored-by: Shizuo Fujita Signed-off-by: Daijiro Fukuda --- lib/fluent/engine.rb | 4 +- lib/fluent/plugin/in_syslog.rb | 4 ++ lib/fluent/plugin/in_tcp.rb | 4 ++ lib/fluent/plugin/in_udp.rb | 4 ++ lib/fluent/plugin/input.rb | 4 ++ lib/fluent/root_agent.rb | 47 +++++++++--- lib/fluent/supervisor.rb | 127 ++++++++++++++++++++++++++++----- 7 files changed, 167 insertions(+), 27 deletions(-) diff --git a/lib/fluent/engine.rb b/lib/fluent/engine.rb index 65db5c7a39..94d09aa334 100644 --- a/lib/fluent/engine.rb +++ b/lib/fluent/engine.rb @@ -49,7 +49,7 @@ def initialize attr_reader :root_agent, :system_config, :supervisor_mode - def init(system_config, supervisor_mode: false) + def init(system_config, supervisor_mode: false, start_in_parallel: false) @system_config = system_config @supervisor_mode = supervisor_mode @@ -58,7 +58,7 @@ def init(system_config, supervisor_mode: false) @log_event_verbose = system_config.log_event_verbose unless system_config.log_event_verbose.nil? - @root_agent = RootAgent.new(log: log, system_config: @system_config) + @root_agent = RootAgent.new(log: log, system_config: @system_config, start_in_parallel: start_in_parallel) self end diff --git a/lib/fluent/plugin/in_syslog.rb b/lib/fluent/plugin/in_syslog.rb index 28ad1c5dd9..b33a479aee 100644 --- a/lib/fluent/plugin/in_syslog.rb +++ b/lib/fluent/plugin/in_syslog.rb @@ -156,6 +156,10 @@ def multi_workers_ready? true end + def restart_without_downtime_ready? + true + end + def start super diff --git a/lib/fluent/plugin/in_tcp.rb b/lib/fluent/plugin/in_tcp.rb index bd2ea83e5b..224eecb043 100644 --- a/lib/fluent/plugin/in_tcp.rb +++ b/lib/fluent/plugin/in_tcp.rb @@ -101,6 +101,10 @@ def multi_workers_ready? true end + def restart_without_downtime_ready? + true + end + def start super diff --git a/lib/fluent/plugin/in_udp.rb b/lib/fluent/plugin/in_udp.rb index c2d436115f..006524a95c 100644 --- a/lib/fluent/plugin/in_udp.rb +++ b/lib/fluent/plugin/in_udp.rb @@ -65,6 +65,10 @@ def multi_workers_ready? true end + def restart_without_downtime_ready? + true + end + def start super diff --git a/lib/fluent/plugin/input.rb b/lib/fluent/plugin/input.rb index 7a6909f7a9..7fb9b3608a 100644 --- a/lib/fluent/plugin/input.rb +++ b/lib/fluent/plugin/input.rb @@ -70,6 +70,10 @@ def metric_callback(es) def multi_workers_ready? false end + + def restart_without_downtime_ready? + false + end end end end diff --git a/lib/fluent/root_agent.rb b/lib/fluent/root_agent.rb index 8411152da4..f5870feaa2 100644 --- a/lib/fluent/root_agent.rb +++ b/lib/fluent/root_agent.rb @@ -48,7 +48,35 @@ module Fluent class RootAgent < Agent ERROR_LABEL = "@ERROR".freeze # @ERROR is built-in error label - def initialize(log:, system_config: SystemConfig.new) + class SourceOnlyMode + DISABELD = 0 + NORMAL = 1 + RESTART_WITHOUT_DOWNTIME_READY_ONLY = 2 + + def initialize(with_source_only, start_in_parallel) + if start_in_parallel + @mode = RESTART_WITHOUT_DOWNTIME_READY_ONLY + elsif with_source_only + @mode = NORMAL + else + @mode = DISABELD + end + end + + def source_only? + @mode != DISABELD + end + + def restart_without_downtime_ready_only? + @mode == RESTART_WITHOUT_DOWNTIME_READY_ONLY + end + + def disable! + @mode = DISABELD + end + end + + def initialize(log:, system_config: SystemConfig.new, start_in_parallel: false) super(log: log) @labels = {} @@ -56,7 +84,7 @@ def initialize(log:, system_config: SystemConfig.new) @suppress_emit_error_log_interval = 0 @next_emit_error_log_time = nil @without_source = system_config.without_source || false - @with_source_only = system_config.with_source_only || false + @source_only_mode = SourceOnlyMode.new(system_config.with_source_only, start_in_parallel) @source_only_buffer_agent = nil @enable_input_metrics = system_config.enable_input_metrics || false @@ -67,7 +95,7 @@ def initialize(log:, system_config: SystemConfig.new) attr_reader :labels def source_only_router - raise "[BUG] 'RootAgent#source_only_router' should not be called when 'with_source_only' is false" unless @with_source_only + raise "[BUG] 'RootAgent#source_only_router' should not be called when 'with_source_only' is false" unless @source_only_mode.source_only? @source_only_buffer_agent.event_router end @@ -154,7 +182,7 @@ def configure(conf) super - setup_source_only_buffer_agent if @with_source_only + setup_source_only_buffer_agent if @source_only_mode.source_only? # initialize elements if @without_source @@ -184,7 +212,7 @@ def setup_source_only_buffer_agent(flush: false) def lifecycle(desc: false, kind_callback: nil, kind_or_agent_list: nil) unless kind_or_agent_list - if @with_source_only + if @source_only_mode.source_only? kind_or_agent_list = [:input, @source_only_buffer_agent] elsif @source_only_buffer_agent # source_only_buffer_agent can re-reroute events, so the priority is equal to output_with_router. @@ -210,6 +238,9 @@ def lifecycle(desc: false, kind_callback: nil, kind_or_agent_list: nil) end display_kind = (kind == :output_with_router ? :output : kind) list.each do |instance| + if @source_only_mode.restart_without_downtime_ready_only? + next unless instance.restart_without_downtime_ready? + end yield instance, display_kind end end @@ -254,9 +285,9 @@ def flush! def cancel_source_only! # TODO exclusive lock - if @with_source_only + if @source_only_mode.source_only? log.info "cancel --with-source-only mode and start the other plugins" - @with_source_only = false + @source_only_mode.disable! start lifecycle_control_list[:input].each(&:event_emitter_cancel_source_only) @@ -371,7 +402,7 @@ def add_source(type, conf) # See also 'fluentd/plugin/input.rb' input.context_router = @event_router input.configure(conf) - input.event_emitter_set_source_only if @with_source_only + input.event_emitter_set_source_only if @source_only_mode.source_only? if @enable_input_metrics @event_router.add_metric_callbacks(input.plugin_id, Proc.new {|es| input.metric_callback(es) }) end diff --git a/lib/fluent/supervisor.rb b/lib/fluent/supervisor.rb index c95ab125fa..851030e0e5 100644 --- a/lib/fluent/supervisor.rb +++ b/lib/fluent/supervisor.rb @@ -43,6 +43,10 @@ def before_run @rpc_endpoint = nil @rpc_server = nil @counter = nil + @socket_manager_server = nil + @starting_new_supervisor_without_downtime = false + @new_supervisor_pid = nil + start_in_parallel = ENV.key?("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD") @fluentd_lock_dir = Dir.mktmpdir("fluentd-lock-") ENV['FLUENTD_LOCK_DIR'] = @fluentd_lock_dir @@ -65,10 +69,21 @@ def before_run if config[:disable_shared_socket] $log.info "shared socket for multiple workers is disabled" + elsif start_in_parallel + begin + raise "[BUG] SERVERENGINE_SOCKETMANAGER_PATH env var must exist when starting in parallel" unless ENV.key?('SERVERENGINE_SOCKETMANAGER_PATH') + @socket_manager_server = ServerEngine::SocketManager::Server.take_over_another_server(ENV['SERVERENGINE_SOCKETMANAGER_PATH']) + $log.info "restart-without-downtime: took over the shared sockets", path: ENV['SERVERENGINE_SOCKETMANAGER_PATH'] + rescue => e + $log.error "restart-without-downtime: cancel sequence because failed to take over the shared sockets", error: e + raise + end else - server = ServerEngine::SocketManager::Server.open - ENV['SERVERENGINE_SOCKETMANAGER_PATH'] = server.path.to_s + @socket_manager_server = ServerEngine::SocketManager::Server.open + ENV['SERVERENGINE_SOCKETMANAGER_PATH'] = @socket_manager_server.path.to_s end + + stop_parallel_old_supervisor_after_delay if start_in_parallel end def after_run @@ -76,7 +91,9 @@ def after_run stop_rpc_server if @rpc_endpoint stop_counter_server if @counter cleanup_lock_dir - Fluent::Supervisor.cleanup_resources + Fluent::Supervisor.cleanup_socketmanager_path unless @starting_new_supervisor_without_downtime + + notify_new_supervisor_that_old_one_has_stopped if @starting_new_supervisor_without_downtime end def cleanup_lock_dir @@ -138,7 +155,7 @@ def run_rpc_server @rpc_server.mount_proc('/api/config.gracefulReload') { |req, res| $log.debug "fluentd RPC got /api/config.gracefulReload request" if Fluent.windows? - supervisor_sigusr2_handler + graceful_reload else Process.kill :USR2, Process.pid end @@ -172,6 +189,47 @@ def stop_counter_server @counter.stop end + def stop_parallel_old_supervisor_after_delay + # TODO if the new supervisor fails to start and this is not called, + # it would be necessary to update the pid in the PID file to the old one when daemonized. + + Thread.new do + # Delay to avoid worker downtime as much as possible. + # Even if the downtime occurs, it is no problem because the socket buffer works, + # as long as the capacity is not exceeded. + sleep 10 + old_pid = ENV["FLUENT_RUNNING_IN_PARALLEL_WITH_OLD"]&.to_i + if old_pid + $log.info "restart-without-downtime: stop the old supervisor" + Process.kill :TERM, old_pid + end + rescue => e + $log.warn "restart-without-downtime: failed to stop the old supervisor." + + " If the old one does not exist, please send '34' signal to this new process to start to work fully." + + " If it exists, something went wrong. Please kill the old one manually.", + error: e + end + end + + def notify_new_supervisor_that_old_one_has_stopped + if config[:pid_path] + new_pid = File.read(config[:pid_path]).to_i + else + raise "[BUG] new_supervisor_pid is not saved" unless @new_supervisor_pid + new_pid = @new_supervisor_pid + end + + $log.info "restart-without-downtime: notify the new supervisor (pid: #{new_pid}) that old one has stopped" + Process.kill 34, new_pid + rescue => e + $log.error( + "restart-without-downtime: failed to notify the new supervisor." + + " Please send '34' signal to the new supervisor process manually" + + " if it does not start to work fully.", + error: e + ) + end + def install_supervisor_signal_handlers return if Fluent.windows? @@ -187,7 +245,11 @@ def install_supervisor_signal_handlers trap :USR2 do $log.debug 'fluentd supervisor process got SIGUSR2' - supervisor_sigusr2_handler + if Fluent.windows? + graceful_reload + else + restart_without_downtime + end end trap 34 do @@ -259,7 +321,7 @@ def install_windows_event_handler when :usr1 supervisor_sigusr1_handler when :usr2 - supervisor_sigusr2_handler + graceful_reload when :cont supervisor_dump_handler_for_windows when :stop_event_thread @@ -289,7 +351,7 @@ def supervisor_sigusr1_handler send_signal_to_workers(:USR1) end - def supervisor_sigusr2_handler + def graceful_reload conf = nil t = Thread.new do $log.info 'Reloading new config' @@ -317,7 +379,38 @@ def supervisor_sigusr2_handler $log.error "Failed to reload config file: #{e}" end + def restart_without_downtime + # TODO exclusive lock + + $log.info "start restart-without-downtime sequence" + + if @starting_new_supervisor_without_downtime + $log.warn "restart-without-downtime: canceled because it is already starting" + return + end + if ENV.key?("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD") + $log.warn "restart-without-downtime: canceled because the previous sequence is still running" + return + end + + @starting_new_supervisor_without_downtime = true + commands = [ServerEngine.ruby_bin_path, $0] + ARGV + env_to_add = { + "SERVERENGINE_SOCKETMANAGER_INTERNAL_TOKEN" => ServerEngine::SocketManager::INTERNAL_TOKEN, + "FLUENT_RUNNING_IN_PARALLEL_WITH_OLD" => "#{Process.pid}", + } + pid = Process.spawn(env_to_add, commands.join(" ")) + @new_supervisor_pid = pid unless config[:daemonize] + rescue => e + $log.error "restart-without-downtime: failed", error: e + @starting_new_supervisor_without_downtime = false + end + def cancel_source_only + if ENV.key?("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD") + $log.info "restart-without-downtime: done all sequences, now the new workers starts to work fully" + ENV.delete("FLUENT_RUNNING_IN_PARALLEL_WITH_OLD") + end send_signal_to_workers(34) end @@ -509,12 +602,11 @@ def self.default_options } end - def self.cleanup_resources - unless Fluent.windows? - if ENV.has_key?('SERVERENGINE_SOCKETMANAGER_PATH') - FileUtils.rm_f(ENV['SERVERENGINE_SOCKETMANAGER_PATH']) - end - end + def self.cleanup_socketmanager_path + return if Fluent.windows? + return unless ENV.key?('SERVERENGINE_SOCKETMANAGER_PATH') + + FileUtils.rm_f(ENV['SERVERENGINE_SOCKETMANAGER_PATH']) end def initialize(cl_opt) @@ -578,7 +670,7 @@ def run_supervisor(dry_run: false) 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) + 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 @@ -623,10 +715,10 @@ def run_worker File.umask(@chumask.to_i(8)) end MessagePackFactory.init(enable_time_support: @system_config.enable_msgpack_time_support) - Fluent::Engine.init(@system_config) + 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_resources if @standalone_worker + self.class.cleanup_socketmanager_path if @standalone_worker exit 0 end end @@ -844,7 +936,8 @@ def install_main_process_signal_handlers end trap :USR2 do - reload_config + # Do nothing + # TODO consider suitable code for this end trap :CONT do