/opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos
Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos/logsetup.py (4142B)
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
from __future__ import absolute_import
import logging
import os
import pwd
from logging import StreamHandler
from logging.handlers import RotatingFileHandler
from clwpos.constants import PROJECT_NAME, SENTRY_DSN, PKG_VERSION
from clsentry import init_sentry_client
ADMIN_LOGFILE_PATH = "/var/log/clwpos/main.log"
USER_WPOS_DIR = ".clwpos"
USER_LOGFILE_PATH = os.path.join("{homedir}", USER_WPOS_DIR, "main.log")
def setup_logging(
caller_name,
console_level=logging.CRITICAL,
file_level=logging.INFO,
logfile_path=ADMIN_LOGFILE_PATH,
):
"""
Sets up logger and return it to the caller
:param str caller_name:
:param console_level:
:param file_level:
:return logging.Logger: log, initialized
"""
euid = os.geteuid()
if euid != 0:
pw = pwd.getpwuid(euid)
logfile_path = USER_LOGFILE_PATH.format(homedir=pw.pw_dir)
from clwpos.utils import create_clwpos_dir_if_not_exists
create_clwpos_dir_if_not_exists(pw)
logging.root.setLevel(logging.DEBUG)
logger = logging.getLogger(caller_name)
has_file_log = any(
[True for handler in logger.handlers if isinstance(handler, RotatingFileHandler)]
)
has_console_log = any(
[True for handler in logger.handlers if isinstance(handler, StreamHandler)]
)
if not has_console_log:
logger.addHandler(get_console_log_handler(console_level))
if not has_file_log:
# try-except is valid only for unittests case
# we cannot simply mock this method, because setup_logging is used upon imports
# in a lot of places
try:
logger.addHandler(get_file_log_handler(file_level, logfile_path))
except Exception:
pass
return logger
def get_package_version(pkg_version=None):
"""
Strips platform-dependent part of package version.
"""
pkg_version = pkg_version or PKG_VERSION
try:
version = pkg_version.split(".el")
release = version[0]
is_developer = not version[1].endswith('cloudlinux')
except IndexError:
release = pkg_version
is_developer = True
return release, is_developer
def init_wpos_sentry_safely(logger=None):
"""
Inits sentry safely, do not break all functionality if init failed
"""
version, is_development = get_package_version()
try:
sentry = init_sentry_client(PROJECT_NAME, release=version, dsn=SENTRY_DSN)
sentry.environment = 'Development' if is_development else 'Release'
sentry.tags['full_version'] = PKG_VERSION
except Exception as e:
if logger:
logger.warning('Error during sentry init: %s', str(e))
def get_file_log_handler(file_level, logfile_path):
if not os.path.lexists(logfile_path):
umask_saved = os.umask(0o77)
try:
if not os.path.exists(os.path.dirname(logfile_path)):
os.makedirs(os.path.dirname(logfile_path), exist_ok=True)
open(logfile_path, "w").close()
finally:
os.umask(umask_saved)
os.chmod(logfile_path, 0o600)
# we save only 5mb of logs per user
file_log = RotatingFileHandler(logfile_path, maxBytes=1024 * 1024, backupCount=5)
formatter = logging.Formatter(
fmt="%(asctime)s: (%(name)s) [%(levelname)s] %(message)s"
)
file_log.setFormatter(formatter)
file_log.setLevel(file_level)
return file_log
def get_console_log_handler(console_level):
console_log = logging.StreamHandler()
formatter_console = logging.Formatter(
fmt="%(asctime)s: (%(name)s) [%(levelname)s] %(message)s"
)
console_log.setFormatter(formatter_console)
console_log.setLevel(console_level)
return console_log
class NullLogger(logging.Logger):
def __init__(self):
super(NullLogger, self).__init__('null_logger')
nh = logging.NullHandler()
self.addHandler(nh)