Migrate to use v0.14 API

Author: cosmo0920

lib/fluent/plugin/out_azure-loganalytics.rb

@@ -1,8 +1,14 @@
 # -*- coding: utf-8 -*-
 
-module Fluent
-  class AzureLogAnalyticsOutput < BufferedOutput
-    Plugin.register_output('azure-loganalytics', self)
+require 'fluent/plugin/output'
+
+module Fluent::Plugin
+  class AzureLogAnalyticsOutput < Output
+    Fluent::Plugin.register_output('azure-loganalytics', self)
+
+    helpers :compat_parameters
+
+    DEFAULT_BUFFER_TYPE = "memory"
 
     def initialize
       super
@@ -32,18 +38,24 @@ def initialize
     config_param :tag_field_name, :string, :default => "tag",
                  :desc => "This is required only when add_time_field is true"
 
+    config_section :buffer do
+      config_set_default :@type, DEFAULT_BUFFER_TYPE
+      config_set_default :chunk_keys, ['tag']
+    end
+
     def configure(conf)
+      compat_parameters_convert(conf, :buffer)
       super
-      raise ConfigError, 'no customer_id' if @customer_id.empty?
-      raise ConfigError, 'no shared_key' if @shared_key.empty?
-      raise ConfigError, 'no log_type' if @log_type.empty?
+      raise Fluent::ConfigError, 'no customer_id' if @customer_id.empty?
+      raise Fluent::ConfigError, 'no shared_key' if @shared_key.empty?
+      raise Fluent::ConfigError, 'no log_type' if @log_type.empty?
       if @add_time_field and @time_field_name.empty?
-        raise ConfigError, 'time_field_name must be set if add_time_field is true'
+        raise Fluent::ConfigError, 'time_field_name must be set if add_time_field is true'
       end
       if @add_tag_field and @tag_field_name.empty?
-        raise ConfigError, 'tag_field_name must be set if add_tag_field is true'
+        raise Fluent::ConfigError, 'tag_field_name must be set if add_tag_field is true'
       end
-      @timef = TimeFormatter.new(@time_format, @localtime)
+      @timef = Fluent::TimeFormatter.new(@time_format, @localtime)
     end
 
     def start
@@ -67,6 +79,14 @@ def format(tag, time, record)
       record.to_msgpack
     end
 
+    def formatted_to_msgpack_binary?
+      true
+    end
+
+    def multi_workers_ready?
+      true
+    end
+
     def write(chunk)
       records = []
       chunk.msgpack_each { |record|
@@ -75,11 +95,11 @@ def write(chunk)
       begin
         res = @client.post_data(@log_type, records, @time_generated_field)
         if not Azure::Loganalytics::Datacollectorapi::Client.is_success(res)
-          $log.fatal "DataCollector API request failure: error code: " 
+          $log.fatal "DataCollector API request failure: error code: "
                   + "#{res.code}, data=>" + records.to_json
         end
       rescue Exception => ex
-        $log.fatal "Exception occured in posting to DataCollector API: " 
+        $log.fatal "Exception occured in posting to DataCollector API: "
                   + "'#{ex}', data=>" + records.to_json
       end
     end

test/helper.rb

@@ -23,7 +23,10 @@ def method_missing(method, *args)
   $log = nulllogger
 end
 
+require 'fluent/test/driver/output'
+require 'fluent/test/helpers'
 require 'fluent/plugin/out_azure-loganalytics'
 
 class Test::Unit::TestCase
+  include Fluent::Test::Helpers
 end

test/plugin/test_azure_loganalytics.rb

@@ -16,73 +16,81 @@ def setup
     tag_field_name tag
   ]
 
-  def create_driver(conf = CONFIG, tag='azure-loganalytics.test')
-    Fluent::Test::BufferedOutputTestDriver.new(Fluent::AzureLogAnalyticsOutput, tag).configure(conf)
+  def create_driver(conf = CONFIG)
+    Fluent::Test::Driver::Output.new(Fluent::Plugin::AzureLogAnalyticsOutput).configure(conf)
   end
 
   def test_configure
