Add fields and tags to the Buffer

Each buffer instance can hold messages, fields and tags. These together
form the log event which will eventually be written to the log target.

By adding fields and tags, you can add highly details structured
information to your logs which allow to filter and analyze the logs
without having to parse complex multi-line logs.

lib/rackstash.rb

@@ -24,6 +24,11 @@ module Rackstash
 
   # How many decimal places to render on ISO 8601 timestamps
   ISO8601_PRECISION = 3
+
+  FIELD_MESSAGE = 'message'.freeze
+  FIELD_TAGS = 'tags'.freeze
+  FIELD_TIMESTAMP = '@timestamp'.freeze
+  FIELD_VERSION = '@version'.freeze
 end
 
 require 'rackstash/logger'

lib/rackstash/buffer.rb

@@ -3,21 +3,105 @@
 # This software may be modified and distributed under the terms
 # of the MIT license. See the LICENSE.txt file for details.
 
+require 'rackstash/fields'
+
 module Rackstash
   # The Buffer holds all the data of a single log event. It can hold multiple
   # messages of multiple calls to the log, additional fields holding structured
   # data about the log event, and tags identiying the type of log.
   class Buffer
-    def initialize
+    # A set of field names which are forbidden from being set as fields. The
+    # fields mentioned here are all either statically set or are accessed by
+    # specialized accessor methods.
+    FORBIDDEN_FIELDS = Set[
+      FIELD_MESSAGE,    # filled with #{add_message}
+      FIELD_TAGS,       # set with {#tag}
+      FIELD_TIMESTAMP,  # an ISO8601 timestamp of the log event
+      FIELD_VERSION,    # the version of the schema. Currently "1"
+    ].freeze
+
+    # @return [Fields::Hash] the defined fields of the current buffer in a
+    #   hash-like structure
+    attr_reader :fields
+
+    # @return [Fields::Tags] a tags list containing the defined tags for the
+    #   current buffer. It contains frozen strings only.
+    attr_reader :tags
+
+    # @param allow_empty [Boolean] When set to `true` the data in this buffer
+    #   will be flushed to the sink, even if no messages were logged but there
+    #   were just added fields or tags. If this is `false` and there were no
+    #   explicit changes to the buffer (e.g. a logged message, added tags or
+    #   fields), the buffer will not be flushed to the sink but will be silently
+    #   dropped.
+    def initialize(allow_empty: false)
+      @allow_empty = !!allow_empty
+
       @messages = []
+      @fields = Rackstash::Fields::Hash.new(forbidden_keys: FORBIDDEN_FIELDS)
+      @tags = Rackstash::Fields::Tags.new
     end
 
+    # Add a new message to the buffer. This will mark the current buffer as
+    #   {pending?} and will result in the eventual flush of the logged data.
+    #
+    # @param message [Message] A {Message} to add to the current message
+    #   buffer.
+    # @return [Message] the passed `message`
     def add_message(message)
       @messages << message
     end
 
+    def allow_empty?
+      @allow_empty
+    end
+
+    # Return all logged messages on the current buffer.
+    #
+    # @return [Array<Message>] the list of messages of the curent buffer
+    # @note You can not add messsages to the buffer by modifying this array.
+    #   Instead, use {#add_message} to add new messages or add filters to the
+    #   responsible codec to remove or change messages.
     def messages
       @messages.dup
     end
+
+    # This flag denotes whether the current buffer holds flushable data. By
+    # default, a new buffer is not pending and will not be flushed to the sink.
+    # Each time there is a new message logged, this is set to `true` for the
+    # buffer. For changes of tags or fields, the `pending?` flag is only
+    # flipped to `true` if {#allow_empty?} is set to `true`.
+    #
+    # @return [Boolean] `true` if the buffer has stored data which should be
+    #   flushed.
+    def pending?
+      return true if @messages.any?
+      if allow_empty?
+        return true unless @fields.empty?
+        return true unless @tags.empty?
+      end
+      false
+    end
+
+    # Set tags on the buffer. Any values given here are appended to the set of
+    # currently defined tags.
+    #
+    # You can give the tags either as Strings, Arrays of Strings or Procs which
+    # return Strings or Arrays of Strings when called. Each Proc will be called
+    # as it is set on the buffer. If you pass the optional `scope` value, the
+    # Procs will be evaluated in the context of this scope.
+    #
+    # @param tags [Array<#to_s, #call>] Strings to add as tags to the buffer.
+    #   You can either give (arrays of) strings here or procs which return
+    #   a string or an array of strings when called.
+    # @param scope [nil, Object] If anything other then `nil` is given here, we
+    #   will evaluate any procs given in the tags in the context of this
+    #   object. If `nil` is given (the default) the procs are directly called
+    #   in the context where they were created.
+    # @return [Fields::Tags] the resolved tags which are set on the buffer.
+    #   All strings are frozen.
+    def tag(*tags, scope: nil)
+      @tags.merge!(tags, scope: scope)
+    end
   end
 end

