github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 06:37:19 -04:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #338 from mzaccari/master

Browse source 
LOGSTASH-691: move amqp plugin to be an external plugin
This commit is contained in:
Jordan Sissel 2013-02-12 12:00:47 -08:00
commit 3146d3cb81
15 changed files with 160 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/index.html.erb
View file

@ -20,7 +20,7 @@ layout: content_right
<li> <a href="tutorials/getting-started-centralized"> getting started (centralized) </a> </li>
<li> <a href="tutorials/10-minute-walkthrough"> 10-minute walkthrough</a> - a simple walkthrough to show you how to configure the logstash agent to process events and even old logs. </li>
<li> <a href="tutorials/metrics-from-logs"> Gathering metrics from logs </a> - take metrics from logs and ship them to graphite, ganglia, and more. </li>
<li> <a href="tutorials/just-enough-amqp-for-logstash">Just enough AMQP for Logstash </a> - Get a quick primer on AMQP and how to use it in Logstash! </li>
<li> <a href="tutorials/just-enough-rabbitmq-for-logstash">Just enough RabbitMQ for Logstash </a> - Get a quick primer on RabbitMQ and how to use it in Logstash! </li>
</ul>
<h3> books and articles </h3>

2
docs/logging-tool-comparisons.md
View file

@ -22,7 +22,7 @@ It provides you a simple event pipeline for taking events and logs from any
input, manipulating them with filters, and sending them to any output. Inputs
can be files, network, message brokers, etc. Filters are date and string
parsers, grep-like, etc. Outputs are data stores (elasticsearch, mongodb, etc),
message systems (amqp, stomp, etc), network (tcp, syslog), etc.
message systems (rabbitmq, stomp, etc), network (tcp, syslog), etc.
It also provides a web interface for doing search and analytics on your
logs.

47
docs/tutorials/just-enough-amqp-for-logstash.md → docs/tutorials/just-enough-rabbitmq-for-logstash.md
View file

