Use an error_flow to log any exceptions during logging

During normal operation, the Flows will rescue all exceptions and log
them to the special error_flow. By default, we will write JSON logs to
STDERR.

The log location and format can either be change globally by setting (or
changing) the Rackstash.error_flow or for each Flow for a Logger
individually by setting Flow#error_flow.

lib/rackstash.rb

@@ -86,6 +86,35 @@ module Rackstash
       SEVERITY_LABELS[severity]
     end
   end
+
+  # Returns a {Flow} which is used by the normal logger {Flow}s to write details
+  # about any unexpected errors during interaction with their {Adapters}.
+  #
+  # By default, this Flow logs JSON-formatted messages to `STDERR`
+  #
+  # @return [Rackstash::Flow] the default error flow
+  def self.error_flow
+    @error_flow ||= Rackstash::Flow.new(STDERR)
+  end
+
+  # Set a {Flow} which is used bythe normal logger {Flow}s to write details
+  # of any unexpected errors during interaction with their {Adapters}.
+  #
+  # You can set a different `error_flow` for each {Flow} if required. You can
+  # also change this flow to match your desired fallback format and log adapter.
+  #
+  # To still work in the face of unexpected availability issues like a full
+  # filesystem, an unavailable network, broken external loggers, or any other
+  # external issues, it is usually desireable to chose a local and mostly
+  # relibable log target.
+  #
+  # @param flow [Flow, Adapters::Adapter, Object] a single {Flow} or an object
+  #   which can be used as a {Flow}'s adapter. See {Flow#initialize}.
+  # @return [Rackstash::Flow] the given `flow`
+  def self.error_flow=(flow)
+    flow = Flow.new(flow) unless flow.is_a?(Rackstash::Flow)
+    @error_flow = flow
+  end
 end
 
 require 'rackstash/logger'

lib/rackstash/flow.rb

@@ -76,11 +76,17 @@ module Rackstash
     # @yieldparam flow [self] if the given block accepts an argument, we yield
     #   `self` as a parameter, else, the block is directly executed in the
     #   context of `self`.
-    def initialize(adapter, encoder: nil, filters: [], &block)
+    def initialize(adapter, encoder: nil, filters: [], error_flow: nil, &block)
       @adapter = Rackstash::Adapters[adapter]
       self.encoder(encoder || @adapter.default_encoder)
       @filter_chain = Rackstash::FilterChain.new(filters)
 
+      if error_flow.nil?
+        @error_flow = nil
+      else
+        self.error_flow(error_flow)
+      end
+
       if block_given?
         if block.arity == 0
           instance_eval(&block)
@@ -126,6 +132,15 @@ module Rackstash
       @encoder = encoder
     end
 
+    def error_flow(flow = nil)
+      if flow.nil?
+        @error_flow || Rackstash.error_flow
+      else
+        flow = Flow.new(flow) unless flow.is_a?(Rackstash::Flow)
+        @error_flow = flow
+      end
+    end
+
     # (see FilterChain#insert_after)
     def filter_after(index, filter = nil, &block)
       @filter_chain.insert_after(index, filter, &block)
@@ -236,9 +251,25 @@ module Rackstash
 
     private
 
-    # TODO: use a fallback flow and send formatted logs there
     def log_error(message, exception)
-      warn("#{message}: #{exception}")
+      error_event = {
+        FIELD_ERROR => exception.class.name,
+        FIELD_ERROR_MESSAGE => exception.message,
+        FIELD_ERROR_TRACE => (exception.backtrace || []).join("\n"),
+
+        FIELD_TAGS => [],
+        FIELD_MESSAGE => message,
+        FIELD_TIMESTAMP => Time.now.utc.iso8601(ISO8601_PRECISION).freeze,
+        FIELD_VERSION => '1'.freeze
+      }
+      error_flow.write!(error_event)
+    rescue
+      # At this place, writing to the error log has also failed. This is a bad
+      # place to be in and there is very little we can sensibly do now.
+      #
+      # To aid in availability of the app using Rackstash, we swallow any
+      # StandardErrors here and just continue, hoping that things will turn out
+      # to be okay in the end.
     end
   end
 end

