Merge pull request #1359 from fluent/introduce-unrecoverable-errors

introduce unrecoverable error type, and logs to show Fluentd process/worker started

lib/fluent/engine.rb

@@ -172,13 +172,15 @@ def log_event_loop
 
     def run
       begin
+        $log.info "starting fluentd worker", pid: Process.pid, ppid: Process.ppid # TODO: worker number
         start
 
         if @event_router.match?($log.tag)
           $log.enable_event
           @log_emit_thread = Thread.new(&method(:log_event_loop))
         end
 
+        $log.info "fluentd worker is now running" # TODO: worker number
         sleep MAINLOOP_SLEEP_INTERVAL until @engine_stopped
 
       rescue Exception => e
@@ -187,7 +189,7 @@ def run
         raise
       end
 
-      $log.info "shutting down fluentd"
+      $log.info "shutting down fluentd worker" # TODO: worker number
       shutdown
       if @log_emit_thread
         @log_event_loop_stop = true

lib/fluent/error.rb

@@ -0,0 +1,30 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+module Fluent
+  class UnrecoverableError < StandardError
+    def initialize(error_message = nil)
+      @message = error_message || "an unrecoverable error occurs in Fluentd process"
+    end
+
+    def to_s
+      @message
+    end
+  end
+
+  class InvalidRootDirectory < UnrecoverableError
+  end
+end

lib/fluent/supervisor.rb

@@ -20,6 +20,7 @@
 require 'fluent/config'
 require 'fluent/env'
 require 'fluent/engine'
+require 'fluent/error'
 require 'fluent/log'
 require 'fluent/plugin'
 require 'fluent/rpc'
@@ -191,7 +192,7 @@ def supervisor_get_dump_config_handler
   module WorkerModule
     def spawn(process_manager)
       main_cmd = config[:main_cmd]
-      @pm = process_manager.spawn(main_cmd)
+      @pm = process_manager.spawn(*main_cmd)
     end
 
     def after_start
@@ -444,7 +445,7 @@ def run_worker
 
       install_main_process_signal_handlers
 
-      $log.info "starting fluentd-#{Fluent::VERSION} without supervision"
+      $log.info "starting fluentd-#{Fluent::VERSION} without supervision", pid: Process.pid
 
       main_process do
         create_socket_manager if @standalone_worker
@@ -501,28 +502,16 @@ def show_plugin_config
     end
 
     def supervise
-      $log.info "starting fluentd-#{Fluent::VERSION}"
+      $log.info "starting fluentd-#{Fluent::VERSION}", pid: Process.pid
 
       rubyopt = ENV["RUBYOPT"]
-      if Fluent.windows?
-        # Shellwords doesn't work on windows, then used gsub for adapting space char instead of Shellwords
-        fluentd_spawn_cmd = ServerEngine.ruby_bin_path + " -Eascii-8bit:ascii-8bit "
-        fluentd_spawn_cmd << ' "' + rubyopt.gsub('"', '""') + '" ' if rubyopt
-        fluentd_spawn_cmd << ' "' + $0.gsub('"', '""') + '" '
-        $fluentdargv.each{|a|
-          fluentd_spawn_cmd << ('"' + a.gsub('"', '""') + '" ')
-        }
-      else
-        fluentd_spawn_cmd = ServerEngine.ruby_bin_path + " -Eascii-8bit:ascii-8bit "
-        fluentd_spawn_cmd << ' ' + rubyopt + ' ' if rubyopt
-        fluentd_spawn_cmd << $0.shellescape + ' '
-        $fluentdargv.each{|a|
-          fluentd_spawn_cmd << (a.shellescape + " ")
-        }
-      end
+      fluentd_spawn_cmd = [ServerEngine.ruby_bin_path, "-Eascii-8bit:ascii-8bit"]
+      fluentd_spawn_cmd << rubyopt if rubyopt
+      fluentd_spawn_cmd << $0
+      fluentd_spawn_cmd += $fluentdargv
+      fluentd_spawn_cmd << "--under-supervisor"
 
-      fluentd_spawn_cmd << ("--under-supervisor")
-      $log.info "spawn command to main: " + fluentd_spawn_cmd
+      $log.info "spawn command to main: ", cmdline: fluentd_spawn_cmd
 
       params = {}
       params['main_cmd'] = fluentd_spawn_cmd
@@ -617,39 +606,54 @@ def flush_buffer
       }.run
     end
 
+    def logging_with_console_output
+      yield $log
+      unless @log.stdout?
+        logger = ServerEngine::DaemonLogger.new(STDOUT)
+        log = Fluent::Log.new(logger)
+        log.level = @log_level
+        console = log.enable_debug
+        yield console
+      end
+    end
+
     def main_process(&block)
       Process.setproctitle("worker:#{@process_name}") if @process_name
 
-      configuration_error = false
+      unrecoverable_error = false
 
       begin
         block.call
-      rescue Fluent::ConfigError
-        $log.error "config error", file: @config_path, error: $!.to_s
-        $log.debug_backtrace
-        unless @log.stdout?
-          logger = ServerEngine::DaemonLogger.new(STDOUT)
-          log = Fluent::Log.new(logger)
-          log.level = @log_level
-          console = log.enable_debug
-          console.error "config error", file: @config_path, error: $!.to_s
-          console.debug_backtrace
+      rescue Fluent::ConfigError => e
+        logging_with_console_output do |log|
+          log.error "config error", file: @config_path, error: e
+          log.debug_backtrace
+        end
+        unrecoverable_error = true
+      rescue Fluent::UnrecoverableError => e
+        logging_with_console_output do |log|
+          log.error e.message, error: e
+          log.error_backtrace
+        end
+        unrecoverable_error = true
+      rescue ScriptError => e # LoadError, NotImplementedError, SyntaxError
+        logging_with_console_output do |log|
+          if e.respond_to?(:path)
+            log.error e.message, path: e.path, error: e
+          else
+            log.error e.message, error: e
+          end
+          log.error_backtrace
         end
-        configuration_error = true
-      rescue
-        $log.error "unexpected error", error: $!.to_s
-        $log.error_backtrace
-        unless @log.stdout?
-          logger = ServerEngine::DaemonLogger.new(STDOUT)
-          log = Fluent::Log.new(logger)
-          log.level = @log_level
-          console = log.enable_debug
-          console.error "unexpected error", error: $!.to_s
-          console.error_backtrace
+        unrecoverable_error = true
+      rescue => e
+        logging_with_console_output do |log|
+          log.error "unexpected error", error: e
+          log.error_backtrace
         end
       end
 
-      exit!(configuration_error ? 2 : 1)
+      exit!(unrecoverable_error ? 2 : 1)
     end
 
     def read_config