@ -1,10 +1,10 @@
---
title: Just Enough AMQP - logstash
title: Just Enough RabbitMQ - logstash
layout: content_right
---
While configuring your AMQP broker is out of scope for logstash, it's important
to understand how logstash uses AMQP. To do that, we need to understand a
While configuring your RabbitMQ broker is out of scope for logstash, it's important
to understand how logstash uses RabbitMQ. To do that, we need to understand a
little about AMQP.
You should also consider reading
@ -35,9 +35,8 @@ routing key. Routing keys are discussed below.
## Broker
A broker is simply the AMQP server software. There are several brokers but the
most common (and arguably popular) is [RabbitMQ](http://www.rabbitmq.com).
Some others are Apache Qpid (and the commercial version - RedHat MRG)
A broker is simply the AMQP server software. There are several brokers, but this
tutorial will cover the most common (and arguably popular), [RabbitMQ](http://www.rabbitmq.com).
# Routing Keys
@ -112,19 +111,19 @@ support routing keys. Topic exchanges do support them. Just like a fanout
exchange, all bound queues see all messages with the additional filter of the
routing key.
# AMQP in logstash
# RabbitMQ in logstash
As stated earlier, in Logstash, Outputs publish to Exchanges. Inputs read from
Queues that are bound to Exchanges. Logstash uses the `bunny` AMQP library for
Queues that are bound to Exchanges. Logstash uses the `bunny` RabbitMQ library for
interaction with a broker. Logstash endeavors to expose as much of the
configuration for both exchanges and queues. There are many different tunables
that you might be concerned with setting - including things like message
durability or persistence of declared queues/exchanges. See the relevant input
and output documentation for AMQP for a full list of tunables.
and output documentation for RabbitMQ for a full list of tunables.
# Sample configurations, tips, tricks and gotchas
There are several examples in the logstash source directory of AMQP usage,
There are several examples in the logstash source directory of RabbitMQ usage,
however a few general rules might help eliminate any issues.
## Check your bindings
@ -136,9 +135,9 @@ sender agent
input { stdin { type = "test" } }
output {
amqp {
name => "test_exchange"
host => "my_amqp_server"
rabbitmq {
exchange => "test_exchange"
host => "my_rabbitmq_server"
exchange_type => "fanout"
}
}
@ -146,9 +145,9 @@ sender agent
receiver agent
input {
amqp {
name => "test_queue"
host => "my_amqp_server"
rabbitmq {
queue => "test_queue"
host => "my_rabbitmq_server"
exchange => "test_exchange" # This matches the exchange declared above
}
}
@ -157,15 +156,15 @@ receiver agent
## Message persistence
By default, logstash will attempt to ensure that you don't lose any messages.
This is reflected in the AMQP default settings as well. However there are
cases where you might not want this. A good example is where AMQP is not your
This is reflected in the RabbitMQ default settings as well. However there are
cases where you might not want this. A good example is where RabbitMQ is not your
primary method of shipping.
In the following example, we use AMQP as a sniffing interface. Our primary
destination is the embedded ElasticSearch instance. We have a secondary AMQP
In the following example, we use RabbitMQ as a sniffing interface. Our primary
destination is the embedded ElasticSearch instance. We have a secondary RabbitMQ
output that we use for duplicating messages. However we disable persistence and
durability on this interface so that messages don't pile up waiting for
delivery. We only use AMQP when we want to watch messages in realtime.
delivery. We only use RabbitMQ when we want to watch messages in realtime.
Additionally, we're going to leverage routing keys so that we can optionally
filter incoming messages to subsets of hosts. The exercise of getting messages
to this logstash agent are left up to the user.
@ -176,9 +175,9 @@ to this logstash agent are left up to the user.
output {
elasticsearch { embedded => true }
amqp {
name => "logtail"
host => "my_amqp_server"
rabbitmq {
exchange => "logtail"
host => "my_rabbitmq_server"
exchange_type => "topic" # We use topic here to enable pub/sub with routing keys
key => "logs.%{host}"
durable => false # If rabbitmq restarts, the exchange disappears.

10
etc/agent.lgtm.conf
View file

@ -17,17 +17,17 @@ input {
output {
# This will be your durable shipping mechanism
amqp {
host => "myamqpserver"
rabbitmq {
host => "myrabbitmqserver"
exchange_type => "fanout"
name => "rawlogs"
exchange => "rawlogs"
}
# This is an optional non-durable shipping mechanism
# With this, you can sniff logs from your own code
amqp {
rabbitmq {
host => "127.0.0.1"
exchange_type => "topic"
name => "logsniff"
exchange => "logsniff"
durable => false
persistent => false
# The following is optional

2
etc/examples/esriver.conf
View file

@ -16,6 +16,6 @@ output {
stdout { }
elasticsearch_river {
es_host => "localhost"
amqp_host => "localhost"
rabbitmq_host => "localhost"
}
}

4
etc/examples/indexer.conf
View file

@ -1,10 +1,10 @@
input {
amqp {
rabbitmq {
host => "127.0.0.1"
user => "guest"
pass => "guest"
exchange => "logstash"
name => "testing"
queue => "testing"
type => "all"
}

4
lib/logstash/agent.rb
View file

@ -152,8 +152,8 @@ class LogStash::Agent
# Load any plugins that we have flags for.
# TODO(sissel): The --<plugin> flag support currently will load
# any matching plugins input, output, or filter. This means, for example,
# that the 'amqp' input *and* output plugin will be loaded if you pass
# --amqp-foo flag. This might cause confusion, but it seems reasonable for
# that the 'rabbitmq' input *and* output plugin will be loaded if you pass
# --rabbitmq-foo flag. This might cause confusion, but it seems reasonable for
# now that any same-named component will have the same flags.
plugins = []
args.each do |arg|

4
lib/logstash/config/file.rb
View file

@ -42,7 +42,7 @@ class LogStash::Config::File
tryload o[:type], :base
type = registry[o[:type]]
# Load the plugin itself (inputs/file, outputs/amqp, etc)
# Load the plugin itself (inputs/file, outputs/rabbitmq, etc)
# TODO(sissel): Error handling
tryload o[:type], o[:plugin]
plugin = registry[o[:plugin]]
@ -83,7 +83,7 @@ class LogStash::Config::File
@config.each do |type, plugin_config_array|
# plugin_config_array has arrays of each component config:
# input {
# amqp { ... }
# rabbitmq { ... }
# file { ... }
# file { ... }
# }

2
lib/logstash/config/test.conf
View file

@ -1,5 +1,5 @@
input {
amqp {
rabbitmq {
port => 12345
tag => [ a, b, c ]
}

2
lib/logstash/inputs/gemfire.rb
View file

@ -234,4 +234,4 @@ class LogStash::Inputs::Gemfire < LogStash::Inputs::Threadable
def afterRegionInvalidate(event)
@logger.debug("afterRegionInvalidate #{event}")
end
end # class LogStash::Inputs::Amqp
end # class LogStash::Inputs::Gemfire

89
lib/logstash/inputs/amqp.rb → lib/logstash/inputs/rabbitmq.rb
View file

@ -2,44 +2,42 @@ require "logstash/inputs/threadable"
require "logstash/namespace"
require "cgi" # for CGI.escape
# Pull events from an AMQP exchange.
#
# <b> NOTE: THIS IS ONLY KNOWN TO WORK WITH RECENT RELEASES OF RABBITMQ. Any
# other amqp broker will not work with this plugin. I do not know why. If you
# need support for brokers other than rabbitmq, please file bugs here:
# <https://github.com/ruby-amqp/bunny> </b>
# Pull events from a RabbitMQ exchange.
#
# The default settings will create an entirely transient queue and listen for all messages by default.
# If you need durability or any other advanced settings, please set the appropriate options
class LogStash::Inputs::Amqp < LogStash::Inputs::Threadable
#
# This has been tested with Bunny 0.9.x, which supports RabbitMQ 2.x and 3.x. You can
# find links to both here:
#
# * RabbitMQ - <http://www.rabbitmq.com/>
# * Bunny - <https://github.com/ruby-amqp/bunny>
class LogStash::Inputs::RabbitMQ < LogStash::Inputs::Threadable
config_name "amqp"
plugin_status "unsupported"
config_name "rabbitmq"
plugin_status "beta"
# Your amqp broker's custom arguments. For mirrored queues in RabbitMQ: [ "x-ha-policy", "all" ]
# Custom arguments. For example, mirrored queues in RabbitMQ 2.x: [ "x-ha-policy", "all" ]
# RabbitMQ 3.x mirrored queues are set by policy. More information can be found
# here: http://www.rabbitmq.com/blog/2012/11/19/breaking-things-with-rabbitmq-3-0/
config :arguments, :validate => :array, :default => []
# Your amqp server address
# Your rabbitmq server address
config :host, :validate => :string, :required => true
# The AMQP port to connect on
# The rabbitmq port to connect on
config :port, :validate => :number, :default => 5672
# Your amqp username
# Your rabbitmq username
config :user, :validate => :string, :default => "guest"
# Your amqp password
# Your rabbitmq password
config :password, :validate => :password, :default => "guest"
# The name of the queue. Depricated due to conflicts with puppet naming convention.
# Replaced by 'queue' variable. See LOGSTASH-755
config :name, :validate => :string, :deprecated => true
# The name of the queue.
config :queue, :validate => :string, :default => ""
# The name of the exchange to bind the queue. This is analogous to the 'amqp
# output' [config 'name'](../outputs/amqp)
# The name of the exchange to bind the queue.
config :exchange, :validate => :string, :required => true
# The routing key to use. This is only valid for direct or fanout exchanges
@ -80,6 +78,9 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Threadable
# Validate SSL certificate
config :verify_ssl, :validate => :boolean, :default => false
# Maximum permissible size of a frame (in bytes) to negotiate with clients
config :frame_max, :validate => :number, :default => 131072
public
def initialize(params)
@ -90,43 +91,41 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Threadable
end # def initialize
public
def register
if @name
if @queue
@logger.error("'name' and 'queue' are the same setting, but 'name' is deprecated. Please use only 'queue'")
end
@queue = @name
end
def register
@logger.info("Registering input #{@url}")
require "bunny" # rubygem 'bunny'
require "bunny"
@vhost ||= "/"
@port ||= 5672
@key ||= "#"
@amqpsettings = {
@rabbitmq_settings = {
:vhost => @vhost,
:host => @host,
:port => @port,
}
@amqpsettings[:user] = @user if @user
@amqpsettings[:pass] = @password.value if @password
@amqpsettings[:logging] = @debug
@amqpsettings[:ssl] = @ssl if @ssl
@amqpsettings[:verify_ssl] = @verify_ssl if @verify_ssl
@amqpurl = "amqp://"
@rabbitmq_settings[:user] = @user if @user
@rabbitmq_settings[:pass] = @password.value if @password
@rabbitmq_settings[:logging] = @debug
@rabbitmq_settings[:ssl] = @ssl if @ssl
@rabbitmq_settings[:verify_ssl] = @verify_ssl if @verify_ssl
@rabbitmq_settings[:frame_max] = @frame_max if @frame_max
@rabbitmq_url = "amqp://"
if @user
@amqpurl << @user if @user
@amqpurl << ":#{CGI.escape(@password.to_s)}" if @password
@amqpurl << "@"
@rabbitmq_url << @user if @user
@rabbitmq_url << ":#{CGI.escape(@password.to_s)}" if @password
@rabbitmq_url << "@"
end
@amqpurl += "#{@host}:#{@port}#{@vhost}/#{@queue}"
@rabbitmq_url += "#{@host}:#{@port}#{@vhost}/#{@queue}"
end # def register
def run(queue)
begin
@logger.debug("Connecting with AMQP settings #{@amqpsettings.inspect} to set up queue #{@queue.inspect}")
@bunny = Bunny.new(@amqpsettings)
@logger.debug("Connecting with RabbitMQ settings #{@rabbitmq_settings.inspect} to set up queue #{@queue.inspect}")
@bunny = Bunny.new(@rabbitmq_settings)
return if terminating?
@bunny.start
@bunny.qos({:prefetch_count => @prefetch_count})
@ -137,14 +136,14 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Threadable
@bunnyqueue.bind(@exchange, :key => @key)
@bunnyqueue.subscribe({:ack => @ack}) do |data|
e = to_event(data[:payload], @amqpurl)
e = to_event(data[:payload], @rabbitmq_url)
if e
queue << e
end
end # @bunnyqueue.subscribe
rescue *[Bunny::ConnectionError, Bunny::ServerDownError] => e
@logger.error("AMQP connection error, will reconnect: #{e}")
@logger.error("RabbitMQ connection error, will reconnect: #{e}")
# Sleep for a bit before retrying.
# TODO(sissel): Write 'backoff' method?
sleep(1)
@ -158,4 +157,4 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Threadable
@bunny.close if @bunny
finished
end # def teardown
end # class LogStash::Inputs::Amqp
end # class LogStash::Inputs::RabbitMQ

22
lib/logstash/inputs/zenoss.rb
View file

@ -1,25 +1,25 @@
require "date"
require "logstash/inputs/amqp"
require "logstash/inputs/rabbitmq"
require "zlib"
# Read Zenoss events from the zenoss.zenevents fanout exchange.
#
class LogStash::Inputs::Zenoss < LogStash::Inputs::Amqp
class LogStash::Inputs::Zenoss < LogStash::Inputs::RabitMQ
config_name "zenoss"
plugin_status "experimental"
# Your amqp server address
# Your rabbitmq server address
config :host, :validate => :string, :default => "localhost"
# Your amqp username
# Your rabbitmq username
config :user, :validate => :string, :default => "zenoss"
# Your amqp password
# Your rabbitmq password
config :password, :validate => :password, :default => "zenoss"
# The name of the exchange to bind the queue. This is analogous to the 'amqp
# output' [config 'name'](../outputs/amqp)
# The name of the exchange to bind the queue. This is analogous to the 'rabbitmq
# output' [config 'name'](../outputs/rabbitmq)
config :exchange, :validate => :string, :default => "zenoss.zenevents"
# The routing key to use. This is only valid for direct or fanout exchanges
@ -41,8 +41,8 @@ class LogStash::Inputs::Zenoss < LogStash::Inputs::Amqp
begin
zep = Org::Zenoss::Protobufs::Zep
@logger.debug("Connecting with AMQP settings #{@amqpsettings.inspect}")
@bunny = Bunny.new(@amqpsettings)
@logger.debug("Connecting with RabbitMQ settings #{@rabbitmq_settings.inspect}")
@bunny = Bunny.new(@rabbitmq_settings)
return if terminating?
@bunny.start
@bunny.qos({:prefetch_count => @prefetch_count})
@ -77,7 +77,7 @@ class LogStash::Inputs::Zenoss < LogStash::Inputs::Amqp
# LogStash event properties.
event = LogStash::Event.new({
"@source" => @amqpurl,
"@source" => @rabbitmq_url,
"@type" => @type,
"@timestamp" => timestamp,
"@source_host" => occurrence.actor.element_title,
@ -130,7 +130,7 @@ class LogStash::Inputs::Zenoss < LogStash::Inputs::Amqp
end # @queue.subscribe
rescue *[Bunny::ConnectionError, Bunny::ServerDownError] => e
@logger.error("AMQP connection error, will reconnect: #{e}")
@logger.error("RabbitMQ connection error, will reconnect: #{e}")
# Sleep for a bit before retrying.
# TODO(sissel): Write 'backoff' method?
sleep(1)

58
lib/logstash/outputs/elasticsearch_river.rb
View file

@ -5,8 +5,8 @@ require "uri"
require "net/http"
# This output lets you store logs in elasticsearch. It's similar to the
# 'elasticsearch' output but improves performance by using an AMQP server,
# such as rabbitmq, to send data to elasticsearch.
# 'elasticsearch' output but improves performance by using a queue server,
# rabbitmq, to send data to elasticsearch.
#
# Upon startup, this output will automatically contact an elasticsearch cluster
# and configure it to read from the queue to which we write.
@ -41,43 +41,42 @@ class LogStash::Outputs::ElasticSearchRiver < LogStash::Outputs::Base
# ElasticSearch river configuration: bulk timeout in milliseconds
config :es_bulk_timeout_ms, :validate => :number, :default => 100
# ElasticSearch river configuration: is ordered?
config :es_ordered, :validate => :boolean, :default => false
# Hostname of AMQP server
config :amqp_host, :validate => :string, :required => true
# Hostname of RabbitMQ server
config :rabbitmq_host, :validate => :string, :required => true
# Port of AMQP server
config :amqp_port, :validate => :number, :default => 5672
# Port of RabbitMQ server
config :rabbitmq_port, :validate => :number, :default => 5672
# AMQP user
# RabbitMQ user
config :user, :validate => :string, :default => "guest"
# AMQP password
# RabbitMQ password
config :password, :validate => :string, :default => "guest"
# AMQP vhost
# RabbitMQ vhost
config :vhost, :validate => :string, :default => "/"
# AMQP queue name. Depricated due to conflicts with puppet naming convention.
# Replaced by 'queue' variable. See LOGSTASH-755
config :name, :validate => :string, :deprecated => true
# AMQP queue name
# RabbitMQ queue name
config :queue, :validate => :string, :default => "elasticsearch"
# AMQP exchange name
# RabbitMQ exchange name
config :exchange, :validate => :string, :default => "elasticsearch"
# The exchange type (fanout, topic, direct)
config :exchange_type, :validate => [ "fanout", "direct", "topic"],
:default => "direct"
# AMQP routing key
# RabbitMQ routing key
config :key, :validate => :string, :default => "elasticsearch"
# AMQP durability setting. Also used for ElasticSearch setting
# RabbitMQ durability setting. Also used for ElasticSearch setting
config :durable, :validate => :boolean, :default => true
# AMQP persistence setting
# RabbitMQ persistence setting
config :persistent, :validate => :boolean, :default => true
# The document ID for the index. Useful for overwriting existing entries in
@ -106,12 +105,12 @@ class LogStash::Outputs::ElasticSearchRiver < LogStash::Outputs::Base
protected
def prepare_river
require "logstash/outputs/amqp"
require "logstash/outputs/rabbitmq"
# Configure the message plugin
params = {
"host" => [@amqp_host],
"port" => [@amqp_port],
"host" => [@rabbitmq_host],
"port" => [@rabbitmq_port],
"user" => [@user],
"password" => [@password],
"exchange_type" => [@exchange_type],
@ -122,7 +121,7 @@ class LogStash::Outputs::ElasticSearchRiver < LogStash::Outputs::Base
"persistent" => [@persistent.to_s],
"debug" => [@debug.to_s],
}.reject {|k,v| v.first.nil?}
@mq = LogStash::Outputs::Amqp.new(params)
@mq = LogStash::Outputs::RabbitMQ.new(params)
@mq.register
# Set up the river
@ -132,13 +131,17 @@ class LogStash::Outputs::ElasticSearchRiver < LogStash::Outputs::Base
# Name the river by our hostname
require "socket"
hostname = Socket.gethostname
api_path = "/_river/logstash-#{hostname.gsub('.','_')}/_meta"
@status_path = "/_river/logstash-#{hostname.gsub('.','_')}/_status"
# Replace spaces with hyphens and remove all non-alpha non-dash non-underscore characters
river_name = "#{hostname} #{@queue}".gsub(' ', '-').gsub(/[^\w-]/, '')
api_path = "/_river/logstash-#{river_name}/_meta"
@status_path = "/_river/logstash-#{river_name}/_status"
river_config = {"type" => "rabbitmq",
"rabbitmq" => {
"host" => @amqp_host=="localhost" ? hostname : @amqp_host,
"port" => @amqp_port,
"host" => @rabbitmq_host=="localhost" ? hostname : @rabbitmq_host,
"port" => @rabbitmq_port,
"user" => @user,
"pass" => @password,
"vhost" => @vhost,
@ -147,10 +150,11 @@ class LogStash::Outputs::ElasticSearchRiver < LogStash::Outputs::Base
"routing_key" => @key,
"exchange_type" => @exchange_type,
"exchange_durable" => @durable.to_s,
"queue_durable" => @durable.to_s,
"queue_durable" => @durable.to_s
},
"index" => {"bulk_size" => @es_bulk_size,
"bulk_timeout" => "#{@es_bulk_timeout_ms}ms",
"ordered" => @es_ordered
},
}
@logger.info("ElasticSearch using river", :config => river_config)

73
lib/logstash/outputs/amqp.rb → lib/logstash/outputs/rabbitmq.rb
View file

@ -1,40 +1,37 @@
require "logstash/outputs/base"
require "logstash/namespace"
# Push events to an AMQP exchange.
# Push events to a RabbitMQ exchange.
#
# <b> NOTE: THIS IS ONLY KNOWN TO WORK WITH RECENT RELEASES OF RABBITMQ. Any
# other amqp broker will not work with this plugin. I do not know why. If you
# need support for brokers other than rabbitmq, please file bugs here:
# <https://github.com/ruby-amqp/bunny> </b>
class LogStash::Outputs::Amqp < LogStash::Outputs::Base
# This has been tested with Bunny 0.9.x, which supports RabbitMQ 2.x and 3.x. You can
# find links to both here:
#
# * RabbitMQ - <http://www.rabbitmq.com/>
# * Bunny - <https://github.com/ruby-amqp/bunny>
class LogStash::Outputs::RabbitMQ < LogStash::Outputs::Base
MQTYPES = [ "fanout", "direct", "topic" ]
config_name "amqp"
plugin_status "unsupported"
config_name "rabbitmq"
plugin_status "beta"
# Your amqp server address
# Your rabbitmq server address
config :host, :validate => :string, :required => true
# The AMQP port to connect on
# The rabbitmq port to connect on
config :port, :validate => :number, :default => 5672
# Your amqp username
# Your rabbitmq username
config :user, :validate => :string, :default => "guest"
# Your amqp password
# Your rabbitmq password
config :password, :validate => :password, :default => "guest"
# The name of the exchange
config :exchange, :validate => :string, :required => true
# The exchange type (fanout, topic, direct)
config :exchange_type, :validate => [ "fanout", "direct", "topic"], :required => true
# The name of the exchange. Depricated due to conflicts with puppet naming convention.
# Replaced by 'exchange' variable. See LOGSTASH-755
config :name, :validate => :string, :deprecated => true
# The name of the exchange
config :exchange, :validate => :string # TODO(sissel): Make it required when 'name' is gone
# Key to route to by default. Defaults to 'logstash'
#
# * Routing keys are ignored on fanout exchanges.
@ -46,7 +43,7 @@ class LogStash::Outputs::Amqp < LogStash::Outputs::Base
# Is this exchange durable? (aka; Should it survive a broker restart?)
config :durable, :validate => :boolean, :default => true
# Should messages persist to disk on the AMQP broker until they are read by a
# Should messages persist to disk on the rabbitmq broker until they are read by a
# consumer?
config :persistent, :validate => :boolean, :default => true
@ -59,16 +56,12 @@ class LogStash::Outputs::Amqp < LogStash::Outputs::Base
# Validate SSL certificate
config :verify_ssl, :validate => :boolean, :default => false
# Maximum permissible size of a frame (in bytes) to negotiate with clients
config :frame_max, :validate => :number, :default => 131072
public
def register
require "bunny" # rubygem 'bunny'
if @name
if @exchange
@logger.error("'name' and 'exchange' are the same setting, but 'name' is deprecated. Please use only 'exchange'")
end
@exchange = @name
end
require "bunny"
@logger.info("Registering output", :plugin => self)
connect
@ -76,27 +69,29 @@ class LogStash::Outputs::Amqp < LogStash::Outputs::Base
public
def connect
amqpsettings = {
rabbitmq_settings = {
:vhost => @vhost,
:host => @host,
:port => @port,
:logging => @debug,
}
amqpsettings[:user] = @user if @user
amqpsettings[:pass] = @password.value if @password
amqpsettings[:ssl] = @ssl if @ssl
amqpsettings[:verify_ssl] = @verify_ssl if @verify_ssl
rabbitmq_settings[:user] = @user if @user
rabbitmq_settings[:pass] = @password.value if @password
rabbitmq_settings[:ssl] = @ssl if @ssl
rabbitmq_settings[:verify_ssl] = @verify_ssl if @verify_ssl
rabbitmq_settings[:frame_max] = @frame_max if @frame_max
begin
@logger.debug("Connecting to AMQP", :settings => amqpsettings,
@logger.debug("Connecting to RabbitMQ", :settings => rabbitmq_settings,
:exchange_type => @exchange_type, :name => @exchange)
@bunny = Bunny.new(amqpsettings)
@bunny = Bunny.new(rabbitmq_settings)
@bunny.start
rescue => e
if terminating?
return
else
@logger.error("AMQP connection error (during connect), will reconnect",
@logger.error("RabbitMQ connection error (during connect), will reconnect",
:exception => e, :backtrace => e.backtrace)
sleep(1)
retry
@ -133,10 +128,10 @@ class LogStash::Outputs::Amqp < LogStash::Outputs::Base
@logger.debug(["Publishing message", { :destination => to_s, :message => message, :key => key }])
@bunnyexchange.publish(message, :persistent => @persistent, :key => key)
else
@logger.warn("Tried to send message, but not connected to amqp yet.")
@logger.warn("Tried to send message, but not connected to rabbitmq yet.")
end
rescue *[Bunny::ServerDownError, Errno::ECONNRESET] => e
@logger.error("AMQP connection error (during publish), will reconnect: #{e}")
@logger.error("RabbitMQ connection error (during publish), will reconnect: #{e}")
connect
retry
end
@ -154,4 +149,4 @@ class LogStash::Outputs::Amqp < LogStash::Outputs::Base
@bunnyexchange = nil
finished
end # def teardown
end # class LogStash::Outputs::Amqp
end # class LogStash::Outputs::RabbitMQ

2
logstash.gemspec
View file

@ -38,7 +38,7 @@ Gem::Specification.new do |gem|
gem.add_runtime_dependency "aws-sdk"
gem.add_runtime_dependency "heroku"
gem.add_runtime_dependency "addressable", ["~> 2.2.6"]
gem.add_runtime_dependency "bunny"
gem.add_runtime_dependency "bunny", [">= 0.9.0.pre6"]
gem.add_runtime_dependency "ffi"
gem.add_runtime_dependency "ffi-rzmq", ["0.9.3"]
gem.add_runtime_dependency "filewatch", ["0.5.1"]