A script for a Logstash Ruby Filter to transpose an array of two-element objects representing key/value tuples into a single hash/map

pipeline.conf

filter {
  # to convert an array of key/value objects into a single unordered
  # key/value map, use the included `transpose` script:
  ruby {
    path => "${PWD}/transpose.logstash-filter-ruby.rb"
    script_params => {
      source => "[proplist]"
    }
  }

  # to convert a single unordered key/value map into an array of key/value
  # objects, use the included `untranspose` script:
  ruby {
    path => "${PWD}/untranspose.logstash-filter-ruby.rb"
    script_params => {
      source => "[proplist]"
    }
  }
}

strip-field-names-in-map.logstash-filter-ruby.rb

###############################################################################
# strip-field-names-in-map.logstash-filter-ruby.rb
# ---------------------------------
# A script for a Ruby filter to strip characters from the field names in a
# key/value map; by default, it strips leading and trailing whitespace, but it
# can be configured with one or more regexp patterns.
###############################################################################
#
# Copyright 2018 Ry Biesemeyer
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
def register(params)
  params = params.dup

    # source: a Field Reference to the key/value map (required)
  @source = params.delete('source') || report_configuration_error('missing required script parameter `source`')

  # target: a Field Reference indicating where to place the untransposed result
  #         (optional; default _replaces_ `source` with untransposed result)
  @target = params.delete('target') || @source

  # pattern: one or more regexp patterns; matching substrings in field names of
  #.         the source object will be stripped. If left unspecified, will strip
  #.         both leading and trailing whitespace.
  @pattern = patterns_for(params.delete('pattern') || [/(?:\A\s+)/, /(?:\s+\Z)/])

  $stderr.puts(@pattern.inspect)

  params.empty? || report_configuration_error("unknown script parameter(s): #{params.keys}")
end

def patterns_for(str_or_array)
  patterns = Array(str_or_array).map { |t| Regexp.compile(t) }
  Regexp.union(patterns)
end

def filter(event)
  return [event] unless event.include?(@source)

  source_map = event.get(@source)

  fail('source not a key/value map') unless source_map.kind_of?(Hash)

  map_with_stripped_field_names = source_map.each_with_object({}) do |(key, value), memo|
    memo[key.gsub(@pattern,'')] = value
  end

  event.set(@target, map_with_stripped_field_names)

rescue => e
  logger.error('failed to strip whitespace from field names in map', exception: e.message)
  event.tag('_stripfailure')
ensure
  return [event]
end

def report_configuration_error(message)
  raise LogStash::ConfigurationError, message
end

test "defaults" do
  parameters do
    { "source" => "[map]" }
  end

  in_event do
    { "map" => {
        "     leading" => "leading",
        "trailing    " => "trailing",
        "  both      " => "both",
        "middle space" => "middle space",
      }
    }
  end

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('result should have the right number of items') do |events|
    events.first.get('[map]').size == 4
  end

  expect('key with leading space should have been stripped correctly') do |events|
    events.first.get("map").has_key?("leading")
  end
  expect('key with trailing space should have been stripped correctly') do |events|
    events.first.get("map").has_key?("trailing")
  end
  expect('key with both space should have been stripped correctly') do |events|
    events.first.get("map").has_key?("both")
  end
  expect('key with middle space should have been stripped correctly') do |events|
    events.first.get("map").has_key?("middle space")
  end
end

transpose.logstash-filter-ruby.rb

###############################################################################
# transpose.logstash-filter-ruby.rb
# ---------------------------------
# A script for a Ruby filter to transpose an array of objects into a single
# unordered key/value map
###############################################################################
#
# Copyright 2018 Ry Biesemeyer
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
def register(params)
  params = params.dup # isolate parent from mutation

  # source: a Field Reference to the array (required)
  @source = params.delete('source') || report_configuration_error('missing required script parameter `source`')

  # target: a Field Reference indicating where to place the transposed result
  #         (optional; default _replaces_ `source` with transposed result)
  @target = params.delete('target') || @source

  # field_name_key: a _relative_ Field Reference to the field within each
  #                 element of the array that holds the name of the element
  @field_name_key = params.delete('field_name_key')   || 'name'

  # field_value_key: a _relative_ Field Reference to the field within each
  #                  element of the array that holds the value of the element
  @field_value_key = params.delete('field_value_key') || 'value'

  params.empty? || report_configuration_error("unknown script parameter(s): #{params.keys}")
