Thrift Monitoring for storm topologies. gen-rb folder has thrift generated ruby wrappers for talking to storm cluster

gistfile1.txt

$:.push('/home/yranadive/gen-rb')

# This file monitors a set of storm topologies and generates
# 1. Zenoss friendly status message. e.g. topo1:1 topo2:0 (1=active, 0=inactive)
# 2. Mails the inactive topologies
# Usage: ruby storm_monitor.rb [mail]
# Optional argument "mail" on the command line will mail the report
require 'rubygems'
require 'thrift'
require 'nimbus'
require 'mail'

def mail_report(params, report_html)
  puts params[:subject]
  mail = Mail.new do
    from         params[:from] #'[email protected]'
    to           params[:to] #'[email protected]'
    subject      params[:subject] #'This is a test html email'
    content_type params[:content_type] || 'text/html; charset=UTF-8'
    body         report_html
  end

  mail.delivery_method :sendmail
  mail.deliver
end

begin
  # list of topologies to monitor
  topology_list = ['device_events']
  socket    = Thrift::Socket.new('dw-utility1', 6627)
  transport = Thrift::FramedTransport.new(socket)
  protocol  = Thrift::BinaryProtocol.new(transport)
  client    = Nimbus::Client.new(protocol)
  params = {
    :from         => '[email protected]',
    :to           => '[email protected]',
  }
  transport.open

  summary = client.getClusterInfo
  registered_topologies = {}
  summary.topologies.each {|x| registered_topologies[x.name] = x.status}
  if(ARGV[0] == 'mail')
    topology_list.each do |top|
      params[:subject] = "Topology #{top} has stopped"
      mail_report(params, "<eom>") if(registered_topologies[top] != 'ACTIVE')
    end
  else
    composite = ""
    topology_list.each do |top|
      composite += "#{top}:#{registered_topologies[top] == 'ACTIVE' ? '1' : '0'} "
    end
    puts composite.strip
#    puts topology_list.inject {|zenoss_output, top| zenoss_output + "#{top}:#{registered_topologies[top] == 'ACTIVE' ? '1' : '0'} "}
  end
  transport.close
rescue Thrift::Exception => tx
  print 'Thrift::Exception: ', tx.message, "\n"
end

To get info for a particular topology use:
client.getTopologyInfo(id)
where id = client.getClusterInfo.each {|x| puts x.name + " " + x.id}