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.
108 lines
3.4 KiB
108 lines
3.4 KiB
import fnmatch
|
|
import logging.config
|
|
import os
|
|
import time
|
|
from unittest import TestCase
|
|
|
|
from .mockLogzioListener import listener
|
|
|
|
|
|
def _find(pattern, path):
|
|
result = []
|
|
for root, dirs, files in os.walk(path):
|
|
for name in files:
|
|
if fnmatch.fnmatch(name, pattern):
|
|
result.append(os.path.join(root, name))
|
|
|
|
break # Not descending recursively
|
|
return result
|
|
|
|
|
|
class TestLogzioSender(TestCase):
|
|
def setUp(self):
|
|
self.logzio_listener = listener.MockLogzioListener()
|
|
self.logzio_listener.clear_server_error()
|
|
self.logs_drain_timeout = 1
|
|
|
|
logging_configuration = {
|
|
"version": 1,
|
|
"formatters": {
|
|
"logzio": {
|
|
"format": '{"key": "value"}'
|
|
}
|
|
},
|
|
"handlers": {
|
|
"LogzioHandler": {
|
|
"class": "logzio.handler.LogzioHandler",
|
|
"formatter": "logzio",
|
|
"level": "DEBUG",
|
|
"token": "token",
|
|
'logzio_type': "type",
|
|
'logs_drain_timeout': self.logs_drain_timeout,
|
|
'url': "http://" + self.logzio_listener.get_host() + ":" + str(self.logzio_listener.get_port()),
|
|
'debug': True
|
|
}
|
|
},
|
|
"loggers": {
|
|
"test": {
|
|
"handlers": ["LogzioHandler"],
|
|
"level": "DEBUG"
|
|
}
|
|
}
|
|
}
|
|
|
|
logging.config.dictConfig(logging_configuration)
|
|
self.logger = logging.getLogger('test')
|
|
|
|
for curr_file in _find("logzio-failures-*.txt", "."):
|
|
os.remove(curr_file)
|
|
|
|
def test_simple_log_drain(self):
|
|
log_message = "Test simple log drain"
|
|
self.logger.info(log_message)
|
|
time.sleep(self.logs_drain_timeout * 2)
|
|
self.assertTrue(self.logzio_listener.find_log(log_message))
|
|
|
|
def test_multiple_lines_drain(self):
|
|
logs_num = 50
|
|
for counter in range(0, logs_num):
|
|
self.logger.info("Test " + str(counter))
|
|
time.sleep(self.logs_drain_timeout * 2)
|
|
|
|
for counter in range(0, logs_num):
|
|
self.logger.info("Test " + str(counter))
|
|
time.sleep(self.logs_drain_timeout * 2)
|
|
|
|
self.assertEqual(self.logzio_listener.get_number_of_logs(), logs_num * 2)
|
|
|
|
def test_server_failure(self):
|
|
log_message = "Failing log message"
|
|
self.logzio_listener.set_server_error()
|
|
self.logger.info(log_message)
|
|
|
|
time.sleep(self.logs_drain_timeout * 2)
|
|
|
|
self.assertFalse(self.logzio_listener.find_log(log_message))
|
|
|
|
self.logzio_listener.clear_server_error()
|
|
|
|
time.sleep(self.logs_drain_timeout * 2 * 4) # Longer, because of the retry
|
|
|
|
self.assertTrue(self.logzio_listener.find_log(log_message))
|
|
|
|
def test_local_file_backup(self):
|
|
log_message = "Backup to local filesystem"
|
|
self.logzio_listener.set_server_error()
|
|
self.logger.info(log_message)
|
|
|
|
# Make sure no file is present
|
|
self.assertEqual(len(_find("logzio-failures-*.txt", ".")), 0)
|
|
|
|
time.sleep(2 * 2 * 2 * 2 * 2) # All of the retries
|
|
|
|
failure_files = _find("logzio-failures-*.txt", ".")
|
|
self.assertEqual(len(failure_files), 1)
|
|
|
|
with open(failure_files[0], "r") as f:
|
|
line = f.readline()
|
|
self.assertTrue(log_message in line)
|