Class: Syslogstash::SyslogReader

Inherits:
Object
  • Object
show all
Includes:
Worker
Defined in:
lib/syslogstash/syslog_reader.rb

Overview

A single socket reader.

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Worker

#stop, #thread, #wait

Constructor Details

#initialize(file, config, logstash, metrics) ⇒ SyslogReader

Returns a new instance of SyslogReader.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/syslogstash/syslog_reader.rb', line 10

def initialize(file, config, logstash, metrics)
	@file, @logstash, @metrics = file, logstash, metrics
	config ||= {}

	@add_fields = config['add_fields'] || {}
	@relay_to   = config['relay_to']   || []

	unless @add_fields.is_a? Hash
		raise ArgumentError,
		      "add_fields parameter to socket #{file} must be a hash"
	end

	unless @relay_to.is_a? Array
		raise ArgumentError,
		      "relay_to parameter to socket #{file} must be an array"
	end

	log { "initialized syslog socket #{file} with config #{config.inspect}" }
end

Instance Attribute Details

#file ⇒ Object (readonly)

Returns the value of attribute file.



8
9
10
 
# File 'lib/syslogstash/syslog_reader.rb', line 8

def file
  @file
end

Instance Method Details

#run ⇒ Object

Start reading from the socket file, parsing entries, and flinging them at logstash. This method will return, with the operation continuing in a separate thread.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/syslogstash/syslog_reader.rb', line 34

def run
	debug { "#run called" }

	begin
		socket = Socket.new(Socket::AF_UNIX, Socket::SOCK_DGRAM, 0)
		socket.bind(Socket.pack_sockaddr_un(@file))
		File.chmod(0666, @file)
	rescue Errno::EEXIST, Errno::EADDRINUSE
		log { "socket file #{@file} already exists; deleting" }
		File.unlink(@file) rescue nil
		retry
	rescue SystemCallError
		log { "Error while trying to bind to #{@file}" }
		raise
	end

	@worker = Thread.new do
		begin
			loop do
				msg = socket.recvmsg
				debug { "Message received: #{msg.inspect}" }
				@metrics.received(@file, Time.now)
				process_message msg.first.chomp
				relay_message msg.first
			end
		ensure
			socket.close
			log { "removing socket file #{@file}" }
			File.unlink(@file) rescue nil
		end
	end
end