Logging handler to send logs to your OpenSearch cluster with bulk SSL. Forked from https://github.com/logzio/logzio-python-handler
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
182 lines
5.6 KiB
182 lines
5.6 KiB
import logging
|
|
import logging.handlers
|
|
import requests
|
|
import traceback
|
|
import datetime
|
|
import json
|
|
|
|
from threading import Event, Thread, Condition, Lock, enumerate
|
|
from time import sleep
|
|
|
|
|
|
class LogzioHandler(logging.Handler):
|
|
|
|
# Hold all logs buffered
|
|
logs = []
|
|
|
|
# Event for locking buffer additions while draining
|
|
buffer_event = Event()
|
|
|
|
# Condition to count log messages
|
|
logs_counter_condition = Condition()
|
|
|
|
# Lock to only drain logs once
|
|
drain_lock = Lock()
|
|
|
|
def __init__(self, token, logs_drain_count=100, logs_drain_timeout=10,
|
|
logzio_type="python", url="https://listener.logz.io:8071/"):
|
|
|
|
if token is "":
|
|
raise Exception("Logz.io Token must be provided")
|
|
|
|
logging.Handler.__init__(self)
|
|
self.logs_drain_count = logs_drain_count
|
|
self.logs_drain_timeout = logs_drain_timeout
|
|
self.logzio_type = logzio_type
|
|
self.url = "{0}?token={1}".format(url, token)
|
|
|
|
self.is_main_thread_active = lambda: any((i.name == "MainThread") and i.is_alive() for i in enumerate())
|
|
|
|
self.buffer_event.set()
|
|
|
|
# Create threads
|
|
timeout_thread = Thread(target=self.wait_to_timeout_and_drain)
|
|
counter_thread = Thread(target=self.count_logs_and_drain)
|
|
|
|
# And start them
|
|
timeout_thread.start()
|
|
counter_thread.start()
|
|
|
|
def wait_to_timeout_and_drain(self):
|
|
|
|
while True:
|
|
sleep(self.logs_drain_timeout)
|
|
if len(self.logs) > 0:
|
|
self.drain_messages()
|
|
|
|
if not self.is_main_thread_active():
|
|
# Signal the counter thread so it would exit as well
|
|
try:
|
|
self.logs_counter_condition.acquire()
|
|
self.logs_counter_condition.notify()
|
|
finally:
|
|
self.logs_counter_condition.release()
|
|
break
|
|
|
|
def count_logs_and_drain(self):
|
|
try:
|
|
# Acquire the condition
|
|
self.logs_counter_condition.acquire()
|
|
|
|
# Running indefinite
|
|
while True:
|
|
|
|
# Waiting for new log lines to come
|
|
self.logs_counter_condition.wait()
|
|
|
|
if not self.is_main_thread_active():
|
|
break
|
|
|
|
# Do we have enough logs to drain?
|
|
if len(self.logs) >= self.logs_drain_count:
|
|
self.drain_messages()
|
|
|
|
finally:
|
|
self.logs_counter_condition.release()
|
|
|
|
def add_to_buffer(self, message):
|
|
|
|
# Check if we are currently draining buffer so we wont loose logs
|
|
self.buffer_event.wait()
|
|
|
|
try:
|
|
# Acquire the condition
|
|
self.logs_counter_condition.acquire()
|
|
self.logs.append(json.dumps(message))
|
|
|
|
# Notify watcher for a new log coming in
|
|
self.logs_counter_condition.notify()
|
|
|
|
finally:
|
|
# Release the condition
|
|
self.logs_counter_condition.release()
|
|
|
|
def handle_exceptions(self, message):
|
|
if message.exc_info:
|
|
return '\n'.join(traceback.format_exception(*message.exc_info))
|
|
else:
|
|
return message.getMessage()
|
|
|
|
def format_message(self, message):
|
|
|
|
message_field = self.handle_exceptions(message)
|
|
now = datetime.datetime.utcnow()
|
|
timestamp = now.strftime("%Y-%m-%dT%H:%M:%S") + ".%03d" % (now.microsecond / 1000) + "Z"
|
|
|
|
return_json = {
|
|
"logger": message.name,
|
|
"line_number": message.lineno,
|
|
"path_name": message.pathname,
|
|
"log_level": message.levelname,
|
|
"message": message_field,
|
|
"type": self.logzio_type,
|
|
"@timestamp": timestamp
|
|
}
|
|
|
|
return return_json
|
|
|
|
def backup_logs(self, logs):
|
|
timestamp = datetime.datetime.now().strftime("%d%m%Y-%H%M%S")
|
|
print("Backing up your logs to logzio-failures-{0}.txt".format(timestamp))
|
|
with open("logzio-failures-{0}.txt".format(timestamp), "a") as f:
|
|
f.writelines('\n'.join(logs))
|
|
|
|
def drain_messages(self):
|
|
try:
|
|
self.buffer_event.clear()
|
|
self.drain_lock.acquire()
|
|
|
|
# Copy buffer
|
|
temp_logs = list(self.logs)
|
|
self.logs = []
|
|
|
|
# Release the event
|
|
self.buffer_event.set()
|
|
|
|
# Not configurable from the outside
|
|
sleep_between_retries = 2000
|
|
number_of_retries = 4
|
|
|
|
success_in_send = False
|
|
headers = {"Content-type": "text/plain"}
|
|
|
|
for current_try in range(number_of_retries):
|
|
response = requests.post(self.url, headers=headers, data='\n'.join(temp_logs))
|
|
|
|
if response.status_code != 200: # 429 400, on 400 print stdout
|
|
if response.status_code == 400:
|
|
|
|
print("Got unexpected 400 code from logz.io, response: {0}".format(response.text))
|
|
self.backup_logs(temp_logs)
|
|
|
|
if response.status_code == 401:
|
|
print("You are not authorized with logz.io! dropping..")
|
|
break
|
|
|
|
sleep(sleep_between_retries)
|
|
sleep_between_retries *= 2
|
|
else:
|
|
success_in_send = True
|
|
break
|
|
|
|
if not success_in_send:
|
|
|
|
# Write to file
|
|
self.backup_logs(temp_logs)
|
|
|
|
finally:
|
|
self.buffer_event.set()
|
|
self.drain_lock.release()
|
|
|
|
def emit(self, record):
|
|
self.add_to_buffer(self.format_message(record))
|