How to setup Logstash agent secure SSL communication with RabbitMQ and eventually to ELK server

[satishkori (Satish)]

Hi all,

Firstly does Logstash agent supports SSL with RabbitMQ?

if so ,could you list the steps to ships logs using logstash agent on one server to RabbitMQ server and finally the ELK server on another machine should be able to read it from the RabbitMQ queue.

[magnusbaeck (Magnus Bäck)]

Firstly does Logstash agent supports SSL with RabbitMQ?

The rabbitmq output plugin has an ssl option, so yes.

if so ,could you list the steps to ships logs using logstash agent on one server to RabbitMQ server and finally the ELK server on another machine should be able to read it from the RabbitMQ queue.

I suspect nobody will take the time to describe this in any great detail. If you ask more specific questions you may have better luck getting response.

[satishkori (Satish)]

Thnx for the reply.However I have already setup using the below configuration but I am getting SSL communication error in logstash logs and logstash agent is unable to connect to RabbitMQ server.

vi /etc/rabbitmq/rabbitmq.config
%% -- mode: erlang --
[{rabbit, [{ssl, true},
{ssl_listeners, ["127.0.0.1", 15671]},
{auth_mechanisms, ['EXTERNAL', 'PLAIN']},
{ssl_options, [{cacertfile,"/etc/pki/tls/testca/cacert.pem"},
{certfile,"/etc/pki/tls/server/cert.pem"},
{keyfile,"/etc/pki/tls/server/key.pem"},
{password, "client1234passwd"},
{verify,verify_peer},
{fail_if_no_peer_cert,true}]}
]},
{rabbitmq_management,
[{listener, [{port, 15671},
{ssl, true},
{auth_mechanisms, ['EXTERNAL', 'PLAIN']},
{ssl_opts, [{cacertfile, "/etc/pki/tls/testca/cacert.pem"},
{certfile, "/etc/pki/tls/server/cert.pem"},
{keyfile, "/etc/pki/tls/server/key.pem"}
{password, "client1234passwd"},
{verify,verify_peer},
{fail_if_no_peer_cert, true}]}
]}
]}
].
For SSL : Ensure to create a CA and sign certificates with the CA.
For creation of user,vhost,exchange,exchange-bindings,queue etc:
./rabbitmqadmin declare exchange name=logstash-exchange type=direct -u sat -p sat

python rabbitmqadmin.py declare exchange name=logger type=topic -u username -p password

./rabbitmqadmin declare queue name=indexer-queue auto_delete=false durable=true -u sat -p sat

./rabbitmqadmin declare binding source=logstash-exchange destination=indexer-queue routing_key=logstash-routing_key -u sat -p sat

./rabbitmqadmin publish exchange=logstash-exchange routing_key=logstash-routing_key payload="hello, world"

Logstash Server configuration:vi /etc/logstash/conf.d/elastic-rabbit.conf

input {
rabbitmq {
host => "hidpuppet.example.com"
queue => "indexer-queue"
durable => true
key => "logstash-routing_key"
exchange => "logstash-exchange"
threads => 3
prefetch_count => 50
port => 5672
user => ""
password => "sat"
ssl => true
ssl_certificate_path => "/etc/pki/tls/server/cert.pem"
ssl_certificate_password => "client1234passwd"
}
}
filter {
if [type] == "syslog" {
syslog_pri { }
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
if !("_grokparsefailure" in [tags]) {
mutate {
replace => [ "@source_host", "%{syslog_hostname}" ]
replace => [ "@message", "%{syslog_message}" ]
}
}
mutate {
remove_field => [ "syslog_hostname", "syslog_message", "syslog_timestamp" ]
}
}
}
output {
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "syslog"
document_type => "log"
}
stdout { codec => rubydebug }
}

Logstash agent server configuration: vi /etc/logstash/conf.d/logstash-forward.conf

input {
file {
type => "syslog"
path => [ "/var/log/syslog" ]
}
heartbeat {
interval => 10
type => "heartbeat"
}
}
output {
rabbitmq {
exchange => "logstash-exchange"
exchange_type => "direct"
key => "logstash-routing_key"
host => "hidpuppet.example.com"
vhost => Some_Virtual_Host
durable => true
persistent => true
port => 5672
user => ""
password => "sat"
ssl => true
ssl_certificate_path => "/etc/pki/tls/client/cert.pem"
ssl_certificate_password => "client1234passwd"
}
stdout {
codec => rubydebug
}
}
logstash-forward.conf (END)

Error in logstash agents logs:

[2017-05-08T02:13:37,159][ERROR][logstash.agent ] Pipeline aborted due to error {:exception=>#<MarchHare::Session::SSLContextException: toDerInputStream rejects tag type 45

[satishkori (Satish)]

Error in logstash agents logs:

[2017-05-08T02:13:37,159][ERROR][logstash.agent ] Pipeline aborted due to error {:exception=>#<MarchHare::Session::SSLContextException: toDerInputStream rejects tag type 45

[2017-05-08T02:13:37,124][ERROR][logstash.pipeline ] Error registering plugin {:plugin=>"#<LogStash::OutputDelegator:0xcf4f6d9 @namespaced_metric=#<LogStash::Instrument::NamespacedMetric:0x2ef5c31f @metric=#<LogStash::Instrument::Metric:0x5d97b6d3 @collector=#<LogStash::Instrument::Collector:0x28b1126b @agent=nil, @metric_store=#<LogStash::Instrument::MetricStore:0x3d87b9a @store=#<Concurrent:0x3b5bdc5 @default_proc=nil>, @structured_lookup_mutex=#Mutex:0x63f4b24e, @fast_lookup=#<Concurrent:0x43f6d611 @default_proc=nil>>>>, @namespace_name=[:stats, :pipelines, :main, :plugins, :outputs, :"8060fc8ed7fdb640b58e0561cd033303b3cfc16a-3"]>, @metric=#<LogStash::Instrument::NamespacedMetric:0x2d47bf78 @metric=#<LogStash::Instrument::Metric:0x5d97b6d3 @collector=#<LogStash::Instrument::Collector:0x28b1126b @agent=nil, @metric_store=#

[magnusbaeck (Magnus Bäck)]

[{rabbit, [{ssl, true},
{ssl_listeners, ["127.0.0.1", 15671]},

Surely you want the SSL-wrapped AMQP listener to run on port 5671, not 15671?

{rabbitmq_management,
[{listener, [{port, 15671},

Especially since you're running the management interface listener on port 15671.

port => 5672

And yet you're telling Logstash to connect to port 5672?