sensu-extensions-logstash.rb

require "sensu/extension"
require 'redis'
require 'json'
require 'socket'
require 'time'

# Author: MOL - November 2016
# Sensu: Tested on 0.26.5
# License: MIT
# This Sensu extension handler sends sensu events to a logstash instance

module Sensu
	module Extension
		class Logstash < Handler
			def name
				"logstash"
			end

			def description
				"sends sensu events to a logstash instance"
			end

			def definition
				{
					name: name,
					type: 'extension',
					mutator: 'ruby_hash',
     					filters: []
				}
			end

			def options
				return @options if @options
				@options = {
					server: "127.0.0.1",
					port: 12347,
					list: "logstash",
					type: "sensu-logstash",
					output: "udp",
					debug: false

				}
				if @settings[:logstash].is_a?(Hash)
					@options.merge!(@settings[:logstash])
				end
				@options
			end

			def post_init
				@logger.warn("LOGSTASH HOST: #{options[:server]}")
				@logger.warn("LOGSTASH  PORT: #{options[:port]}")
				@logger.warn("LOGSTASH  LIST: #{options[:list]}")
				@logger.warn("LOGSTASH  TYPE: #{options[:type]}")
				@logger.warn("LOGSTASH  OUTPUT: #{options[:output]}")
				begin
					if options[:output] == 'udp' 
						socket = UDPSocket.new
						delivery = socket.send('PING', 0, options[:server], options[:port].to_i)
						socket.close
						@logger.warn("LOGSTASH INSTANCE TELNET OK #{delivery}")
					end
				rescue Exception => e
					@logger.warn('LOGSTASH TELNET IMPOSSIBLE')
					e.inspect
				end
			end


			def event_status(event)
				case event[:check][:status]
				when 0
					'OK'
				when 1
					'WARNING'
				when 2
					'CRITICAL'
				else
					'unknown'
				end
			end

			def action_to_string(event)
				check_name = "#{event[:check][:name]}"
				if event[:action].eql?('resolve') and !check_name.include?('metric')
					return 'RESOLVE'
				else
					if (event_status(event).eql?('WARNING') or event_status(event).eql?('CRITICAL')) and !check_name.include?('metric')
						return 'ALERT'
					elsif event_status(event).eql?('OK') and !check_name.include?('metric')
						return 'OK'
					elsif event_status(event).eql?('unknown')
						return 'UNKNOWN'
					else
						return 'METRIC'
					end
				end
				#event[:action].eql?('resolve') ? 'RESOLVE' : 'ALERT'
			end

			def forwarder(event)
				begin
					@logger.warn("CALLING LOGSTASH EXTENSION") if options[:debug]
					time = Time.now.utc.iso8601
					logstash_msg = {
						:@timestamp    => time,
						:@version      => 1,
						:source        => ::Socket.gethostname,
						:tags          => ["sensu-#{action_to_string(event)}"],
						:message       => event[:check][:output],
							:host          => event[:client][:name],
							:timestamp     => event[:check][:issued],
							:address       => event[:client][:address],
							:check_name    => event[:check][:name],
							:command       => event[:check][:command],
							:status        => event_status(event),
							:flapping      => event[:check][:flapping],
							:occurrences   => event[:occurrences],
							:action        => event[:action]
					}
					logstash_msg[:type] = options[:type] if options[:type]

					@logger.warn("LOGSTASH OUTPUT: #{logstash_msg.to_json}") if options[:debug]
					case options[:output]
					when 'redis'
						redis = Redis.new(host: options[:server], port: options[:port])
						redis.lpush(options[:list], logstash_msg.to_json)
					when 'udp'
						socket = UDPSocket.new
						socket.send(logstash_msg.to_json, 0, options[:server], options[:port])
						socket.close
					when 'tcp'
						socket = TCPSocket.new(options[:server], options[:port])
						socket.puts(JSON.generate(logstash_msg))
						socket.close
					end
				rescue => e
					@logger.warn("LOGSTASH RUN ERROR: #{e.message}")
					yield e.inspect, 0
				end
			end

			def run(event)
				forwarder(event)
			end
			
			# Called when Sensu begins to shutdown.
			def stop
				true
			end
		end
	end
end