Class: Observability::Collector::RabbitMQ

Inherits:
Observability::Collector show all
Extended by:
Configurability, Loggability
Defined in:
lib/observability/collector/rabbitmq.rb

Overview

A collector that re-injects events over AMQP to a RabbitMQ cluster.

Constant Summary collapse

MAX_EVENT_BYTES =

The maximum size of event messages

64 * 1024
LOOP_TIMER =

The number of seconds to wait between IO loops

0.25
DEFAULT_PUBLISH_OPTIONS =

Default options for publication

{
	mandatory:  false,
	persistent: true
}

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Observability::Collector

configured_type, inherited, start

Constructor Details

#initialize ⇒ RabbitMQ

Create a new UDP collector



106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/observability/collector/rabbitmq.rb', line 106

def initialize
	super

	@socket        = UDPSocket.new
	@amqp_session  = nil
	@amqp_channel  = Concurrent::ThreadLocalVar.new { @amqp_session.create_channel }
	@amqp_exchange = Concurrent::ThreadLocalVar.new do
		@amqp_channel.value.headers( self.class.exchange, passive: true )
	end
	@processing    = false
end

Class Method Details

.amqp_session_options ⇒ Object

Fetch a Hash of AMQP options.



67
68
69
70
71
72
73
74
75
 
# File 'lib/observability/collector/rabbitmq.rb', line 67

def self::amqp_session_options
	return {
		logger: Loggability[ Observability ],
		heartbeat: self.heartbeat,
		exchange: self.exchange,
		vhost: self.vhost,
		threaded: self.threaded,
	}
end

.capabilities_list(server_info) ⇒ Object

Return a formatted list of the server’s capabilities listed in server_info.



79
80
81
82
83
 
# File 'lib/observability/collector/rabbitmq.rb', line 79

def self::capabilities_list( server_info )
	server_info.
		map {|name,enabled| enabled ? name : nil }.
		compact.join(', ')
end

.configured_amqp_session ⇒ Object

Establish the connection to RabbitMQ based on the loaded configuration.



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/observability/collector/rabbitmq.rb', line 87

def self::configured_amqp_session
	uri = self.broker_uri or raise "No broker_uri configured."
	options = self.amqp_session_options

	session = Bunny.new( uri, options )
	session.start

	self.log.info "Connected to %s v%s server: %s" % [
		session.server_properties['product'],
		session.server_properties['version'],
		self.capabilities_list( session.server_properties['capabilities'] ),
	]

	return session
end

Instance Method Details

#read_next_event ⇒ Object

Read the next event from the socket



161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/observability/collector/rabbitmq.rb', line 161

def read_next_event
	self.log.debug "Reading next event."
	data = @socket.recv_nonblock( MAX_EVENT_BYTES, exception: false )

	if data == :wait_readable
		IO.select( [@socket], nil, nil, LOOP_TIMER )
		return nil
	elsif data.empty?
		return nil
	else
		self.log.info "Read %d bytes" % [ data.bytesize ]
		return JSON.parse( data )
	end
end

#start ⇒ Object

Start receiving events.



124
125
126
127
128
129
130
131
 
# File 'lib/observability/collector/rabbitmq.rb', line 124

def start
	self.log.info "Starting up."

	@amqp_session = self.class.configured_amqp_session
	@socket.bind( self.class.host, self.class.port )

	self.start_processing
end

#start_processing ⇒ Object

Start consuming incoming events and storing them.



144
145
146
147
148
149
150
151
 
# File 'lib/observability/collector/rabbitmq.rb', line 144

def start_processing
	@processing = true
	while @processing
		event = self.read_next_event or next
		self.log.debug "Read event: %p" % [ event ]
		self.store_event( event )
	end
end

#stop ⇒ Object

Stop receiving events.



135
136
137
138
139
140
 
# File 'lib/observability/collector/rabbitmq.rb', line 135

def stop
	self.stop_processing

	@socket.shutdown( :SHUT_RDWR )
	@amqp_session.close
end

#stop_processing ⇒ Object

Stop consuming events.



155
156
157
 
# File 'lib/observability/collector/rabbitmq.rb', line 155

def stop_processing
	@processing = false
end

#store_event(event) ⇒ Object

Store the specified event.



178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/observability/collector/rabbitmq.rb', line 178

def store_event( event )
	time    = event.delete( '@timestamp' )
	type    = event.delete( '@type' )
	version = event.delete( '@version' )

	data = JSON.generate( event )
	headers = {
		time: time,
		type: type,
		version: version,
		content_type: 'application/json',
		content_encoding: data.encoding.name,
		timestamp: Time.now.to_f,
	}

	@amqp_exchange.value.publish( data, headers )
end