Updated kusto_spec.rb and ingestor_spec.rb

lib/logstash/outputs/kusto.rb

@@ -68,11 +68,11 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
   # Check Proxy URL can be over http or https. Dowe need it this way or ignore this & remove this
   config :proxy_protocol, validate: :string, required: false , default: 'http'
 
-  # Maximum size of the buffer before it gets flushed, defaults to 2000
-  config :max_size, validate: :number, default: 2000
+  # Maximum size of the buffer before it gets flushed, defaults to 1000
+  config :max_size, validate: :number, default: 1000
 
-  # Maximum interval (in seconds) before the buffer gets flushed, defaults to 5
-  config :max_interval, validate: :number, default: 5
+  # Maximum interval (in seconds) before the buffer gets flushed, defaults to 10
+  config :max_interval, validate: :number, default: 10
 
   default :codec, 'json_lines'
 

spec/outputs/kusto/ingestor_spec.rb

@@ -17,7 +17,6 @@
   let(:proxy_port) { 80 }
   let(:proxy_protocol) { "http" }
   let(:json_mapping) { "mymapping" }
-  let(:delete_local) { false }
   let(:logger) { spy('logger') }
 
   describe '#initialize' do
@@ -26,7 +25,7 @@
       # note that this will cause an internal error since connection is being tried.
       # however we still want to test that all the java stuff is working as expected
       expect { 
-        ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol, logger)
+        ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, proxy_host, proxy_port,proxy_protocol, logger)
         ingestor.stop
       }.not_to raise_error
     end
@@ -37,7 +36,7 @@
       dynamic_name_array.each do |test_database|
         it "with database: #{test_database}" do
           expect {
-            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, test_database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
+            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, test_database, table, json_mapping, proxy_host, proxy_port,proxy_protocol,logger)
             ingestor.stop
           }.to raise_error(LogStash::ConfigurationError)          
         end
@@ -48,7 +47,7 @@
       dynamic_name_array.each do |test_table|
         it "with database: #{test_table}" do
           expect {
-            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, test_table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
+            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, test_table, json_mapping, proxy_host, proxy_port,proxy_protocol,logger)
             ingestor.stop
           }.to raise_error(LogStash::ConfigurationError)          
         end
@@ -59,7 +58,7 @@
       dynamic_name_array.each do |json_mapping|
         it "with database: #{json_mapping}" do
           expect {
-            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
+            ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, proxy_host, proxy_port,proxy_protocol,logger)
             ingestor.stop
           }.to raise_error(LogStash::ConfigurationError)          
         end
@@ -69,7 +68,7 @@
     context 'proxy protocol has to be http or https' do
       it "with proxy protocol: socks" do
         expect {
-          ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, delete_local, proxy_host, proxy_port,'socks',logger)
+          ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, cliauth, database, table, json_mapping, proxy_host, proxy_port,'socks',logger)
           ingestor.stop
         }.to raise_error(LogStash::ConfigurationError)          
       end
@@ -78,7 +77,7 @@
     context 'one of appid or managedid has to be provided' do
       it "with empty managed identity and appid" do
         expect {
-          ingestor = described_class.new(ingest_url, "", app_key, app_tenant, "", cliauth, database, table, json_mapping, delete_local, proxy_host, proxy_port,'socks',logger)
+          ingestor = described_class.new(ingest_url, "", app_key, app_tenant, "", cliauth, database, table, json_mapping, proxy_host, proxy_port,'socks',logger)
           ingestor.stop
         }.to raise_error(LogStash::ConfigurationError)          
       end

spec/outputs/kusto_spec.rb

@@ -15,42 +15,61 @@
     "json_mapping" => "mymapping",
     "proxy_host" => "localhost",
     "proxy_port" => 3128,
-    "proxy_protocol" => "https"
+    "proxy_protocol" => "https",
+    "max_size" => 2000,
+    "max_interval" => 10
   } }
-
+  
   describe '#register' do
-
-    it 'doesnt allow the path to start with a dynamic string' do
-      kusto = described_class.new(options.merge( {'path' => '/%{name}'} ))
-      expect { kusto.register }.to raise_error(LogStash::ConfigurationError)
+    it 'allows valid configuration' do
+      kusto = described_class.new(options)
+      expect { kusto.register }.not_to raise_error
       kusto.close
     end
+  end
+
+  describe '#multi_receive_encoded' do
+    it 'buffers events and flushes based on max_size' do
+      kusto = described_class.new(options.merge( {'max_size' => 2} ))
+      kusto.register
+
+      event1 = LogStash::Event.new("message" => "event1")
+      event2 = LogStash::Event.new("message" => "event2")
+      event3 = LogStash::Event.new("message" => "event3")
+
+      expect(kusto.instance_variable_get(:@buffer)).to receive(:flush).twice.and_call_original
+
+      kusto.multi_receive_encoded([[event1, event1.to_json], [event2, event2.to_json]])
+      kusto.multi_receive_encoded([[event3, event3.to_json]])
 
-    it 'path must include a dynamic string to allow file rotation' do
-      kusto = described_class.new(options.merge( {'path' => '/{name}'} ))
-      expect { kusto.register }.to raise_error(LogStash::ConfigurationError)
       kusto.close
     end
 
+    it 'flushes events based on max_interval' do
+      kusto = described_class.new(options.merge( {'max_interval' => 1} ))
+      kusto.register
 
-    dynamic_name_array = ['/a%{name}/', '/a %{name}/', '/a- %{name}/', '/a- %{name}']
+      event1 = LogStash::Event.new("message" => "event1")
 
-    context 'doesnt allow the root directory to have some dynamic part' do
-      dynamic_name_array.each do |test_path|
-         it "with path: #{test_path}" do
-           kusto = described_class.new(options.merge( {'path' => test_path} ))
-           expect { kusto.register }.to raise_error(LogStash::ConfigurationError)
-           kusto.close
-         end
-       end
-    end
+      expect(kusto.instance_variable_get(:@buffer)).to receive(:flush).at_least(:once).and_call_original
+
+      kusto.multi_receive_encoded([[event1, event1.to_json]])
+
+      sleep 2
 
-    it 'allow to have dynamic part after the file root' do
-      kusto = described_class.new(options.merge({'path' => '/tmp/%{name}'}))
-      expect { kusto.register }.not_to raise_error
       kusto.close
     end
-
   end
 
-end
+  describe '#close' do
+    it 'shuts down the buffer and ingestor' do
+      kusto = described_class.new(options)
+      kusto.register
+
+      expect(kusto.instance_variable_get(:@buffer)).to receive(:shutdown)
+      expect(kusto.instance_variable_get(:@ingestor)).to receive(:stop)
+
+      kusto.close
+    end
+  end
+end