Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 43 additions & 19 deletions logstash-core/lib/logstash/java_pipeline.rb
Original file line number Diff line number Diff line change
Expand Up @@ -137,9 +137,7 @@ def start

@thread = Thread.new do
error_log_params = ->(e) {
default_logging_keys(
:exception => e,
:backtrace => e.backtrace,
exception_logging_keys(e,
"pipeline.sources" => pipeline_source_details
)
}
Expand Down Expand Up @@ -312,7 +310,7 @@ def start_workers
@crash_detected.make_true
@logger.error(
"Pipeline worker error, the pipeline will be stopped",
default_logging_keys(:error => e.cause.message, :exception => e.cause.class, :backtrace => e.cause.backtrace)
exception_logging_keys(e.cause)
)
end
end
Expand Down Expand Up @@ -422,21 +420,14 @@ def inputworker(plugin)
if plugin.stop?
@logger.debug(
"Input plugin raised exception during shutdown, ignoring it.",
default_logging_keys(
:plugin => plugin.class.config_name,
:exception => e.message,
:backtrace => e.backtrace))
exception_logging_keys(e, :plugin => plugin.class.config_name))
return
end

# otherwise, report error and restart
@logger.error(I18n.t(
"logstash.pipeline.worker-error-debug",
**default_logging_keys(
:plugin => plugin.inspect,
:error => e.message,
:exception => e.class,
:stacktrace => e.backtrace.join("\n"))))
**exception_logging_keys(e, :plugin => plugin.inspect)))

# Assuming the failure that caused this exception is transient,
# let's sleep for a bit and execute #run again
Expand Down Expand Up @@ -580,10 +571,7 @@ def close_plugin_and_ignore(plugin)
rescue => e
@logger.warn(
"plugin raised exception while closing, ignoring",
default_logging_keys(
:plugin => plugin.class.config_name,
:exception => e.message,
:backtrace => e.backtrace))
exception_logging_keys(e, :plugin => plugin.class.config_name))
end
end

Expand All @@ -607,12 +595,12 @@ def init_worker_loop
rescue => e
@logger.error(
"Worker loop initialization error",
default_logging_keys(:error => e.message, :exception => e.class, :stacktrace => e.backtrace.join("\n")))
exception_logging_keys(e))
nil
rescue Java::java.lang.StackOverflowError => se
@logger.error(
"Stack overflow error while compiling Pipeline. Please increase thread stack size using -Xss",
default_logging_keys())
exception_logging_keys(se))
nil
end
end
Expand All @@ -630,6 +618,42 @@ def default_logging_keys(other_keys = {})
keys
end

def exception_logging_keys(active_exception = $!, other_keys = {})
base = active_exception ? unwind_cause_chain(active_exception) : {}
default_logging_keys(base.merge(other_keys))
end

##
# Yields once per exception in the provided exception's cause chain
# @param exception [Exception]
# @yield_param [Exception]
def cause_chain(exception, &block)
return enum_for(:cause_chain, exception) unless block_given?

current = exception
while !current.nil?
yield current
cause = current.cause
current = (cause == current ? nil : cause)
end
end

##
# unwinds the provided exception to create a deeply-nested structure
# representing the exception and its cause chain
# @param [Exception] exception
# @return [Hash{Symbol=>[String|Hash]}]
def unwind_cause_chain(exception)
cause_chain(exception).reverse_each.reduce(nil) do |cause, exception|
{
:exception => exception.class.name,
:error => exception.message,
:stacktrace => exception.backtrace.join("\n"),
:cause => cause,
}.compact
end
end

def preserve_event_order?(pipeline_workers)
case settings.get("pipeline.ordered")
when "auto"
Expand Down