spec/rackstash/flow_spec.rb

@@ -75,9 +75,26 @@ describe Rackstash::Flow do
     end
 
     it 'rescues any exception thrown by the adapter' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!)
+        .with hash_including(
+          'message' => /^close failed for adapter/,
+          'error' => 'RuntimeError',
+          'error_message' => 'ERROR'
+        )
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
       expect(flow).to receive(:close!).and_raise('ERROR')
-      expect(flow).to receive(:warn).with(/^close failed for adapter/)
+      expect(flow.close).to be nil
-      flow.close
+    end
+
+    it 'rescues errors thrown by the error_flow' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!).and_raise('DOUBLE ERROR')
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
+      expect(flow).to receive(:close!).and_raise('ERROR')
+      expect(flow.close).to be nil
     end
   end
 
@@ -192,9 +209,26 @@ describe Rackstash::Flow do
     end
 
     it 'rescues any exception thrown by the adapter' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!)
+        .with hash_including(
+          'message' => /^reopen failed for adapter/,
+          'error' => 'RuntimeError',
+          'error_message' => 'ERROR'
+        )
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
       expect(flow).to receive(:reopen!).and_raise('ERROR')
-      expect(flow).to receive(:warn).with(/^reopen failed for adapter/)
+      expect(flow.reopen).to be nil
-      flow.reopen
+    end
+
+    it 'rescues errors thrown by the error_flow' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!).and_raise('DOUBLE ERROR')
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
+      expect(flow).to receive(:reopen!).and_raise('ERROR')
+      expect(flow.reopen).to be nil
     end
   end
 
@@ -263,9 +297,26 @@ describe Rackstash::Flow do
     end
 
     it 'rescues any exception thrown by the adapter' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!)
+        .with hash_including(
+          'message' => /^write failed for adapter/,
+          'error' => 'RuntimeError',
+          'error_message' => 'ERROR'
+        )
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
       expect(flow).to receive(:write!).and_raise('ERROR')
-      expect(flow).to receive(:warn).with(/^write failed for adapter/)
+      expect(flow.write(event)).to be false
-      flow.write(event)
+    end
+
+    it 'rescues errors thrown by the error_flow' do
+      error_flow = instance_double(described_class)
+      expect(error_flow).to receive(:write!).and_raise('DOUBLE ERROR')
+      expect(flow).to receive(:error_flow).and_return(error_flow)
+
+      expect(flow).to receive(:write!).and_raise('ERROR')
+      expect(flow.write(event)).to be false
     end
   end
 end

spec/rackstash_spec.rb

@@ -83,4 +83,38 @@ describe Rackstash do
       expect(described_class.severity_label(nil)).to eql 'ANY'
     end
   end
+
+  describe '.error_flow' do
+    it 'returns a default Flow' do
+      expect(described_class.error_flow).to be_instance_of Rackstash::Flow
+
+      expect(described_class.error_flow.encoder).to be_instance_of Rackstash::Encoders::JSON
+      expect(described_class.error_flow.adapter).to be_instance_of Rackstash::Adapters::IO
+    end
+
+    it 'caches the flow' do
+      expect(described_class.error_flow).to equal described_class.error_flow
+    end
+  end
+
+  describe '.error_flow=' do
+    let(:flow) {
+      flow = instance_double('Rackstash::Flow')
+      allow(flow).to receive(:is_a?).with(Rackstash::Flow).and_return(true)
+      flow
+    }
+
+    it 'can set a new flow' do
+      described_class.error_flow = flow
+      expect(described_class.error_flow).to equal flow
+    end
+
+    it 'wraps a non-flow' do
+      adapter = 'spec.log'
+      expect(Rackstash::Flow).to receive(:new).with(adapter).and_return(flow)
+
+      described_class.error_flow = adapter
+      expect(described_class.error_flow).to equal flow
+    end
+  end
 end