end

def filter(event)
  return [event] unless event.include?(@source)
  source_array = event.get(@source)

  fail('source not an array') unless source_array.kind_of?(Array)

  transposed = source_array.size.times.each_with_object({}) do |index, memo|
    key_field_reference = "[#{@source}][#{index}][#{@field_name_key}]"
    value_field_reference = "[#{@source}][#{index}][#{@field_value_key}]"

    fail("source malformed; field name key not present at #{index}") unless event.include?(key_field_reference)
    fail("source malformed; field value key not present at #{index}") unless event.include?(value_field_reference)

    key = event.get(key_field_reference)
    value = event.get(value_field_reference)

    memo[key] = value
  end

  event.set(@target, transposed)
rescue => e
  logger.error('failed to transpose array to hash', exception: e.message)
  event.tag('_transposefailure')
ensure
  return [event]
end

def report_configuration_error(message)
  raise LogStash::ConfigurationError, message
end

test "defaults" do
  parameters do
    { "source" => "[foo][bar]" }
  end

  in_event { { "foo" => { "bar" => [{"name"=>"baz","value"=>"bingo"},{"name"=>"another","value"=>"value"}] } } }

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('values have been transposed in place') do |events|
    events.first.get('[foo][bar]') == {'baz' => 'bingo', 'another' => 'value'}
  end
end

test 'alternate target' do
  parameters do
    {
      "source" => "[foo][bar]",
      "target" => "[foo][transposed]"
    }
  end

  in_event { { "foo" => { "bar" => [{"name"=>"baz","value"=>"bingo"},{"name"=>"another","value"=>"value"}] } } }

  expect('produces single event') do |events|
    events.size == 1
  end

  expect('values have been transposed to target') do |events|
    events.first.get('[foo][transposed]') == {'baz' => 'bingo', 'another' => 'value'}
  end

  expect('source left alone') do |events|
    events.first.get('[foo][bar]').kind_of?(Array)
  end
end

test 'malformed' do
  parameters do
    { "source" => "[foo][bar]" }
  end

  in_event { { "foo" => { "bar" => "string" } } }

  expect('tag with failure tag') do |events|
    events.first.get('tags').include?('_transposefailure')
  end
end

test 'alternate k/v names' do
  parameters do
    {
      "source" => "[foo]",
      "field_name_key" => "Property",
      "field_value_key" => "Value"
    }
  end

  in_event { {"foo" => [{"Property" => "currency", "Value" => "USD"}] } }

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('values have been transposed in place') do |events|
    events.first.get('[foo]') == {'currency' => 'USD'}
  end
end

test 'metadata and nesting' do
  parameters do
    {
      "source" => "[foo]",
      "field_name_key" => "[Name]",
      "field_value_key" => "[Value][VALUE]"
    }
  end

  in_event do
    {
      "foo" => [
         { "Name" => "bar", "Meta" => "irrelevant", "Value" => { "IS_NULL" => false, "TYPE" => "System.String", "VALUE" => "1.0" }},
         { "Name" => "baz", "Meta" => "irrelevant", "Value" => { "IS_NULL" => false, "TYPE" => "System.String", "VALUE" => 1 }},
         { "Name" => "bingo", "Meta" => "irrelevant", "Value" => { "IS_NULL" => false, "TYPE" => "System.String", "VALUE" => false }}
      ]
    }
  end

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('values have been transposed in place') do |events|
    events.first.get('[foo]') == {'bar' => "1.0", 'baz' => 1, 'bingo' => false}
  end
end

untranspose.logstash-filter-ruby.rb

