X7ROOT File Manager
Current Path:
/opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/lib/commons
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
lvestats
/
lib
/
commons
/
??
..
??
__init__.py
(219 B)
??
__pycache__
??
argparse_utils.py
(11.25 KB)
??
dateutil.py
(5.6 KB)
??
decorators.py
(893 B)
??
func.py
(15.89 KB)
??
htpasswd.py
(2.25 KB)
??
litespeed.py
(6.67 KB)
??
logsetup.py
(4.5 KB)
??
proctitle.py
(2.9 KB)
??
profiler.py
(575 B)
??
progress.py
(1016 B)
??
sentry.py
(6.17 KB)
??
server_status.py
(1.31 KB)
??
sizeutil.py
(2.59 KB)
??
users_manager.py
(2.97 KB)
Editing: sentry.py
# coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import logging # NOQA import os import time import traceback from types import TracebackType # NOQA from typing import Dict, Set, Tuple, Type # NOQA from urllib.error import URLError import raven.processors from raven import Client from raven.transport import ThreadedHTTPTransport import lvestats.version from clcommon import cpapi, get_lve_version from clcommon.utils import get_rhn_systemid_value DSN = "https://9713d1296f804031b058b8f2d789d7ac:8ddacae32d8246cf8b25cf826bf3fc0a@cl.sentry.cloudlinux.com/12" class RemoveConnectString(raven.processors.Processor): def filter_stacktrace(self, data): # type: (Dict) -> None for frame in data.get('frames', []): for variables in frame.get('vars', []): if isinstance(variables, dict): variables.pop('connect_string', None) class LveStatsSentryFilter(object): _record_list_file = "/var/lve/errors_record_list" _last_clean = time.time() _record_list_cache = set() # type: Set[str] @classmethod def clear_record_list(cls): # type: () -> None cls._record_list_cache = set() try: os.unlink(cls._record_list_file) except OSError: pass @classmethod def _clean_muted_records(cls): if cls._last_clean < time.time() - 43200: # more than 12 hours cls.clear_record_list() cls._last_clean = time.time() def in_record_list(self, record_fingerprint): # type: (str) -> bool try: with open(self._record_list_file, 'r', encoding='utf-8') as f: result = [line.rstrip('\n') for line in f] except IOError: return False else: return record_fingerprint in result def append_to_record_list(self, record_fingerprint): # type: (str) -> None self._record_list_cache.add(record_fingerprint) try: with open(self._record_list_file, "a", encoding='utf-8') as f: f.write(record_fingerprint + '\n') except IOError: pass def filter(self, record): # type: (logging.LogRecord) -> bool return self.check_fingerprint_absent( record.name, record.lineno, str(record.msg)[:25], record.exc_info[0].__name__ if record.exc_info else "") def check_fingerprint_absent(self, name, lineno, message, exc_name): # type: (str, int, str, str) -> bool if not os.environ.get('LVESTATSWITHOUTSENTR1Y'): record_fingerprint = repr(f"{name}.{lineno}.{message}.{exc_name}") self._clean_muted_records() if record_fingerprint in self._record_list_cache: return False elif self.in_record_list(record_fingerprint): self._record_list_cache.add(record_fingerprint) return False else: self.append_to_record_list(record_fingerprint) return True else: return False class SafeThreadedHTTPTransport(ThreadedHTTPTransport): def send_sync(self, url, data, headers, success_cb, failure_cb): try: super().send_sync(url, data, headers, success_cb, failure_cb) except URLError: # We hide errors while sending message to Sentry in varied cases: # problems with network, Sentry server is down, incorrect firewall's rules, etc pass class LveStatsSentryClient(Client): _filter = LveStatsSentryFilter() def is_new_exception(self, exc_info): # type: (Tuple[Type, Exception, TracebackType]) -> bool type_, _, tb = exc_info filepath, line_no, _, _ = traceback.extract_tb(tb)[-1] name = os.path.basename(filepath) # 'message' param is not used here # line number and file must be enough return self._filter.check_fingerprint_absent( name, line_no, '', type_.__name__) def should_capture(self, exc_info): # type: (Tuple[Type, Exception, TracebackType]) -> bool return super().should_capture(exc_info) and self.is_new_exception(exc_info) def init_sentry_client(): client = LveStatsSentryClient( DSN, transport=SafeThreadedHTTPTransport, ) # uncomment for lower issues count # client.sample_rate = 0.5 client.auto_log_stacks = True client.string_max_length = 500 client.user_context({"id": get_rhn_systemid_value("system_id")}) client.tags["Project"] = "lve-stats" try: client.tags["Email"] = cpapi.get_admin_email() except cpapi.NotSupported: client.tags["Email"] = "unknown" client.tags["Control Panel Name"] = cpapi.CP_NAME lve_version, _ = get_lve_version() client.tags["LVE"] = 'unknown' if lve_version is None else lve_version client.tags["Cloud Linux version"] = get_rhn_systemid_value("os_release") client.tags["Architecture"] = get_rhn_systemid_value("architecture") try: version = lvestats.version.VERSION.split(".el") release = version[0] is_developer = len(version[1]) > 2 or os.path.exists( f"/root/rpmbuild/RPMS/noarch/lve-stats-{lvestats.version.VERSION}.noarch.rpm") except IndexError: release = lvestats.version.VERSION is_developer = True client.tags["Developer"] = is_developer client.release = release client.exclude_paths = ['sentry', 'raven'] client.ignore_exceptions.add(SystemExit) client.ignore_exceptions.add(KeyboardInterrupt) client.ignore_exceptions.add("lvestats.eventloop.plugin_executors.PluginExecutionException") test_sentry_message(client=client) return client def test_sentry_message(client): if os.environ.get('LVESTATSSENTRYDONTSEND'): def dont_send(data, headers, success_cb, failure_cb): time.sleep(0.5) transport = client.remote.get_transport() # type: raven.transport.threaded.ThreadedHTTPTransport transport.send_sync = dont_send client.captureMessage('Something went fundamentally wrong')
Upload File
Create Folder