spec/rackstash/buffer_spec.rb

@@ -8,7 +8,14 @@ require 'spec_helper'
 require 'rackstash/buffer'
 
 describe Rackstash::Buffer do
-  let(:buffer) { Rackstash::Buffer.new }
+  let(:buffer_options) { {} }
+  let(:buffer) { Rackstash::Buffer.new(**buffer_options) }
+
+  describe '#allow_empty?' do
+    it 'defaults to false' do
+      expect(buffer.allow_empty?).to be false
+    end
+  end
 
   describe '#add_message' do
     it 'adds a message to the buffer' do
@@ -19,7 +26,7 @@ describe Rackstash::Buffer do
     end
   end
 
-  describe 'messages' do
+  describe '#messages' do
     it 'returns an array of messages' do
       msg = double('Rackstash::Message')
       buffer.add_message(msg)
@@ -35,4 +42,102 @@ describe Rackstash::Buffer do
       expect(buffer.messages).to eql []
     end
   end
+
+  describe '#pending?' do
+    it 'sets pending when adding a message' do
+      buffer.add_message double(message: 'some message')
+      expect(buffer.pending?).to be true
+    end
+
+    context 'allow_empty == true' do
+      before do
+        buffer_options[:allow_empty] = true
+        expect(buffer.allow_empty?).to be true
+      end
+
+      it 'defaults to false' do
+        expect(buffer.pending?).to be false
+      end
+
+      it 'is true if there are any fields' do
+        buffer.fields['alice'] = 'bob'
+        expect(buffer.pending?).to be true
+      end
+
+      it 'is true if there are any tags' do
+        buffer.tags << 'alice'
+        expect(buffer.pending?).to be true
+      end
+    end
+
+    context 'allow_empty == false' do
+      before do
+        buffer_options[:allow_empty] = false
+        expect(buffer.allow_empty?).to be false
+      end
+
+      it 'defaults to false' do
+        expect(buffer.pending?).to be false
+      end
+
+      it 'ignores fields' do
+        buffer.fields['alice'] = 'bob'
+        expect(buffer.pending?).to be false
+      end
+
+      it 'ignores tags' do
+        buffer.tags << 'alice'
+        expect(buffer.pending?).to be false
+      end
+    end
+  end
+
+  describe '#tag' do
+    it 'adds tags' do
+      buffer.tag # don't fail with empty argument list
+      buffer.tag 'tag1', 'tag2'
+      expect(buffer.tags).to contain_exactly('tag1', 'tag2')
+    end
+
+    it 'adds tags only once' do
+      buffer.tag 'hello'
+      buffer.tag :hello
+
+      expect(buffer.tags).to contain_exactly('hello')
+    end
+
+    it 'stringifys tags and expands procs' do
+      buffer.tag 123, :symbol, -> { :proc }
+      expect(buffer.tags).to contain_exactly('123', 'symbol', 'proc')
+    end
+
+    it 'does not set blank tags' do
+      buffer.tag 'tag', nil, [], '', {}
+      expect(buffer.tags).to contain_exactly('tag')
+    end
+
+    describe 'when passing procs' do
+      let(:struct) {
+        Struct.new(:value) do
+          def to_s
+            value
+          end
+        end
+      }
+
+      let(:object) {
+        struct.new('Hello')
+      }
+
+      it 'expands single-value proc objects' do
+        buffer.tag(-> { self }, scope: object)
+        expect(buffer.tags).to contain_exactly('Hello')
+      end
+
+      it 'expands multi-value proc objects' do
+        buffer.tag(-> { [[self, 'foobar'], 123] }, scope: object)
+        expect(buffer.tags).to contain_exactly('Hello', 'foobar', '123')
+      end
+    end
+  end
 end