Add encoding selector option (base64 or binary)

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.4.0
+  - Add `base64_encoding` option to disable the base64 encoding of Avro payload [#39](https://github.com/logstash-plugins/logstash-codec-avro/pull/39)
+
 ## 3.3.1
   - Pin avro gem to 1.10.x, as 1.11+ requires ruby 2.6+ [#37](https://github.com/logstash-plugins/logstash-codec-avro/pull/37)
 

docs/index.asciidoc

@@ -81,6 +81,7 @@ output {
 [cols="<,<,<",options="header",]
 |=======================================================================
 |Setting |Input type|Required
+| <<plugins-{type}s-{plugin}-base64_encoding>> | <<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
 | <<plugins-{type}s-{plugin}-schema_uri>> |<<string,string>>|Yes
 | <<plugins-{type}s-{plugin}-tag_on_failure>> |<<boolean,boolean>>|No
@@ -99,6 +100,16 @@ output {
 
 Controls this plugin's compatibility with the {ecs-ref}[Elastic Common Schema (ECS)].
 
+[id="plugins-{type}s-{plugin}-base64_encoding"]
+===== `base64_encoding`
+
+* Value type is <<boolean,boolean>>
+* Default value is `true`
+
+Select whether use or not the base64 encoding.
+When it's `true` and plain message arrives then it's however decoded.
+Raises an error if it's `false` and base64 encodec message arrives.
+
 
 [id="plugins-{type}s-{plugin}-schema_uri"]
 ===== `schema_uri` 

lib/logstash/codecs/avro.rb

@@ -59,6 +59,9 @@ class LogStash::Codecs::Avro < LogStash::Codecs::Base
 
   include LogStash::PluginMixins::EventSupport::EventFactoryAdapter
 
+  # make base64 encoding optional
+  config :base64_encoding, :validate => :boolean, :default => true
+
   # schema path to fetch the schema from.
   # This can be a 'http' or 'file' scheme URI
   # example:
@@ -92,7 +95,11 @@ def register
 
   public
   def decode(data)
-    datum = StringIO.new(Base64.strict_decode64(data)) rescue StringIO.new(data)
+    if base64_encoding
+      datum = StringIO.new(Base64.strict_decode64(data)) rescue StringIO.new(data)
+    else
+      datum = StringIO.new(data)
+    end
     decoder = Avro::IO::BinaryDecoder.new(datum)
     datum_reader = Avro::IO::DatumReader.new(@schema)
     event = targeted_event_factory.new_event(datum_reader.read(decoder))
@@ -113,6 +120,10 @@ def encode(event)
     buffer = StringIO.new
     encoder = Avro::IO::BinaryEncoder.new(buffer)
     dw.write(event.to_hash, encoder)
-    @on_event.call(event, Base64.strict_encode64(buffer.string))
+    if base64_encoding
+      @on_event.call(event, Base64.strict_encode64(buffer.string))
+    else
+      @on_event.call(event, buffer.string)
+    end  
   end
 end

logstash-codec-avro.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-codec-avro'
-  s.version         = '3.3.1'
+  s.version         = '3.4.0'
   s.platform        = 'java'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Reads serialized Avro records as Logstash events"

spec/codecs/avro_spec.rb

@@ -55,6 +55,35 @@
         end
       end
 
+      context "base64_encoding is false" do
+        let (:avro_config) { super().merge('base64_encoding' => false) }
+
+        it "should return an LogStash::Event from raw and base64 encoded avro data" do
+          schema = Avro::Schema.parse(avro_config['schema_uri'])
+          dw = Avro::IO::DatumWriter.new(schema)
+          buffer = StringIO.new
+          encoder = Avro::IO::BinaryEncoder.new(buffer)
+          dw.write(test_event.to_hash, encoder)
+
+          subject.decode(buffer.string) do |event|
+            expect(event).to be_a_kind_of(LogStash::Event)
+            expect(event.get("foo")).to eq(test_event.get("foo"))
+            expect(event.get("bar")).to eq(test_event.get("bar"))
+            expect(event.get('[event][original]')).to eq(buffer.string) if ecs_compatibility != :disabled
+          end
+        end
+
+        it "should raise an error if base64 encoded data is provided" do
+          schema = Avro::Schema.parse(avro_config['schema_uri'])
+          dw = Avro::IO::DatumWriter.new(schema)
+          buffer = StringIO.new
+          encoder = Avro::IO::BinaryEncoder.new(buffer)
+          dw.write(test_event.to_hash, encoder)
+
+          expect {subject.decode(Base64.strict_encode64(buffer.string))}.to raise_error
+        end
+      end
+
       context "#decode with tag_on_failure" do
         let (:avro_config) {{ 'schema_uri' => '
                         {"type": "record", "name": "Test",
@@ -111,6 +140,28 @@
           insist {got_event}
         end
 
+        context "base64_encoding is false" do
+          let (:avro_config) { super().merge('base64_encoding' => false) }
+
+          it "should return avro data from a LogStash::Event not base64 encoded" do
+            got_event = false
+            subject.on_event do |event, data|
+              schema = Avro::Schema.parse(avro_config['schema_uri'])
+              datum = StringIO.new(data)
+              decoder = Avro::IO::BinaryDecoder.new(datum)
+              datum_reader = Avro::IO::DatumReader.new(schema)
+              record = datum_reader.read(decoder)
+
+              expect(event).to be_a_kind_of(LogStash::Event)
+              expect(event.get("foo")).to eq(test_event.get("foo"))
+              expect(event.get("bar")).to eq(test_event.get("bar"))
+              got_event = true
+            end
+            subject.encode(test_event)
+            expect(got_event).to be true
+          end
+        end
+
         context "binary data" do
 
           let (:avro_config) {{ 'schema_uri' => '{"namespace": "com.systems.test.data",