refactor input specs and shutdown sequence

lib/logstash/inputs/pipe.rb

@@ -26,21 +26,26 @@ class LogStash::Inputs::Pipe < LogStash::Inputs::Base
  # command => "echo hello world"
  config :command, :validate => :string, :required => true
 
+ def initialize(params)
+ super
+ @shutdown_requested = false
+ @pipe = nil
+ end # def initialize
+
  public
  def register
  @logger.info("Registering pipe input", :command => @command)
  end # def register
 
  public
  def run(queue)
- loop do
+ while !@shutdown_requested
  begin
- @pipe = IO.popen(@command, mode="r")
+ @pipe = IO.popen(@command, mode = "r")
  hostname = Socket.gethostname
 
  @pipe.each do |line|
  line = line.chomp
- source = "pipe://#{hostname}/#{@command}"
  @logger.debug? && @logger.debug("Received line", :command => @command, :line => line)
  @codec.decode(line) do |event|
  event["host"] = hostname
@@ -49,6 +54,8 @@ def run(queue)
  queue << event
  end
  end
+ @pipe.close
+ @pipe = nil
  rescue LogStash::ShutdownSignal => e
  break
  rescue Exception => e
@@ -59,4 +66,15 @@ def run(queue)
  sleep(10)
  end
  end # def run
+
+ def teardown
+ @shutdown_requested = true
+ if @pipe
+ Process.kill("KILL", @pipe.pid) rescue nil
+ @pipe.close rescue nil
+ @pipe = nil
+ end
+ finished
+ end
+
 end # class LogStash::Inputs::Pipe

logstash-input-pipe.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
  s.name = 'logstash-input-pipe'
- s.version = '0.1.4'
+ s.version = '0.1.5'
  s.licenses = ['Apache License (2.0)']
  s.summary = "Stream events from a long running command pipe"
  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/pipe_spec.rb

@@ -4,56 +4,52 @@
 
 describe "inputs/pipe", :unix => true do
 
- describe "echo" do
-  event_count = 1
- tmp_file = Tempfile.new('logstash-spec-input-pipe')
-
- config <<-CONFIG
- input {
- pipe {
- command => "echo ☹"
+ # rince and repeat a few times to stress the shutdown sequence
+ 5.times.each do
+ it "should pipe from echo" do
+ conf = <<-CONFIG
+  input {
+  pipe {
+  command => "echo ☹"
+ }
  }
- }
- CONFIG
+ CONFIG
 
- input do |pipeline, queue|
- Thread.new { pipeline.run }
- sleep 0.1 while !pipeline.ready?
-
- events = event_count.times.collect { queue.pop }
- event_count.times do |i|
- insist { events[i]["message"] } == "☹"
+ event = input(conf) do |pipeline, queue|
+ queue.pop
  end
- end # input
- end
 
- describe "tail -f" do
- event_count = 10
-  tmp_file = Tempfile.new('logstash-spec-input-pipe')
+  insist { event["message"] } == "☹"
+ end
+ end
 
- config <<-CONFIG
- input {
- pipe {
- command => "tail -f #{tmp_file.path}"
+ # rince and repeat a few times to stress the shutdown sequence
+ 5.times.each do
+ it "should pipe from tail -f" do
+ event_count = 10
+ tmp_file = Tempfile.new('logstash-spec-input-pipe')
+
+ conf = <<-CONFIG
+ input {
+ pipe {
+ command => "tail -n +0 -f #{tmp_file.path}"
+ }
  }
- }
- CONFIG
-
- input do |pipeline, queue|
- Thread.new { pipeline.run }
- sleep 0.1 while !pipeline.ready?
-
- File.open(tmp_file, "a") do |fd|
- event_count.times do |i|
- # unicode smiley for testing unicode support!
- fd.puts("#{i} ☹")
+ CONFIG
+
+ events = input(conf) do |pipeline, queue|
+ File.open(tmp_file, "a") do |fd|
+ event_count.times do |i|
+ # unicode smiley for testing unicode support!
+ fd.puts("#{i} ☹")
+ end
  end
+ event_count.times.map { queue.pop }
  end
- events = event_count.times.collect { queue.pop }
+
  event_count.times do |i|
  insist { events[i]["message"] } == "#{i} ☹"
  end
- end # input
+ end
  end
-
 end