org.jruby.exceptions.JumpException$BreakJump in /logstash/inputs/ganglia.rb:38
Description
Gliffy Diagrams
Activity
Former user December 3, 2013 at 12:22 PM
That does fix the bug originally reported, but it leads to a plugin that does not work.
The other changes fix the root cause of the error (the reference to the non-existent variable, "source") and the run method that starts a thread and then returns, which causes the input worker to call terminate on the plugin before it has even started!
That gets the plugin to run, but output incorrect information.
The remaining changes fix the broken functionality.
Jordan Sissel December 3, 2013 at 4:57 AM
Should be fixed in next release 
Jordan Sissel December 3, 2013 at 4:44 AM
Jordan Sissel December 3, 2013 at 2:14 AM
The 'break if ...' on line 42 (in v1.2.2) is the problem. Changing this to simply use another means to abort than 'break' (which isn't valid at all in this case)
Former user December 2, 2013 at 11:27 PMEdited
OK. This patch seems to work pretty good:
diff --git a/lib/logstash/inputs/ganglia.rb b/lib/logstash/inputs/ganglia.rb
index f491fc9..f7e9ced 100644
--- a/lib/logstash/inputs/ganglia.rb
+++ b/lib/logstash/inputs/ganglia.rb
@@ -24,7 +24,6 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
public
def initialize(params)
super
- @shutdown_requested = false
BasicSocket.do_not_reverse_lookup = true
end # def initialize
@@ -34,20 +33,8 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
public
def run(output_queue)
- # udp server
- Thread.new do
- begin
- udp_listener(output_queue)
- rescue => e
- break if @shutdown_requested
- @logger.warn("ganglia udp listener died",
- :address => "#{@host}:#{@port}", :exception => e,
- :backtrace => e.backtrace)
- sleep(5)
- retry
- end # begin
- end # Thread.new
-
+ # udp server retries, exceception handling/logging all happens upstairs.
+ udp_listener(output_queue)
end # def run
private
@@ -67,8 +54,9 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
loop do
packet, client = @udp.recvfrom(9000)
# TODO(sissel): make this a codec...
- e = parse_packet(packet,source)
+ e = parse_packet(packet)
unless e.nil?
+ decorate(e)
e["host"] = client[3] # the IP address
output_queue << e
end
@@ -81,7 +69,6 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
public
def teardown
- @shutdown_requested = true
close_udp
finished
end
@@ -96,7 +83,7 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
end
public
- def parse_packet(packet,source)
+ def parse_packet(packet)
gmonpacket=GmonPacket.new(packet)
if gmonpacket.meta?
@@ -116,11 +103,18 @@ class LogStash::Inputs::Ganglia < LogStash::Inputs::Base
event=LogStash::Event.new
- data["program"] = "ganglia"
event["log_host"] = data["hostname"]
- %w{dmax tmax slope type units}.each do |info|
+ # Fields in the data packet itself
+ %w{name spoof format val}.each do |info|
+ event[info] = data[info]
+ end
+ # Fields that are from MetaData
+ %w{dmax tmax slope units}.each do |info|
event[info] = @metadata[data["name"]][info]
end
+ # Change the Ganglia type to dtype, so it can be decorated later.
+ event["dtype"] = @metadata[data["name"]]["type"]
+ # Let it rip!
return event
else
# Skipping unknown packet typesI get this as a JSON event:
{"@timestamp":"2013-12-02T23:25:00.726Z","@version":"1","log_host":"XXXX.com","name":"metricssystem.MetricsSystem.snapshot_num_ops","spoof":0,"format":"%s","val":8.407790785948902e-45,"dmax":0,"tmax":60,"slope":"positive","units":"","dtype":"float","type":"tasktracker-metrics2","host":"127.0.0.1"}Not perfect. The blank units field is troubling.
I am getting:
Exception in thread "RubyThread-5: file:/opt/logstash/agent/lib/logstash-1.2.2-flatjar-backtrace.jar!/logstash/inputs/ganglia.rb:38" org.jruby.exceptions.JumpException$BreakJump
when trying to process the first ganglia packet in version 1.2.X . It used to work fine with version 1.1.13 .
I have filed a bug with the JRUBY folks at https://github.com/jruby/jruby/issues/1268 since I suspect there is a problem in JRUBY.