-    #### set configurations
-    # d = create_driver %[
-    #   path test_path
-    #   compress gz
-    # ]
-    #### check configurations
-    # assert_equal 'test_path', d.instance.path
-    # assert_equal :gz, d.instance.compress
+    d = create_driver
+    assert_equal '<Customer ID aka WorkspaceID String>', d.instance.customer_id
+    assert_equal '<Primary Key String>', d.instance.shared_key
+    assert_equal 'ApacheAccessLog', d.instance.log_type
+    assert_true d.instance.add_time_field
+    assert_true d.instance.localtime
+    assert_true d.instance.add_tag_field
+    assert_equal 'tag', d.instance.tag_field_name
   end
 
   def test_format
-    # d = create_driver
-    # time = Time.parse("2011-01-02 13:14:15 UTC").to_i
-    # d.emit({"a"=>1}, time)
-    # d.emit({"a"=>2}, time)
-
-    # d.expect_format %[2011-01-02T13:14:15Z\ttest\t{"a":1}\n]
-    # d.expect_format %[2011-01-02T13:14:15Z\ttest\t{"a":2}\n]
-
-    # d.run
+    d = create_driver
+    time = event_time("2011-01-02 13:14:15 UTC")
+    d.run(default_tag: 'test') do
+      d.feed(time, {"a"=>1})
+      d.feed(time, {"a"=>2})
+    end
+    formatted = d.formatted
+    unpacker = Fluent::Engine.msgpack_factory.unpacker
+    formatted.each_with_index {|f, idx|
+      unpacker.feed_each(f).each {|e|
+        assert_equal idx+1, e["a"]
+        assert_equal "test", e["tag"]
+        assert_equal "#{time.to_i}", e["time"]
+      }
+    }
   end
 
   def test_write
     d = create_driver
 
-    time = Time.parse("2016-12-10 13:14:15 UTC").to_i
-    d.emit(
-      {
-        :Log_ID => "5cdad72f-c848-4df0-8aaa-ffe033e75d57",
-        :date => "2016-12-10 09:44:32 JST",
-        :processing_time => "372",
-        :remote => "101.202.74.59",
-        :user => "-",
-        :method => "GET / HTTP/1.1",
-        :status => "304",
-        :size => "-",
-        :referer => "-",
-        :agent => "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:27.0) Gecko/20100101 Firefox/27.0",
-        :eventtime => "2016-12-10T09:44:32Z"
-      }, time)
-
-    d.emit(
-      {
-        :Log_ID => "7260iswx-8034-4cc3-uirtx-f068dd4cd659",
-        :date => "2016-12-10 09:45:14 JST",
-        :processing_time => "105",
-        :remote => "201.78.74.59",
-        :user => "-",
-        :method => "GET /manager/html HTTP/1.1",
-        :status =>"200",
-        :size => "-",
-        :referer => "-",
-        :agent => "Mozilla/5.0 (Windows NT 5.1; rv:5.0) Gecko/20100101 Firefox/5.0",
-        :eventtime => "2016-12-10T09:45:14Z"
-      }, time)
+    time = event_time("2016-12-10 13:14:15 UTC")
+    d.run(default_tag: 'azure-loganalytics.test') do
+      d.feed(
+        time,
+        {
+          :Log_ID => "5cdad72f-c848-4df0-8aaa-ffe033e75d57",
+          :date => "2016-12-10 09:44:32 JST",
+          :processing_time => "372",
+          :remote => "101.202.74.59",
+          :user => "-",
+          :method => "GET / HTTP/1.1",
+          :status => "304",
+          :size => "-",
+          :referer => "-",
+          :agent => "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:27.0) Gecko/20100101 Firefox/27.0",
+          :eventtime => "2016-12-10T09:44:32Z"
+        })
 
-    data = d.run
+      d.feed(
+        time,
+        {
+          :Log_ID => "7260iswx-8034-4cc3-uirtx-f068dd4cd659",
+          :date => "2016-12-10 09:45:14 JST",
+          :processing_time => "105",
+          :remote => "201.78.74.59",
+          :user => "-",
+          :method => "GET /manager/html HTTP/1.1",
+          :status =>"200",
+          :size => "-",
+          :referer => "-",
+          :agent => "Mozilla/5.0 (Windows NT 5.1; rv:5.0) Gecko/20100101 Firefox/5.0",
+          :eventtime => "2016-12-10T09:45:14Z"
+        })
+    end
+    data = d.events
     puts data
     # ### FileOutput#write returns path
     # path = d.run
     # expect_path = "#{TMP_DIR}/out_file_test._0.log.gz"
     # assert_equal expect_path, path
   end
 end
-