Test: ('timeout') behavior with channels

kares · Apr 12, 2021 · bd20251 · bd20251
1 parent fa9f89d
commit bd20251
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 7 deletions.
diff --git a/lib/logstash/inputs/redis.rb b/lib/logstash/inputs/redis.rb
@@ -126,7 +126,6 @@ def redis_params
   end
 
   TIMEOUT = 5 # Redis only supports Integer values
-  private_constant :TIMEOUT
 
   def new_redis_instance
     ::Redis.new(redis_params)

diff --git a/spec/inputs/redis_spec.rb b/spec/inputs/redis_spec.rb
@@ -254,9 +254,9 @@ def process(conf, event_count)
 
     before { subject.register }
 
-    def run_it_thread(inst)
-      Thread.new(inst) do |subj|
-        subj.run(queue)
+    def run_it_thread(plugin)
+      Thread.new(plugin) do |subject|
+        subject.run(queue)
       end
     end
 
@@ -269,16 +269,29 @@ def publish_thread(new_redis, prefix)
       end
     end
 
-    def close_thread(inst, rt)
-      Thread.new(inst, rt) do |subj, runner|
+    def close_thread(plugin, runner_thread)
+      Thread.new(plugin, runner_thread) do |subject, runner|
         # block for the messages
         e1 = queue.pop
         e2 = queue.pop
         # put em back for the tests
         queue.push(e1)
         queue.push(e2)
         runner.raise(LogStash::ShutdownSignal)
-        subj.close
+        subject.close
+      end
+    end
+
+    def stub_plugin_timeout(timeout)
+      value = LogStash::Inputs::Redis::TIMEOUT
+      begin
+        LogStash::Inputs::Redis.send :remove_const, :TIMEOUT
+        LogStash::Inputs::Redis.const_set :TIMEOUT, timeout
+
+        yield
+      ensure
+        LogStash::Inputs::Redis.send :remove_const, :TIMEOUT rescue nil
+        LogStash::Inputs::Redis.const_set :TIMEOUT, value
       end
     end
 
@@ -315,6 +328,23 @@ def close_thread(inst, rt)
 
           expect(queue.size).to eq(2)
         end
+
+        it 'calling the run method, adds events to the queue (after timeout)' do
+          stub_plugin_timeout(0.5) do
+            #simulate the input thread
+            rt = run_it_thread(subject)
+            [ :warn, :error ].each { |level| expect(subject.logger).not_to receive(level) }
+            #make sure the Redis call times out and gets retried
+            sleep(LogStash::Inputs::Redis::TIMEOUT * 4)
+            #simulate the other system thread
+            publish_thread(subject.send(:new_redis_instance), 'c').join
+            #simulate the pipeline thread
+            close_thread(subject, rt).join
+
+            expect(queue.size).to eq(2)
+          end
+        end
+
         it 'events had redis_channel' do
           #simulate the input thread
           rt = run_it_thread(subject)
@@ -354,6 +384,22 @@ def close_thread(inst, rt)
           expect(queue.size).to eq(2)
         end
 
+        it 'calling the run method, adds events to the queue (after timeout)' do
+          stub_plugin_timeout(0.5) do
+            #simulate the input thread
+            rt = run_it_thread(subject)
+            [ :warn, :error ].each { |level| expect(subject.logger).not_to receive(level) }
+            #make sure the Redis call times out and gets retried
+            sleep(LogStash::Inputs::Redis::TIMEOUT * 4)
+            #simulate the other system thread
+            publish_thread(subject.send(:new_redis_instance), 'c').join
+            #simulate the pipeline thread
+            close_thread(subject, rt).join
+
+            expect(queue.size).to eq(2)
+          end
+        end
+
         it 'events had redis_channel' do
           #simulate the input thread
           rt = run_it_thread(subject)