forked from vklochan/python-logstash
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathhandler_amqp.py
More file actions
132 lines (104 loc) · 4.73 KB
/
Copy pathhandler_amqp.py
File metadata and controls
132 lines (104 loc) · 4.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import json
try:
from urllib import urlencode
except ImportError:
from urllib.parse import urlencode
from logging import Filter
from logging.handlers import SocketHandler
import pika
from logstash import formatter
class AMQPLogstashHandler(SocketHandler, object):
"""AMQP Log Format handler
:param host: AMQP host (default 'localhost')
:param port: AMQP port (default 5672)
:param username: AMQP user name (default 'guest', which is the default for
RabbitMQ)
:param password: AMQP password (default 'guest', which is the default for
RabbitMQ)
:param exchange: AMQP exchange. Default 'logging.gelf'.
A queue binding must be defined on the server to prevent
log messages from being dropped.
:param exchange_type: AMQP exchange type (default 'fanout').
:param durable: AMQP exchange is durable (default False)
:param virtual_host: AMQP virtual host (default '/').
:param passive: exchange is declared passively, meaning that an error is
raised if the exchange does not exist, and succeeds otherwise. This is
useful if the user does not have configure permission on the exchange.
:param tags: list of tags for a logger (default is None).
:param message_type: The type of the message (default logstash).
:param version: version of logstash event schema (default is 0).
:param extra_fields: Send extra fields on the log record to graylog
if true (the default)
:param fqdn: Use fully qualified domain name of localhost as source
host (socket.getfqdn()).
:param facility: Replace facility with specified value. If specified,
record.name will be passed as `logger` parameter.
"""
def __init__(self, host='localhost', port=5672, username='guest',
password='guest', exchange='logstash', exchange_type='fanout',
virtual_host='/', message_type='logstash', tags=None,
durable=False, passive=False, version=0, extra_fields=True,
fqdn=False, facility=None, exchange_routing_key=''):
# AMQP parameters
self.host = host
self.port = port
self.username = username
self.password = password
self.exchange_type = exchange_type
self.exchange = exchange
self.exchange_is_durable = durable
self.declare_exchange_passively = passive
self.virtual_host = virtual_host
self.routing_key = exchange_routing_key
SocketHandler.__init__(self, host, port)
# Extract Logstash paramaters
self.tags = tags or []
fn = formatter.LogstashFormatterVersion1 if version == 1 \
else formatter.LogstashFormatterVersion0
self.formatter = fn(message_type, tags, fqdn)
# Standard logging parameters
self.extra_fields = extra_fields
self.fqdn = fqdn
self.facility = facility
def makeSocket(self, **kwargs):
return PikaSocket(self.host,
self.port,
self.username,
self.password,
self.virtual_host,
self.exchange,
self.routing_key,
self.exchange_is_durable,
self.declare_exchange_passively,
self.exchange_type)
def makePickle(self, record):
return self.formatter.format(record)
class PikaSocket(object):
def __init__(self, host, port, username, password, virtual_host, exchange,
routing_key, durable, passive, exchange_type):
# create connection parameters
credentials = pika.PlainCredentials(username, password)
parameters = pika.ConnectionParameters(host, port, virtual_host,
credentials)
# create connection & channel
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
# create an exchange, if needed
self.channel.exchange_declare(exchange=exchange,
exchange_type=exchange_type,
passive=passive,
durable=durable)
# needed when publishing
self.spec = pika.spec.BasicProperties(delivery_mode=2)
self.routing_key = routing_key
self.exchange = exchange
def sendall(self, data):
self.channel.basic_publish(self.exchange,
self.routing_key,
data,
properties=self.spec)
def close(self):
try:
self.connection.close()
except Exception:
pass