Browse Source

extract the queue draining action out of the loop. this way it can be manually called externally

opensearch
Oren Mazor 7 years ago
parent
commit
df76f91e04
3 changed files with 52 additions and 47 deletions
  1. +0
    -1
      .cache/v/cache/lastfailed
  2. +3
    -0
      logzio/handler.py
  3. +49
    -46
      logzio/sender.py

+ 0
- 1
.cache/v/cache/lastfailed View File

@ -1 +0,0 @@
{}

+ 3
- 0
logzio/handler.py View File

@ -44,6 +44,9 @@ class LogzioHandler(logging.Handler):
return extra_fields return extra_fields
def flush(self):
self.logzio_sender._flush_queue()
def format(self, record): def format(self, record):
message = super(LogzioHandler, self).format(record) message = super(LogzioHandler, self).format(record)
try: try:


+ 49
- 46
logzio/sender.py View File

@ -57,52 +57,7 @@ class LogzioSender:
last_try = True last_try = True
try: try:
# Sending logs until queue is empty
while not self.queue.empty():
logs_list = self._get_messages_up_to_max_allowed_size()
self._debug("Starting to drain " + str(len(logs_list)) + " logs to Logz.io")
# Not configurable from the outside
sleep_between_retries = 2
number_of_retries = 4
should_backup_to_disk = True
headers = {"Content-type": "text/plain"}
for current_try in range(number_of_retries):
should_retry = False
try:
response = requests.post(self.url, headers=headers, data='\n'.join(logs_list))
if response.status_code != 200:
if response.status_code == 400:
print("Got 400 code from Logz.io. This means that some of your logs are too big, or badly formatted. response: {0}".format(response.text))
should_backup_to_disk = False
break
if response.status_code == 401:
print("You are not authorized with Logz.io! Token OK? dropping logs...")
should_backup_to_disk = False
break
else:
print("Got {} while sending logs to Logz.io, Try ({}/{}). Response: {}".format(response.status_code, current_try + 1, number_of_retries, response.text))
should_retry = True
else:
self._debug("Successfully sent bulk of " + str(len(logs_list)) + " logs to Logz.io!")
should_backup_to_disk = False
break
except Exception as e:
print("Got exception while sending logs to Logz.io, Try ({}/{}). Message: {}".format(current_try + 1, number_of_retries, e))
should_retry = True
if should_retry:
sleep(sleep_between_retries)
sleep_between_retries *= 2
if should_backup_to_disk:
# Write to file
print("Could not send logs to Logz.io after " + str(number_of_retries) + " tries, backing up to local file system.")
backup_logs(logs_list)
self._flush_the_queue()
except Exception as e: except Exception as e:
self._debug("Unexpected exception while draining queue to Logz.io, swallowing. Exception: " + str(e)) self._debug("Unexpected exception while draining queue to Logz.io, swallowing. Exception: " + str(e))
@ -110,6 +65,54 @@ class LogzioSender:
if not last_try: if not last_try:
sleep(self.logs_drain_timeout) sleep(self.logs_drain_timeout)
def _flush_the_queue(self):
# Sending logs until queue is empty
while not self.queue.empty():
logs_list = self._get_messages_up_to_max_allowed_size()
self._debug("Starting to drain " + str(len(logs_list)) + " logs to Logz.io")
# Not configurable from the outside
sleep_between_retries = 2
number_of_retries = 4
should_backup_to_disk = True
headers = {"Content-type": "text/plain"}
for current_try in range(number_of_retries):
should_retry = False
try:
response = requests.post(self.url, headers=headers, data='\n'.join(logs_list))
if response.status_code != 200:
if response.status_code == 400:
print("Got 400 code from Logz.io. This means that some of your logs are too big, or badly formatted. response: {0}".format(response.text))
should_backup_to_disk = False
break
if response.status_code == 401:
print("You are not authorized with Logz.io! Token OK? dropping logs...")
should_backup_to_disk = False
break
else:
print("Got {} while sending logs to Logz.io, Try ({}/{}). Response: {}".format(response.status_code, current_try + 1, number_of_retries, response.text))
should_retry = True
else:
self._debug("Successfully sent bulk of " + str(len(logs_list)) + " logs to Logz.io!")
should_backup_to_disk = False
break
except Exception as e:
print("Got exception while sending logs to Logz.io, Try ({}/{}). Message: {}".format(current_try + 1, number_of_retries, e))
should_retry = True
if should_retry:
sleep(sleep_between_retries)
sleep_between_retries *= 2
if should_backup_to_disk:
# Write to file
print("Could not send logs to Logz.io after " + str(number_of_retries) + " tries, backing up to local file system.")
backup_logs(logs_list)
def _get_messages_up_to_max_allowed_size(self): def _get_messages_up_to_max_allowed_size(self):
logs_list = [] logs_list = []
current_size = 0 current_size = 0


Loading…
Cancel
Save