forked from LogentriesCommunity/le_python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
128 lines (99 loc) · 3.48 KB
/
Copy pathutils.py
File metadata and controls
128 lines (99 loc) · 3.48 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
VERSION = '0.1'
import logging
import Queue
import threading
import socket
import random
import time
# Size of the internal event queue
QUEUE_SIZE = 32768
# Logentries API server address
LE_API = "api.logentries.com"
# Port number for token logging to Logentries API server
LE_PORT = 10000
# Minimal delay between attempts to reconnect in seconds
MIN_DELAY = 0.1
# Maximal delay between attempts to recconect in seconds
MAX_DELAY = 10
# LE appender signature - used for debugging messages
LE = "LE: "
# Error message displayed when an incorrect Token has been detected
INVALID_TOKEN = "\n\nIt appears the LOGENTRIES_TOKEN parameter you entered is incorrect!\n\n"
def dbg(msg):
print LE + msg
def check_token(token):
import re
valid = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
return valid.match(token)
class SocketAppender(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.daemon = True
self._conn = None
self._queue = Queue.Queue(QUEUE_SIZE)
def openConnection(self):
self._conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._conn.connect((LE_API, LE_PORT))
def reopenConnection(self):
self.closeConnection()
root_delay = MIN_DELAY
while True:
try:
self.openConnection()
return
except Exception:
dbg("Unable to connect to Logentries")
root_delay *= 2
if(root_delay > MAX_DELAY):
root_delay = MAX_DELAY
wait_for = root_delay + random.uniform(0, root_delay)
try:
time.sleep(wait_for)
except KeyboardInterrupt:
raise KeyboardInterrupt
def closeConnection(self):
if(self._conn is not None):
self._conn.close()
def run(self):
try:
# Open connection
self.reopenConnection()
# Send data in queue
while True:
# Take data from queue
data = self._queue.get(block=True)
# Send data, reconnect if needed
while True:
try:
self._conn.send(data)
except socket.error:
self.reopenConnection()
continue
break
except KeyboardInterrupt:
dbg("Logentries asynchronous socket client interrupted")
self.closeConnection()
class LogentriesHandler(logging.Handler):
def __init__(self, token):
logging.Handler.__init__(self)
self.token = token
self.good_config = True
if not check_token(token):
dbg(INVALID_TOKEN)
self.good_config = False
format = logging.Formatter('%(asctime)s : %(levelname)s, %(message)s', '%a %b %d %H:%M:%S %Z %Y')
self.setFormatter(format)
self.setLevel(logging.DEBUG)
self._thread = SocketAppender()
self._started = False
def emit(self, record):
if not self._started and self.good_config:
dbg("Starting Logentries Asynchronous Socket Appender")
if not self._thread.is_alive():
self._thread.start()
self._started = True
msg = self.format(record).rstrip('\n')
msg = self.token + msg + '\n'
self._thread._queue.put(msg)
def close(self):
logging.Handler.close(self)