Updating to amqp version 2.4.2

graypy/rabbitmq.py

@@ -8,7 +8,7 @@
 from logging import Filter
 from logging.handlers import SocketHandler
 
-from amqplib import client_0_8 as amqp  # pylint: disable=import-error
+import amqp  # pylint: disable=import-error
 
 from graypy.handler import BaseGELFHandler
 
@@ -78,7 +78,7 @@ def __init__(
         self.routing_key = routing_key
         BaseGELFHandler.__init__(self, **kwargs)
         SocketHandler.__init__(self, host, port)
-        self.addFilter(ExcludeFilter("amqplib"))
+        self.addFilter(ExcludeFilter("amqp"))
 
     def makeSocket(self, timeout=1):
         return RabbitSocket(
@@ -97,7 +97,8 @@ def __init__(self, cn_args, timeout, exchange, exchange_type, routing_key):
         self.exchange = exchange
         self.exchange_type = exchange_type
         self.routing_key = routing_key
-        self.connection = amqp.Connection(connection_timeout=timeout, **self.cn_args)
+        self.connection = amqp.Connection(connect_timeout=timeout, **self.cn_args)
+        self.connection.connect()
         self.channel = self.connection.channel()
         self.channel.exchange_declare(
             exchange=self.exchange,

setup.py

@@ -80,10 +80,10 @@ def run_tests(self):
         "pylint>=1.9.3,<2.0.0",
         "mock>=2.0.0,<3.0.0",
         "requests>=2.20.1,<3.0.0",
-        "amqplib>=1.0.2,<2.0.0",
+        "amqp>=2.4.2,<2.5.1",
     ],
     extras_require={
-        "amqp": ["amqplib==1.0.2"],
+        "amqp": ["amqp==2.4.2"],
         "docs": [
             "sphinx>=2.1.2,<3.0.0",
             "sphinx_rtd_theme>=0.4.3,<1.0.0",

tests/config/docker-compose-rmq.yml

@@ -0,0 +1,9 @@
+version: '2'
+services:
+  rabbitmq:
+    image: "rabbitmq:3.8-management-alpine"
+    volumes:
+      - ./setup_rabbitmq.sh:/usr/local/bin/setup_rabbitmq.sh
+    ports:
+      - "5672:5672"
+      - "15672:15672"

tests/config/inputs.json

@@ -39,6 +39,32 @@
             },
             "type": "org.graylog2.inputs.gelf.tcp.GELFTCPInput",
             "global": true
+        },
+        {
+            "title":"rmq",
+            "configuration": {
+                "heartbeat":60,
+                "prefetch":100,
+                "exchange_bind":true,
+                "broker_vhost":"/",
+                "broker_username":"graylog",
+                "decompress_size_limit":8388608,
+                "broker_port":5672,
+                "parallel_queues":1,
+                "broker_password":"graylog",
+                "throttling_allowed":false,
+                "exchange":"log-messages",
+                "tls":false,
+                "override_source":null,
+                "routing_key":"#",
+                "requeue_invalid_messages":true,
+                "broker_hostname":"rabbitmq",
+                "queue":"log-messages"
+            },
+            "static_fields":{},
+            "type":"org.graylog2.inputs.gelf.amqp.GELFAMQPInput",
+            "global":true,
+            "extractors":[]
         }
     ],
     "streams": [],

tests/config/setup_rabbitmq.sh

@@ -0,0 +1,5 @@
+rabbitmqctl add_user graylog graylog
+rabbitmqctl set_user_tags graylog administrator
+rabbitmqctl set_permissions -p / graylog ".*" ".*" ".*"
+
+rabbitmqadmin declare exchange name=log-messages type=direct -u graylog -p graylog

tests/config/start_local_graylog_server.sh

@@ -6,12 +6,25 @@
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 cd ${DIR}
 
+# remove a potential previous setup
+docker-compose -f docker-compose-rmq.yml down
+docker-compose -f docker-compose.yml down
+
+# first setup RabbitMQ, as it requires extra setup before Graylog can connect to it.
+docker-compose -f docker-compose-rmq.yml up -d
+
+# wait for RabbitMQ to start
+sleep 30
+
+# Configure a user and exchange for graylog to use
+RMQ_CONTAINER_ID="$(docker ps -qf "name=rabbit")"
+docker exec -ti ${RMQ_CONTAINER_ID} sh /usr/local/bin/setup_rabbitmq.sh
+
 # create ssl certs for enabling the graylog server to use a
 # TLS connection for GELF input
 bash create_ssl_certs.sh -h localhost -i 127.0.0.1
 
 # start the graylog server docker container
-docker-compose -f docker-compose.yml down
 docker-compose -f docker-compose.yml up -d
 
 # wait for the graylog server docker container to start

tests/config/stop_local_graylog_server.sh

@@ -6,4 +6,5 @@
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 cd ${DIR}
 
+docker-compose -f docker-compose-rmq.yml down
 docker-compose -f docker-compose.yml down

tests/integration/test_rmq_logging.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""pytests sending logs to Graylog through RabbitMQ"""
+
+import logging
+
+import pytest
+
+from graypy.rabbitmq import GELFRabbitHandler
+from graypy.handler import SYSLOG_LEVELS
+
+from tests.integration import LOCAL_GRAYLOG_UP
+from tests.integration.helper import get_unique_message, get_graylog_response
+
+
+@pytest.mark.skipif(not LOCAL_GRAYLOG_UP, reason="local Graylog instance not up")
+def test_rmq_logging():
+    """Test that verifies the log message was received by Graylog"""
+    logger = logging.getLogger("test_rmq_logging")
+    handler = GELFRabbitHandler(
+        url="amqp://graylog:graylog@127.0.0.1",
+        exchange="log-messages",
+        exchange_type="direct",
+        routing_key="#",
+    )
+    logger.addHandler(handler)
+    message = get_unique_message()
+    logger.error(message)
+    graylog_response = get_graylog_response(message)
+    assert message == graylog_response["message"]
+    assert "long_message" not in graylog_response
+    assert "timestamp" in graylog_response
+    assert SYSLOG_LEVELS[logging.ERROR] == graylog_response["level"]