###############################################################################
# untranspose.logstash-filter-ruby.rb
# ---------------------------------
# A script for a Ruby filter to transpose a key/value map into an unordered
#.array of objects.
###############################################################################
#
# Copyright 2018 Ry Biesemeyer
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
def register(params)
  params = params.dup # isolate parent from mutation

  # source: a Field Reference to the key/value map (required)
  @source = params.delete('source') || report_configuration_error('missing required script parameter `source`')

  # target: a Field Reference indicating where to place the untransposed result
  #         (optional; default _replaces_ `source` with untransposed result)
  @target = params.delete('target') || @source

  # field_name_key: the desired name of the field in the resulting object
  #.                that should hold the key from the key/value map
  @field_name_key = params.delete('field_name_key')   || 'name'

  # field_value_key: the desired name of the field in the resulting object
  #.                 that should hold the value from the key/value map
  @field_value_key = params.delete('field_value_key') || 'value'

  params.empty? || report_configuration_error("unknown script parameter(s): #{params.keys}")

  # nested field references aren't supported due to unspecified behaviour in the Logstash Event API
  nested_field_reference?(@field_name_key) && report_configuration_error('given `field_name_key` cannot represent a nested field_reference')
  nested_field_reference?(@field_value_key) && report_configuration_error('given `field_value_key` cannot represent a nested field_reference')
end

def filter(event)
  return [event] unless event.include?(@source)

  source_map = event.get(@source)

  fail('source not a key/value map') unless source_map.kind_of?(Hash)

  untransposed = source_map.map.with_index do |(key, value), index|
    fail("source malformed; field name key not a string at #{index}") unless key.kind_of?(String)

    {
      @field_name_key => key,
      @field_value_key => value
    }
  end

  event.set(@target, untransposed)
rescue => e
  logger.error('failed to untranspose hash to array', exception: e.message)
  event.tag('_untransposefailure')
ensure
  return [event]
end

def report_configuration_error(message)
  raise LogStash::ConfigurationError, message
end

def nested_field_reference?(field_reference)
  field_reference.include?('][')
end

# TODO trans -> untrans
test "defaults" do
  parameters do
    { "source" => "[foo][bar]" }
  end

  in_event { { "foo" => { "bar" => {'baz' => 'bingo', 'another' => 'value'} } } }

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('untransposed result should have the right number of items') do |events|
    events.first.get('[foo][bar]').size == 2
  end

  expect('untransposed result should include relevant values') do |events|
    [{"name"=>"baz","value"=>"bingo"},{"name"=>"another","value"=>"value"}].each do |result|
      events.first.get('[foo][bar]').include?(result)
    end
  end
end

test 'alternate target' do
  parameters do
    {
      "source" => "[foo][bar]",
      "target" => "[foo][untransposed]"
    }
  end

  in_event { { "foo" => { "bar" => {'baz' => 'bingo', 'another' => 'value'} } } }

  expect('produces single event') do |events|
    events.size == 1
  end

  expect('untransposed result should have the right number of items') do |events|
    events.first.get('[foo][untransposed]').size == 2
  end

  expect('untransposed result should include relevant values') do |events|
    [{"name"=>"baz","value"=>"bingo"},{"name"=>"another","value"=>"value"}].each do |result|
      events.first.get('[foo][untransposed]').include?(result)
    end
  end

  expect('source left alone') do |events|
    events.first.get('[foo][bar]').kind_of?(Hash)
  end
end

test 'malformed' do
  parameters do
    { "source" => "[foo][bar]" }
  end

  in_event { { "foo" => { "bar" => "string" } } }

  expect('tag with failure tag') do |events|
    events.first.get('tags').include?('_untransposefailure')
  end
end

test 'alternate k/v names' do
  parameters do
    {
      "source" => "[foo]",
      "field_name_key" => "Property",
      "field_value_key" => "Value"
    }
  end

  in_event { {"foo" => {'currency' => 'USD'} } }

  expect("produces single event") do |events|
    events.size == 1
  end

  expect('values have been transposed in place') do |events|
    events.first.get('[foo]') == [{"Property" => "currency", "Value" => "USD"}]
  end
end