PR for Refactor shutdown sequence to use plugin.stop

[guyboertje]

add it_behaves_like "an interruptible input plugin" shared example
refactor plugin slightly according to elastic/logstash#3210

Depends on elastic/logstash-devutils#32 and elastic/logstash#3812

Fixes #11

[purbon]

This would be required to be tracked too, imagine what would happen if you aim to stop when are into this sleep.

[guyboertje]

@purbon - now OK?

[purbon]

Why doing this? and not simply tracking the state and waking up the thread? just curious to know your way of thinking 👍 might be we should apply this pattern in other places 😄

[guyboertje]

The pipeline thread is calling stop on the input instance, so AFAIK the current thread is not the input thread (which might be sleeping). Is it true to say that in the pipeline code the input_thread and the input instance do not know about each other? If so then, in the pipeline thread, when calling the stop method, the instance would need to hold a reference to the input thread to wake itself up OR the pipeline should wakeup the input thread before/after calling stop on the input instance.

See this gist

[purbon]

what we/I did in other plugins is to actually hold an instance of the current thread in the run method and use it to wakeup via an atomic value. I guess both does the same at the end of the day. It would be nice however to uniform somehow the way we apply this pattern either in the core level or with other plugins, what do you think?

[ph]

Yep, been there and I was bite by it too, See:
jordansissel/ruby-stud#27 (comment)

I think we need a common way of doing it or the pipeline should provide it or enforce a strategies.

[guyboertje]

For my part I don't like the way the pipeline holds unrelated references to the thread and the plugin in different arrays. I would consider a tuple value object PluginRunner(?), with the thread and instance as tuple values and this instance being stored in one array.

def start_input
  thread = Thread.new { inputworker(plugin) }
  @input_runners << PluginRunner.new(input, thread)
end

def shutdown
  #...
  @input_runners.each do |runner|
    runner.plugin.stop
    runner.thread.run #noop if thread is running
  end
end

Obviously the names of ivars, variables and classes should be appropriate.
I'm thinking this should work for thread sleep called from Stud as well?

[guyboertje]

However the pipeline thread (code) does not know how the plugin thread came to be asleep or dead or whether the plugin implements the sleep protocol correctly (third party closed source plugins perhaps?), therefore, is it safer for the pipeline to check if the corresponding thread is dead or asleep before stopping the input or waking the corresponding thread? Do you need to stop an input if its thread is dead. Our secondary threads die on unhandled exceptions e.g. OOM.

And we should summarize all the above is in the discussion issue that Colin is creating.

[jsvd]

actually, why is it necessary to go to such lengths here? it will be quite rare to hit stop! during sleep, so we can just double check:

def run
  start_listener
rescue => e
  sleep(5) unless !stop
  retry unless !stop
end

[guyboertje]

Thats kind of what I wrote originally, but the concurrency issues that @colinsurprenant mentioned mean that the retry may occur anyway. Oops, in this PR I am not checking for stopped in the 'start_listener' equivalent

def run
  return if stop?
  start_listener
rescue => e
  sleep(5) unless !stop?
  retry
end

# or perhaps (sorry, I could not resist)
def run
  return if stop?
  sleep(2)
  start_listener
rescue => e
  retry
end

[jsvd]

that's why I explicitly added two checks for !stop, both to sleep and retry

[colinsurprenant]

even if you check before and after, nothing prevent having a stop occur while in the sleep and the shutdown sequence will be tied to the sleep period. only way to prevent this is to simply use a "stoppable_sleep" which check for stop periodically.

other than that, we cannot control the code that is not under our control. we can try our best to force stop foreign code by using whichever api they provide by if it does not cooperate it will be caught by the (eventual) stuck shutdown logic, otherwise we can submit upstream patches on 3rd party libraries that do not correctly stopping or closing etc.

the hope is that such scenarios will be rare. the ones that are obvious are the input plugins that are blocked on IO and that the stop method will need to unblock.

[purbon]

two small comments here:

1.- I run the test in this PR and got this error:

  1) LogStash::Inputs::Snmptrap behaves like an interruptible input plugin #stop returns from run
     Failure/Error: subject { described_class.new(config) }
     LogStash::PluginLoadingError:
       Couldn't find any codec plugin named 'plain'. Are you sure this is correct? Trying to load the plain codec plugin resulted in this error: no such file to load -- logstash/codecs/plain
     Shared Example Group: "an interruptible input plugin" called from ./spec/inputs/snmptrap_spec.rb:5
     # ./lib/logstash/inputs/snmptrap.rb:34:in `initialize'

this is because the codec_plain should be included as development dependency for this plugin.

2.- On the other side, as we merged in recently a PR that has changes in the gemspec, it would be nice, even if not 100% necessary, to rebase PR and get to test with the final gemspec config.

[guyboertje]

I will rebase

[guyboertje]

Rebased and added plain codec dependency. Tests pass.

[purbon]

LGTM, keep in mind to squash the commits 👍 /cheers

[elasticsearch-bot]

Merged sucessfully into master!