-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathlogger.py
More file actions
120 lines (96 loc) · 4.1 KB
/
Copy pathlogger.py
File metadata and controls
120 lines (96 loc) · 4.1 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
import os
import random
import datetime
import traceback
from lambda_powertools.config import Config
from lambda_powertools.utils import serialize_log
class Logger:
def __init__(self):
self.LEVELS = {
"DEBUG": 20,
"INFO": 30,
"WARN": 40,
"ERROR": 50,
}
self._level = None
self._context = None
self.reset()
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(Logger, cls).__new__(cls)
return cls.instance
def reset(self):
self._level = self.LEVELS[os.environ.get("LOG_LEVEL", "INFO")]
self._context = {}
def set_level(self, level_name):
self._level = self.LEVELS.get(level_name, self._level)
def capture(self, env=None, event=None, context=None):
function_name = None
function_memory = None
function_version = None
aws_request_id = None
aws_apigw_request_id = None
aws_cf_request_id = None
if env is not None:
function_name = env.get("AWS_LAMBDA_FUNCTION_NAME", None)
function_memory = int(env.get("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", 0))
function_version = env.get("AWS_LAMBDA_FUNCTION_VERSION", None)
if function_version == "$LATEST":
function_version = None
if context is not None:
aws_request_id = context.aws_request_id
if event is not None:
if event.get("requestContext") and event.get("requestContext").get("apiId"):
aws_apigw_request_id = event.get("requestContext").get("requestId", None)
if event.get("headers"):
aws_cf_request_id = event.get("headers").get("x-amz-cf-id", None)
self._context = {}
if function_name is not None:
self._context["function_name"] = function_name
if function_version is not None:
self._context["function_version"] = function_version
if function_memory is not None:
self._context["function_memory"] = function_memory
if aws_request_id is not None:
self._context["aws_request_id"] = aws_request_id
if aws_apigw_request_id is not None:
self._context["aws_apigw_request_id"] = aws_apigw_request_id
if aws_cf_request_id is not None:
self._context["aws_cf_request_id"] = aws_cf_request_id
if event is not None and event.get("headers") and event["headers"].get("x-debug") == "true":
self.set_level("DEBUG")
def should_throttle(self, level_name):
# If the throttle is disabled or the logger is configured in debug level we don't throttle anything
if not Config.is_log_throttle_enabled() or self._level == self.LEVELS["DEBUG"]:
return False
return random.random() > (Config.get_log_throttle(level_name) or 1)
def log(self, level_name, message, context=None, error=None):
if self.LEVELS[level_name] < self._level or self.should_throttle(level_name):
return
if context is None:
context = {}
if isinstance(error, Exception):
"""
If the error is provided we use the context to pass custom object
"""
context["error_message"] = str(error)
context["error_stack"] = "\n".join(traceback.format_exc().splitlines())
log_str = serialize_log(
{
**self._context,
**(context or {}),
"time": int(datetime.datetime.now().timestamp() * 1000),
"loglevel": level_name,
"message": message,
}
)
print(f"{log_str}\n")
def debug(self, message, context=None, error=None):
self.log("DEBUG", message, context, error)
def info(self, message, context=None, error=None):
self.log("INFO", message, context, error)
def warn(self, message, context=None, error=None):
self.log("WARN", message, context, error)
def error(self, message, context=None, error=None):
self.log("ERROR", message, context, error)
logger = Logger()