Merge pull request #75 from guyboertje/fix/73

Fix for #73

CHANGELOG.md

@@ -1,3 +1,6 @@
+# 2.2.8
+  - Fix #73 Bug in EventTransformCommon#codec_name, use config_name
+  - add regression test for fix to 73
 # 2.2.7
   - More robust test when using a random port #60
   - Fix LSF integration tests #52

CONTRIBUTORS

@@ -9,6 +9,7 @@ Contributors:
 * Pier-Hugues Pellerin (ph)
 * Richard Pijnenburg (electrical)
 * Suyog Rao (suyograo)
+* Guy Boertje (guyboertje)
 
 Note: If you've sent us patches, bug reports, or otherwise contributed to
 Logstash, and you aren't on the list above and want to be, please let us know

lib/logstash/inputs/beats_support/event_transform_common.rb

@@ -33,7 +33,7 @@ def decorate(event)
 
     def codec_name
       @codec_name ||= if @input.codec.respond_to?(:base_codec)
-                        @input.codec.base_codec.class.config
+                        @input.codec.base_codec.class.config_name
                       else
                         @input.codec.class.config_name
                       end

logstash-input-beats.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = "logstash-input-beats"
-  s.version         = '2.2.7'
+  s.version         = '2.2.8'
   s.licenses        = ["Apache License (2.0)"]
   s.summary         = "Receive events using the lumberjack protocol."
   s.description     = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"

spec/inputs/beats_spec.rb

@@ -10,8 +10,8 @@
 
 describe LogStash::Inputs::Beats do
   let(:connection) { double("connection") }
-  let(:certificate) { LogStashTest.certificate }
-  let(:port) { LogStashTest.random_port }
+  let(:certificate) { BeatsInputTest.certificate }
+  let(:port) { BeatsInputTest.random_port }
   let(:queue)  { Queue.new }
   let(:config)   { { "port" => 0, "ssl_certificate" => certificate.ssl_cert, "ssl_key" => certificate.ssl_key, "type" => "example", "tags" => "beats"} }
 
@@ -97,8 +97,8 @@
   context "#handle_new_connection" do
     let(:config) {{ "ssl" => false, "port" => 0, "type" => "example", "tags" => "beats" }}
     let(:plugin) { LogStash::Inputs::Beats.new(config) }
-    let(:connection) { DummyConnection.new(events) }
-    let(:buffer_queue) { DummyNeverBlockedQueue.new }
+    let(:connection) { BeatsInputTest::DummyConnection.new(events) }
+    let(:buffer_queue) { BeatsInputTest::DummyNeverBlockedQueue.new }
     let(:pipeline_queue)  { [] }
     let(:events) {
       [

spec/inputs/beats_support/connection_handler_spec.rb

@@ -11,20 +11,20 @@
       "tags" => "beats"
     }
   end
-
+  # logger is not used and DummyLogger needs implementing
   let(:logger) { DummyLogger.new  }
   let(:input) do
     LogStash::Inputs::Beats.new(config).tap do |i|
       i.register
     end
   end
   let(:connection) { double("connection") }
-  let(:queue) { DummyNeverBlockedQueue.new }
+  let(:queue) { BeatsInputTest::DummyNeverBlockedQueue.new }
 
   subject { described_class.new(connection, input, queue) }
-  
+
   context "#accept" do
-    let(:connection) { DummyConnection.new(events) }
+    let(:connection) { BeatsInputTest::DummyConnection.new(events) }
     let(:events) {
       [
         { :map => { "id" => 1 }, :identity_stream => "/var/log/message" },
@@ -59,7 +59,7 @@
       context "queue is blocked" do
         let(:queue_timeout) { 1 }
         let(:queue) { LogStash::Inputs::BeatsSupport::SynchronousQueueWithOffer.new(queue_timeout) }
-      
+
         it "raise an exception" do
           expect { subject.process(map, identity_stream) }.to raise_error(LogStash::Inputs::Beats::InsertingToQueueTakeTooLong)
         end

spec/inputs/beats_support/decoded_event_transform_spec.rb

@@ -64,4 +64,11 @@
       expect(subject["@timestamp"]).to be_kind_of(LogStash::Timestamp)
     end
   end
+
+  context "when the codec is a base_codec wrapper" do
+    before { config.update("codec" => BeatsInputTest::DummyCodec.new) }
+    it "gets the codec config name from the base codec" do
+      expect(subject["tags"]).to include("beats_input_codec_dummy_applied")
+    end
+  end
 end

spec/support/logstash_test.rb

@@ -1,5 +1,9 @@
 require "stud/temporary"
-module LogStashTest
+
+
+# namespace the Dummy* classes, they are reused names
+# use a more specific module name to prevent clashes
+module BeatsInputTest
   class Certicate
     attr_reader :ssl_key, :ssl_cert
 
@@ -20,29 +24,43 @@ def random_port
       rand(2000..10000)
     end
   end
-end
 
-class DummyNeverBlockedQueue < Array
-  def offer(element, timeout = nil)
-    push(element)
+  class DummyNeverBlockedQueue < Array
+    def offer(element, timeout = nil)
+      push(element)
+    end
+
+    alias_method :take, :shift
   end
 
-  alias_method :take, :shift
-end
+  class DummyConnection
+    def initialize(events)
+      @events = events
+    end
 
-class DummyConnection
-  def initialize(events)
-    @events = events
-  end
+    def run
+      @events.each do |element|
+        yield element[:map], element[:identity_stream]
+      end
+    end
 
-  def run
-    @events.each do |element|
-      yield element[:map], element[:identity_stream] 
+    def peer
+      "localhost:5555"
     end
   end
 
-  def peer
-    "localhost:5555"
+  class DummyCodec
+    def register() end
+    def decode(*) end
+    def clone() self; end
+    def base_codec
+      self
+    end
+    def self.config_name
+      "dummy"
+    end
   end
 end
 
+
+