Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py	0 B	2026-02-10 14:37
📁 __pycache__	-	2026-03-21 15:47
📄 autotracer.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains Autotracer class """ __package__ = 'ssa.modules' import json import logging import os from collections import defaultdict from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from fnmatch import fnmatchcase from typing import Iterator, Optional, NamedTuple, List, Iterable from urllib.parse import urlparse import numpy as np from ssa.db import setup_database from ssa.internal.exceptions import SSAError from ssa.modules.common import Common from ssa.modules.storage import ( iter_domains_data, iter_urls_data, get_url_durations ) from ssa.autotracing import status, disabled_users, misconfiguration_checks from ssa.configuration import load_tunables from ssa.configuration.schemes import autotracing_tunables_schema from ssa.internal.constants import autotracing_stats_file from ssa.internal.utils import ( is_xray_version_supported, is_kernel_version_supported, sentry_init ) xray_import_error = None try: import xray.shared_library as xray_lib except ImportError as e: xray_import_error = e xray_lib = None @dataclass class URL: """URL data container""" uri: str # full URL including domain and URI avg_duration: int # average duration during the day max_duration: int # maximum duration during the day total_slow_reqs: int # total SLOW request count during the day total_reqs: int # total request count during the day belongs_to: str # domain belongs_to_user: str # domain owner @dataclass class RulesRejects: """Rejects counters container""" non_wp: int = 0 # number of URLs, skipped due to non-WP site throttled: int = 0 # number of URLs, skipped due to throttling density: int = 0 # number of URLs, skipped due to low density slowness: int = 0 # number of URLs, skipped because they are considered fast max_slowness: int = 0 # number of URLs, skipped because they are considered too slow disabled: int = 0 # number of URLs, skipped due to disabled autotracing nginx: int = 0 # number of URLs, skipped due to nginx caching frequency: int = 0 # number of URLs, skipped due to min_retracing_interval server_limit: int = 0 # number of URLs, skipped due to per-server limit domain_limit: int = 0 # number of URLs, skipped due to per-domain limit no_domain: int = 0 # number of URLs, skipped due to unavailable domain info (e.g. no domain) @dataclass class Stats: """AutoTracer statistics container""" rules_version: str # version of rules applied from autotracing.json urls_processed: int = 0 # number of URLs, processed within the latest iteration urls_selected: int = 0 # number of URLs, selected to launch autotracing tasks rejects: RulesRejects = field(default_factory=RulesRejects) # rejects counters class URLS(NamedTuple): """ Representation of an URL """ domain_name: str uri_path: str def url_split(url: str) -> URLS: """ Split URL into domain_name and uripath including query string :param url: URL of format protocol://domain/path;parameters?query#fragment :return: namedtuple URL(domain_name, uripath) """ fragments = urlparse(url) qs = f'?{fragments.query}' if fragments.query else '' uri = f'{fragments.path}{qs}' if fragments.path else '/' _no_www_netloc = fragments.netloc.replace('www.', '') _no_port_netloc = _no_www_netloc.split(':')[0] return URLS(_no_port_netloc, uri) class AutoTracer(Common): """ SSA autotracing module implementation. """ def __init__(self, engine=None): super().__init__() self.logger = logging.getLogger('auto_tracer') self.logger.info('AutoTracer enabled: %s', __package__) self.disabled_users = list() self.tasks_list = list() self.stats = None # initial value for an iteration statistics self.engine = engine if engine else setup_database() def __call__(self): try: misconfiguration_checks() except SSAError as exc: self.logger.info('AutoTracer skipped: %s', exc.reason) return if xray_lib is None: # to send sentry event only if x-ray really not installed if os.path.exists('/usr/sbin/cloudlinux-xray-agent'): self.logger.error('AutoTracer skipped: X-ray module import error: %s', str(xray_import_error)) else: self.logger.info('AutoTracer skipped: X-Ray not installed') return start_tool = xray_lib.start_autotracing if start_tool is not None: self.logger.info('AutoTracer started') # reload config super().__init__() self.logger.debug('AutoTracer loaded config: %s', self.config) # start gather own statistics for a current iteration self.stats = Stats(self.rules_version) for url in self.urls_scheduled(): self.logger.info("Starting auto task for %s", url.uri) self.stats.urls_selected += 1 try: start_tool(url=url.uri, tracing_count=self.request_number) except Exception as e: # most likely XRayError self.logger.error('Failed to start task: %s', str(e)) self.save_iteration_stats() @staticmethod def load_conf() -> dict: """ Load configuration """ return load_tunables('autotracing.json', autotracing_tunables_schema) @staticmethod def gets_domaininfo(domain_name: str) -> Optional[object]: """ Gets domain info for the specified domain """ if xray_lib is None: return None return xray_lib.domain_info(domain_name) def nginx_is_enabled(self, user_name: str) -> bool: """ Says if nginx is enabled for a specific user """ if self.skip_nginx: if xray_lib is None: return False return xray_lib.NginxUserCache(user_name).is_enabled return False def per_server_limit_recalculated(self) -> int: """Recalculate limit per server taking into account running tasks""" running_count = len([task for task in self.tasks_list if task['status'] == 'running' and task[ 'user'] == 'autotracing']) self.logger.debug('Number of running autotracing tasks %s', running_count) return self.per_server_limit - running_count def excludes_old_tasks(self, full_list: Optional[list] = None) -> list: """ Excludes tasks older than N days from the general list of tasks """ if full_list is None: full_list = xray_lib.tasks_list() self.logger.debug('Task list loaded %s', full_list) n_days_ago = datetime.now() - timedelta( days=self.min_retracing_interval) return [task for task in full_list if (task.get('createtime') if task.get('createtime') is not None else 0) > int(n_days_ago.timestamp())] def exclude_thesame_urls(self, current_url: str) -> bool: """ Excludes url from the list if it completely matches the current url or if domain names match and "" follows the domain name in the list """ # c - current parsed url c = url_split(current_url) for task_data in self.tasks_list: # t - task parsed url t = url_split(task_data['url']) direct_match = c.domain_name == t.domain_name and c.uri_path == t.uri_path wildcard_match = fnmatchcase( c.domain_name, t.domain_name) and fnmatchcase(c.uri_path, t.uri_path) if direct_match or wildcard_match: self.logger.debug( 'Skipped: URL %s was traced recently. Matched by %s', current_url, task_data['url']) return True return False def pass_by_density(self, url_total_reqs: list, domain_total_reqs: list) -> bool: """Check that URL density passes given threshold""" if self.density: url_density = np.amin( np.corrcoef(url_total_reqs, domain_total_reqs)) self.logger.debug('Calculated density %s', url_density) return url_density > self.density_threshold return True def pass_by_slowness_percentile(self, url_durations: Iterable[int]) -> bool: """ The measure of "slowness" for URL is: at least N% of its requests take more than X seconds. N% -- self.slow_duration_percentage X -- self.slow_duration_threshold """ reversed_percentile = 100 - self.slow_duration_percentage reversed_percentile_value = np.percentile(url_durations, reversed_percentile) self.logger.debug('Calculated %sth percentile %s for min duration', reversed_percentile, reversed_percentile_value) return reversed_percentile_value >= self.slow_duration_threshold def pass_by_max_slowness_percentile(self, url_durations: Iterable[int]) -> bool: """ The opposite to pass_by_slowness_percentile method. The measure of "much slowness" for URL is: at least N% of its requests must take less than X seconds. N% -- self.max_slow_duration_percentage X -- self.max_slow_duration_threshold """ percentile_value = np.percentile(url_durations, self.max_slow_duration_percentage) self.logger.debug('Calculated %sth percentile %s for max duration', self.max_slow_duration_percentage, percentile_value) return percentile_value <= self.max_slow_duration_threshold def pass_by_allowed_throttling(self, url_throttled_reqs: Optional[list], url_total_reqs: list) -> bool: """ Check that percent of throttled requests per URL passes given threshold """ if url_throttled_reqs is None: # skip URL with unavailable throttling info return False throttled_percent = (sum(url_throttled_reqs)/sum(url_total_reqs))100 self.logger.debug('Calculated throttled percent %s', throttled_percent) return throttled_percent <= self.allowed_throttling_percentage def pass_by_engine(self, wp_status: Optional[bool]) -> bool: """ Check that URLs of a particular domain should be analyzed. For now we skip non-wordpress sites """ if self.only_wp: # turn unavailable wp_status, aka None, into False return bool(wp_status) return True def urls_computationally_filtered(self) -> Iterator[URL]: """ Select all URLs suitable for auto tracing by very basic rules: - WP site - suitable throttling - suitable density - measure of "slow" URL ORDER OF RULES MUST NOT BE CHANGED: IT AFFECTS STATISTICS COUNTERS """ for domain_data in iter_domains_data(self.engine): domain_owner = None if not self.pass_by_engine(domain_data.is_a_wordpress_domain): # rule ordering #1: skip non-wordpress sites # corresponds to non_wp RulesRejects counter self.logger.debug('Skipped by engine: non-wordpress') # all URLs of non-WP site are skipped then, # thus counters (reject abd total) are to be increased # on number of these URLs ignoring non-URL data fields skipped_count = domain_data.urls_number self.stats.rejects.non_wp += skipped_count self.stats.urls_processed += skipped_count continue domain_url_durations = dict(get_url_durations( self.engine, domain_data.domain_name)) for url, data in iter_urls_data(self.engine, domain_data.domain_name, list(domain_url_durations.keys())): if url in self.non_url_fields: # skip entry of domain requests counter continue self.logger.debug('Processing URL %s', url) self.stats.urls_processed += 1 if not self.pass_by_allowed_throttling( data.get('url_throttled_reqs'), data['url_total_reqs']): # rule ordering #2: percent of throttled requests # corresponds to throttled RulesRejects counter self.logger.debug('Skipped by throttled percent') self.stats.rejects.throttled += 1 continue if not self.pass_by_density(data['url_total_reqs'], domain_data.domain_total_reqs): # rule ordering #3: density threshold # corresponds to density RulesRejects counter self.logger.debug('Skipped by density') self.stats.rejects.density += 1 continue durations = domain_url_durations[url] if not self.pass_by_slowness_percentile(durations): # rule ordering #4: slowness assessment # corresponds to slowness RulesRejects counter self.logger.debug('Skipped by slowness percentile') self.stats.rejects.slowness += 1 continue if not self.pass_by_max_slowness_percentile(durations): # rule ordering #5: maximum allowed slowness assessment # corresponds to max_slowness RulesRejects counter self.logger.debug('Skipped by max slowness percentile') self.stats.rejects.max_slowness += 1 continue # the URL has passed all computational checks # from here we already need username if domain_owner is None: domain_info = self.gets_domaininfo(domain_data.domain_name) if domain_info is None: # this generally indicates "Domain does not exist" # error, we should skip such URLs self.logger.debug('Skipped by unavailable domain info') self.stats.rejects.no_domain += 1 continue else: domain_owner = domain_info.user yield URL(url, avg_duration=int(np.mean(durations)), total_slow_reqs=sum(data['url_slow_reqs']), total_reqs=sum(data['url_total_reqs']), max_duration=max(durations), belongs_to=domain_data.domain_name, belongs_to_user=domain_owner) def urls_selected(self, stats_collected: Optional[dict] = None) -> Iterator[URL]: """ From selected by computed thresholds URLs take those for which: - autotracing enabled - nginx disabled ORDER OF RULES MUST NOT BE CHANGED: IT AFFECTS STATISTICS COUNTERS """ # fill disabled users list, initially empty self.fill_in_disabled_users() for url in self.urls_computationally_filtered(): if url.belongs_to_user in self.disabled_users: # rule ordering #6: skip users for whom autotracing is disabled # corresponds to disabled RulesRejects counter self.logger.debug('Skipped: autotracing is disabled for %s', url.belongs_to_user) self.stats.rejects.disabled += 1 continue if self.nginx_is_enabled(url.belongs_to_user): # rule ordering #7: skip url the owner of which uses nginx # corresponds to nginx RulesRejects counter self.logger.debug('Skipped: nginx is enabled for %s', url.belongs_to_user) self.stats.rejects.nginx += 1 continue yield url def urls_scheduled(self) -> Iterator[URL]: """ Schedule autotracing by sorted list taking into account the limits: - no same task for 10 days - limit per server - limit per domain ORDER OF RULES MUST NOT BE CHANGED: IT AFFECTS STATISTICS COUNTERS Return resulting list of URLs scheduled for auto tracing """ # initial limit counters general_tasks_counter = 0 tasks_counter_per_domain = defaultdict(int) # sort selected URLs sorted_urls = self.urls_sorted() self.logger.debug('Sorted scheduled list %s', sorted_urls) # get task list and filter out all old tasks in one turn self.tasks_list = self.excludes_old_tasks() per_server_smart_limit = self.per_server_limit_recalculated() for url in sorted_urls: if self.exclude_thesame_urls(url.uri): # rule ordering #8: limit of days for url (min_retracing_interval) # corresponds to frequency RulesRejects counter self.stats.rejects.frequency += 1 continue if general_tasks_counter < per_server_smart_limit: if tasks_counter_per_domain[url.belongs_to] < self.per_domain_limit: general_tasks_counter += 1 tasks_counter_per_domain[url.belongs_to] += 1 yield url else: # rule ordering #9: domain limit is hit # corresponds to domain_limit RulesRejects counter self.logger.debug('Skipped URL %s by domain limit (%s)', url.uri, self.per_domain_limit) self.stats.rejects.domain_limit += 1 else: # rule ordering #10: server limit is hit # corresponds to server_limit RulesRejects counter self.logger.debug('Skipped URL %s by server limit (%s)', url.uri, per_server_smart_limit) self.stats.rejects.server_limit += 1 def urls_sorted(self) -> List[URL]: """ Sort URLs by total number of requests first and by average duration second """ first_serie = sorted(list(self.urls_selected()), key=lambda u: u.avg_duration, reverse=True) return sorted(first_serie, key=lambda u: u.total_reqs, reverse=True) def fill_in_disabled_users(self) -> None: """ Fill internal list of disabled users """ self.disabled_users = disabled_users() def save_iteration_stats(self) -> None: """ Save collected statistics for current iteration to file """ if self.stats is not None: try: with open(autotracing_stats_file, 'w') as stats_file: json.dump(asdict(self.stats), stats_file) except OSError as e: self.logger.warning( 'Unable to save iteration stats to file: %s', str(e)) def load_iteration_stats(self) -> dict: """ Load statistics for latest iteration from file """ try: with open(autotracing_stats_file) as stats_file: _data = json.load(stats_file) stat_data = Stats(_data['rules_version'], _data['urls_processed'], _data['urls_selected'], RulesRejects(**_data['rejects'])) except (OSError, json.JSONDecodeError, KeyError): stat_data = Stats(self.rules_version) return asdict(stat_data) def get_stats(self) -> dict: """""" # load saved stats stats_loaded = self.load_iteration_stats() stats_loaded.update(dict( status=status()[0], disabled_users_quantity=len(disabled_users()) )) return stats_loaded if __name__ == "__main__": sentry_init() logging.basicConfig(filename='auto_tracer_standalone.log', level=logging.DEBUG) t = AutoTracer() t()	20.75 KB	2026-02-10 14:37
📄 common.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains a base class including methods, common for SSA Request Processor and SSA Decision Maker """ import fnmatch from typing import Tuple from datetime import date from ..configuration import load_configuration class Common: """ Base SSA module class """ def __init__(self): self.config = self.load_conf() def __getattr__(self, item): """ All undefined attributes, e.g. - domains_number - urls_number - correlation_coefficient - correlation - request_number - time are queried in config. Others would be empty """ conf = getattr(self, 'config') return conf.get(item, '') @property def non_url_fields(self) -> set: return {'domain_total_reqs', 'is_a_wordpress_domain'} @property def configured_ignores(self) -> list: """ Render configured ignore list """ elements = self.ignore_list.split(',') return [i.strip() for i in elements if i.strip()] def is_ignored(self, _url: str) -> bool: """ Check if given URL matches ignore list """ for pattern in self.configured_ignores: if '' in pattern: # set on the first position if it not set to allow filter without scheme (e.g. domain.com) if not pattern.startswith(''): pattern = f"{pattern}" if fnmatch.fnmatch(_url, pattern): return True else: if pattern in _url: return True return False @staticmethod def load_conf() -> dict: """ Load configuration """ return load_configuration()	1.97 KB	2026-02-10 14:37
📄 decision_maker.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains DecisionMaker class """ __package__ = 'ssa.modules' import json import logging import os from os.path import isfile import numpy as np from ssa.db import setup_database from .common import Common from .storage import ( iter_domains_data, iter_urls_data, get_url_durations ) from ..configuration import load_tunables from ..configuration.schemes import ssa_tunables_schema from ..internal.constants import report_path from ..internal.utils import previous_day_date, sentry_init class DecisionMaker(Common): """ SSA Decision maker implementation. """ def __init__(self, engine=None): super().__init__() self.logger = logging.getLogger('decision_maker') self.logger.info('DecisionMaker enabled: %s', __package__) self.engine = engine if engine else setup_database() def __call__(self): self.logger.info('DecisionMaker started') self.logger.debug('DecisionMaker loaded config: %s', self.config) self.external_tunables = self.load_external_conf() self.logger.debug('DecisionMaker loaded tunables: %s', self.external_tunables) report = self.data_processing() self.add_json_report(report) self.logger.info('DecisionMaker report: %s', report) return report @staticmethod def _report_file(name) -> str: """ Full path to given filename in DM reports directory """ return os.path.join(report_path, name) @property def current_report_file(self) -> str: """ Full path to current DM report: report.json in DM reports directory """ return self._report_file('report.json') @property def _empty_report(self) -> dict: """ Returns empty report """ return dict(date=previous_day_date(), domains=[]) @property def solo_filtered_options(self) -> set: return {'correlation'} @staticmethod def load_external_conf(): """Load external configuration values""" return load_tunables('ssa.json', ssa_tunables_schema) def data_processing(self) -> dict: """ Going through the list of domains, for each domain we go through the list of urls. During data processing, we will form the resulting dictionary. """ report = self._empty_report for domain_data in iter_domains_data(self.engine): # goes through the list of domains urls_data = list() domain_slow_reqs = 0 domain_url_durations = dict(get_url_durations( self.engine, domain_data.domain_name)) for domain_data_key, domain_data_value in iter_urls_data(self.engine, domain_data.domain_name, list(domain_url_durations.keys())): if self.is_ignored(domain_data_key): self.logger.debug('%s ignored', domain_data_key) continue # goes through the list of urls, "domain_total_reqs" is also here if domain_data_key not in self.non_url_fields: # domain_data_key below - it is current url if not self.is_throttling_suitable( domain_data_value.get('url_throttled_reqs', list([0] * 24)), domain_data_value['url_total_reqs']): # skip by allowed throttling percentage continue correlation_value = self.get_correlation( domain_data_value['url_total_reqs'], domain_data.domain_total_reqs) durations = domain_url_durations.get(domain_data_key) if durations is None: self.logger.error('Unable to get durations for %s', str(domain_data_key)) continue if (self.request_number_exceeded( domain_data_value['url_slow_reqs']) and self.correlation_conditions(correlation_value)): average_duration_calculation = np.mean(durations) sum_url_slow_reqs = sum( domain_data_value['url_slow_reqs']) domain_slow_reqs += sum_url_slow_reqs urls_data.append(dict( name=domain_data_key, reqs_num=sum_url_slow_reqs, average_duration=int(average_duration_calculation), correlation=float(f'{correlation_value:.2f}'))) if urls_data: sorted_urls = self.report_sorting( list_to_sort=urls_data, leave_top=self.urls_number, key_for_sorting='reqs_num') report['domains'].append(dict( name=domain_data.domain_name, slow_urls=len(sorted_urls), slow_reqs=domain_slow_reqs, total_reqs=sum(domain_data.domain_total_reqs), urls=sorted_urls)) if report['domains']: report['domains'] = self.report_sorting( list_to_sort=report['domains'], leave_top=self.domains_number, key_for_sorting='slow_reqs') return report def list_handling_considering_time(self, url_slow_reqs: list) -> list: """ Based on the 'url_slow_reqs' list, a new list will be formed, where the elements of the original list will be iteratively summed by the number of elements equal to 'time' """ time = self.time or 24 return [sum(url_slow_reqs[i:time + i]) for i in range(0, len(url_slow_reqs), time)] def compare_elements_with_request_number(self, url_slow_reqs_by_time: list) -> bool: """ This functions will check if any of elements is greater than "request_number" """ for i in url_slow_reqs_by_time: if i >= self.request_number: return True return False def get_correlation(self, url_total_reqs: list, domain_total_reqs: list): """ Calculates the correlation coefficient using the "url_total_reqs" and the "domain_total_reqs" lists """ if not self.correlation: return 0 return np.amin(np.corrcoef(url_total_reqs, domain_total_reqs)) @staticmethod def report_sorting(list_to_sort: list, leave_top: int, key_for_sorting: str) -> list: """ Will sort the domain list by "slow_reqs", the goal is to leave only "domains_number" of uppers, also per each domain will sort urls by "reqs_num", the goal is to leave only "urls_number" of uppers. leave_top == 0 allows to keep the full list """ list_to_sort.sort(key=lambda dict_: dict_[key_for_sorting], reverse=True) if leave_top: return list_to_sort[:leave_top] else: return list_to_sort def rename_old_report(self): """ Rename old report """ old_report = self.current_report_file if isfile(old_report): with open(old_report) as json_data: try: d = json.load(json_data) except json.JSONDecodeError: date_from_report = 'unknown' else: date_from_report = d.get('date', 'dd.mm.yyyy').replace('.', '_') new_report_name = f'report__{date_from_report}.json' new_report = self._report_file(new_report_name) os.rename(old_report, new_report) def add_json_report(self, report: dict): """ Makes json report """ self.rename_old_report() with open(self.current_report_file, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=4) def get_json_report(self) -> dict: """ Return contents of current report or empty report in case of error """ _filtering_hook = None try: with open(self.current_report_file) as report: report_dict = json.load(report, object_hook=_filtering_hook) except (OSError, json.JSONDecodeError): report_dict = self._empty_report return report_dict def correlation_conditions(self, correlation_value: int) -> bool: """ If correlation flag is enabled - we'll compare correlation_coefficient from configuration with calculated correlation coefficient. If the calculated value exceeds the configuration value - we return True otherwise False. At the same time if correlation flag is disabled - we'll also return "True" since in this case the correlation coefficient is not checked and its value is specified as zero in final report. """ if not self.correlation: return True return correlation_value > self.correlation_coefficient def request_number_exceeded(self, url_slow_reqs): """ At least one element from the received list (url_slow_reqs_by_time) must be greater than request_number """ url_slow_reqs_by_time = self.list_handling_considering_time( url_slow_reqs) return self.compare_elements_with_request_number(url_slow_reqs_by_time) def is_throttling_suitable(self, url_throttled_reqs: list, url_total_reqs: list) -> bool: """ Check that percent of throttled requests per URL passes given threshold """ throttled_percent = (sum(url_throttled_reqs) / sum( url_total_reqs)) * 100 self.logger.debug('Calculated throttled percent %s', throttled_percent) return throttled_percent <= self.external_tunables.get( 'allowed_throttling_percentage', 0) if __name__ == "__main__": sentry_init() logging.basicConfig(filename='decision_maker_standalone.log', level=logging.INFO) dm = DecisionMaker() dm()	10.38 KB	2026-02-10 14:37
📄 processor.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains RequestProcessor class """ import logging import sys import time import signal import traceback from datetime import datetime, timedelta, timezone from threading import Thread, RLock, current_thread from typing import Callable, Any from queue import Queue, Empty from sqlalchemy.exc import OperationalError from .autotracer import AutoTracer from .common import Common from .decision_maker import DecisionMaker from .stat_sender import StatisticsSender from ..db import session_scope, setup_database, RequestResult, cleanup_old_data, restore_database, is_malformed_database from ..internal.exceptions import SSAError from ..internal.utils import ( singleton, url_split, switch_schedstats ) @singleton class RequestProcessor(Common): """ SSA Request processor implementation. Only one instance is allowed to be created """ BUFFER_SIZE = 100 DB_ACCESS_RETRIES = 5 def __init__(self, engine=None): super().__init__() self.logger = logging.getLogger('req_processor') self.logger.info('Processor enabled: %s', __package__) # enable throttling detection kernel mechanism on service start switch_schedstats(enabled=True) self.engine = engine if engine else setup_database() self._lock = RLock() self.decision_maker = DecisionMaker(engine=self.engine) self.sender = StatisticsSender() self.auto_tracer = AutoTracer(engine=self.engine) self._queue = Queue() self._buffer = [] self.start_background_routine() self.start_flush_worker() # Catch the shutdown signal for save last data from buffer if < BUFFER_SIZE signal.signal(signal.SIGTERM, self.shutdown_handler) signal.signal(signal.SIGINT, self.shutdown_handler) @property def configured_duration(self): """ Return config file value multiplied by 1000000, as we receive duration in microseconds """ return self.requests_duration * 1000000 def send_stats(self, report: dict): """ Call Statistics Sender """ try: self.sender.send(report) except SSAError as e: self.logger.error('StatisticsSender failed: %s', str(e)) def start_background_routine(self) -> None: """ Start dumper\|DecisionMaker thread in background """ t = Thread(target=self.background_routine, daemon=True) t.start() self.logger.info('[%s] Routine started', t.name) def start_flush_worker(self) -> None: """ Start flush worker thread """ t = Thread(target=self.flush_worker, daemon=True) t.start() self.logger.info('[%s] Flush worker started', t.name) def background_routine(self) -> None: """ Dumps collected stats to file once an hour. Runs DecisionMaker once a day Cleanup storage after DecisionMaker run """ while True: tick = datetime.now(timezone.utc) if tick.minute == 0: if tick.hour == 0: if is_malformed_database(self.engine): self._safe_exec(self.restore_db_with_lock(self.engine)) self.logger.info( '[%s] Routine thread found Database disk image is malformed and now restored (%s)', current_thread().name, tick) self.logger.info( '[%s] Routine thread launching buffer flushing (%s)', current_thread().name, tick) self._safe_exec(self.flush_remaining_objects) self.logger.info( '[%s] Routine thread launching AutoTracer (%s)', current_thread().name, tick) self._safe_exec(self.auto_tracer) self.logger.info( '[%s] Routine thread launching DecisionMaker (%s)', current_thread().name, tick) report = self._safe_exec(self.decision_maker) self.logger.info( '[%s] Routine thread launching cleanup (%s)', current_thread().name, tick) cleanup_old_data(self.engine) self._safe_exec(self.send_stats, report) # attempt to enable throttling detection kernel mechanism # in case it was accidentally switched off switch_schedstats(enabled=True) self._simple_sleep(60) else: self._sleep_till_next_hour(tick.minute) def _safe_exec(self, action: Callable, args) -> Any: """Call requested Callable with given args and capture any exception""" try: return action(args) except Exception: et, ev, _ = sys.exc_info() self.logger.exception('%s failed with exception %s, %s', str(action), et, ev, extra={'orig_traceback': traceback.format_exc()}) def _simple_sleep(self, to_sleep: int = 15 * 60): """ Log and sleep given number of seconds or 15 minutes by default """ self.logger.info('[%s] Routine thread sleeping for (%s)', current_thread().name, to_sleep) time.sleep(to_sleep) def _sleep_till_next_hour(self, start_minute): """ Sleep the number of minutes remaining till next hour """ sleep_for = (timedelta(hours=1) - timedelta( minutes=start_minute)).total_seconds() self._simple_sleep(int(sleep_for)) def restore_db_with_lock(self, engine): with self._lock: restore_database(engine) @staticmethod def get_interval_for(timestamp: int) -> int: """ Takes an hour of a day, to which the given timestamp belongs """ return datetime.fromtimestamp(timestamp, timezone.utc).hour def flush_worker(self): """ Continuously flush queued request results to the database """ while True: try: obj = self._queue.get() self._buffer.append(obj) if len(self._buffer) >= self.BUFFER_SIZE: self._flush_objects(self._buffer) self._buffer = [] except Exception: self.logger.exception('Flush worker failed') def flush_remaining_objects(self): """ Flush all remaining objects even if less than BUFFER_SIZE. Should be called once a day. """ if self._buffer: try: self._flush_objects(self._buffer) except Exception: self.logger.exception('Flush remaining objects failed') finally: self._buffer = [] def _flush_objects(self, objects): for attempt in range(self.DB_ACCESS_RETRIES): try: with session_scope(self.engine) as db: db.bulk_save_objects(objects) break except OperationalError as e: if "database is locked" in str(e): self.logger.warning('Database is locked, retrying attempt %s/%s...', attempt + 1, self.DB_ACCESS_RETRIES) time.sleep(0.1) else: raise def handle(self, data: dict) -> None: """ Process given request data """ if not data: self.logger.info('[%s] has empty request, skipping', current_thread().name) return url = data.get('url') if self.is_ignored(url): self.logger.debug('%s ignored', url) return domain, uri = url_split(url) request = RequestResult( domain=domain, path=url, timestamp=data['timestamp'], duration=data['duration'], is_slow_request=data['duration'] > self.configured_duration, hitting_limits=data['hitting_limits'], throttled_time=data['throttled_time'], io_throttled_time=data['io_throttled_time'], wordpress=data['wordpress'], ) self._queue.put(request) def shutdown_handler(self, signum, frame): """ Handle shutdown signals to flush remaining objects before exit. """ self.logger.info(f'Received shutdown signal {signum}, flushing queue and buffer before shutdown') try: self._drain_queue_to_buffer() self.flush_remaining_objects() except Exception: self.logger.exception('Failed to flush remaining objects during shutdown') finally: sys.exit(0) def _drain_queue_to_buffer(self): """ Drain all remaining objects from queue to buffer. """ while not self._queue.empty(): try: obj = self._queue.get_nowait() self._buffer.append(obj) except Empty: break	9.24 KB	2026-02-10 14:37
📄 stat_sender.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains SSA classes for sending e-mails with report and sending reports to ClickHouse """ __package__ = 'ssa.modules' import json import logging import subprocess from email.message import EmailMessage from functools import partial from typing import Optional, Any from socket import gethostname from clcommon.cpapi import get_admin_email from requests import Session, Response from requests.adapters import HTTPAdapter from requests.exceptions import RequestException from requests.packages.urllib3.util.retry import Retry from .common import Common from .decision_maker import DecisionMaker from ..internal.constants import stat_server from ..internal.exceptions import SSAError from ..internal.mailer import Mailer, render_report_table from ..internal.utils import read_sys_id, sentry_init, duration_cast, \ format_date class StatisticsSender(Common): """ Send report to ClickHouse and over e-mails """ def __init__(self, ch_endpoint: str = f'https://{stat_server}/api/clos-ssa'): super().__init__() self.logger = logging.getLogger('stat_sender') self.logger.info('StatisticsSender enabled: %s', __package__) self.sys_id = read_sys_id() self.ch_endpoint = ch_endpoint self.mail_sender = Mailer() retry_conf = Retry(total=3, allowed_methods=frozenset(['POST']), status_forcelist=frozenset([502, 503, 504]), backoff_factor=3) # sleeps 0s, 6s, 18s adapter = HTTPAdapter(max_retries=retry_conf) self.session = Session() self.session.mount('https://', adapter) self.session.request = partial(self.session.request, timeout=10) def send(self, report: dict = None) -> None: """ Send given report over e-mail If no report given, get it from DecisionMaker API """ if report is None: report = DecisionMaker().get_json_report() if self.summary_notification_enabled: self.email_report(report) self.clickhouse_report(report) def email_report(self, report_view: dict) -> Optional[EmailMessage]: """ Create and send e-mail with report """ mail_to = self.get_mail_recipient() if mail_to and report_view.get('domains'): report_table, mail = render_report_table(report_view) msg = self.mail_sender._message( recipient=mail_to, template='ssa_report', date=format_date(report_view['date']), hostname=gethostname(), html=mail ) msg.add_attachment(json.dumps(report_view), subtype='json', filename=f"report_{report_view['date']}.json") msg.add_attachment(report_table, subtype='html', filename=f"report_{report_view['date']}.html") self.mail_sender._send(msg) return msg def get_mail_recipient(self) -> Optional[str]: """ Retrieve a recipient's e-mail: 1. get address from a wmt-api utility 2. if command failed or address is empty, get address of server admin """ try: # get_admin_email could return '', None, or throw unexpected errors return self.wmt_api_report_email() or get_admin_email() except Exception as e: self.logger.error('get_admin_email failed with: %s', str(e)) def wmt_api_report_email(self) -> Optional[str]: """ Retrieve a recipient's e-mail address from WMT API """ _util = 'wmt-api' try: api_response = subprocess.run( [f'/usr/share/web-monitoring-tool/wmtbin/{_util}', '--config-get'], check=True, text=True, capture_output=True).stdout.strip() except (subprocess.CalledProcessError, AttributeError, OSError, ValueError) as e: self.logger.error('wmt-api utility failed: %s', str(e)) return try: return json.loads(api_response).get('config').get('report_email') except json.JSONDecodeError as e: self.logger.error('wmt-api returned invalid json: %s', str(e)) except AttributeError: self.logger.error('wmt-api returned unexpected response: %s', api_response) def clickhouse_report(self, report_view: dict) -> bool: """ Send report to ClickHouse """ if report_view.get('domains'): self.logger.info('Sending POST request to %s', self.ch_endpoint) try: resp = self.session.post(self.ch_endpoint, json=self._ch_pack( self.clickhouse_format(report_view))) except RequestException as e: self.logger.error('POST failed with %s', e, extra={'endpoint': self.ch_endpoint}) raise SSAError( f'Failed to POST data to SSA API server: {str(e)}') from e return self._process_response(resp) else: self.logger.info('Report is empty, not sending to ClickHouse') return False @staticmethod def _ch_pack(value: Any) -> dict: """ Pack given value into data field of a dict """ return dict(data=value) def clickhouse_format(self, original_report: dict) -> list: """ Format local report for sending to ClickHouse (required structures differ) """ ch_report = list() for domain in original_report.get('domains'): ch_report.append({ 'system_id': self.sys_id, 'domain': domain.get('name'), 'count_slow_urls': domain.get('slow_urls'), 'count_slow_requests': domain.get('slow_reqs'), 'total_requests': domain.get('total_reqs'), 'details': [ { "url": u.get('name'), "count_requests": u.get('reqs_num'), "avg_duration": duration_cast( u.get('average_duration')), "correlation": float(u.get('correlation', 0)) } for u in domain.get('urls') ] }) return ch_report def _process_response(self, response: Response) -> bool: """ Check received response :param response: a requests.Response object :return: True in case of success, False otherwise """ if not response.ok: self.logger.error('Unable to connect to server with %s:%s', response.status_code, response.reason, extra={'resp_text': response.text}) return False else: self.logger.info('[%s:%s] Response received %s', response.status_code, response.reason, response.url) result = response.json() if result['status'] != 'ok': self.logger.error('Received response with status %s', result['status'], extra={'response': result}) return False self.logger.info('Sent to ClickHouse successfully') return True if __name__ == "__main__": sentry_init() logging.basicConfig(filename='stat_sender_standalone.log', level=logging.INFO) try: StatisticsSender().send() except SSAError as exc: print(exc) raise SystemExit(1)	7.9 KB	2026-02-10 14:37
📄 storage.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ Represents storage where ssa data is collected, stored and extracted """ import itertools from dataclasses import dataclass from typing import List, Iterator, Tuple, Dict import sqlalchemy from sqlalchemy import func, cast, distinct, text from ssa.db import session_scope, RequestResult @dataclass class DomainData: domain_name: str domain_total_reqs: List[int] is_a_wordpress_domain: bool urls_number: int def iter_domains_data(engine) -> Iterator[DomainData]: """ Iterates data from database domain-by-domain. """ with session_scope(engine) as db: results_by_hour = db.query( RequestResult.domain, func.strftime('%H', RequestResult.created_at), func.Count(RequestResult.id), func.max(RequestResult.wordpress), func.count(distinct(RequestResult.path)) ).group_by( RequestResult.domain, func.strftime('%H', RequestResult.created_at) ).order_by( RequestResult.domain, func.strftime('%H', RequestResult.created_at) ) results_by_hour_grouped = itertools.groupby(results_by_hour, key=lambda item: item[0]) for domain_name, group in results_by_hour_grouped: domain_results_by_hour = tuple(group) urls_number = 0 # at some hours there may be no requests # so we must normalize data to match 24h data format requests_number_by_hour = [0] * 24 for _, hour, requests_num, is_wordpress, urls in domain_results_by_hour: requests_number_by_hour[int(hour)] = requests_num urls_number = max(urls_number, urls) yield DomainData( domain_name=domain_name, domain_total_reqs=requests_number_by_hour, is_a_wordpress_domain=is_wordpress, urls_number=urls_number ) def iter_urls_data(engine, domain_name, all_paths): """ Iterates urls data from database url-by-url. """ with session_scope(engine) as db: all_paths_escaped = [path.replace(":", "\\:") for path in all_paths] urls_data = db.query( RequestResult.path, func.strftime('%H', RequestResult.created_at), func.Sum(cast( RequestResult.hitting_limits, sqlalchemy.Integer )).label('url_throttled_reqs'), func.Count( RequestResult.id ).label('url_total_reqs'), func.Sum(cast( RequestResult.is_slow_request, sqlalchemy.Integer) ).label('url_slow_reqs') ).filter( RequestResult.domain == domain_name ).filter( text(RequestResult.path.in_(all_paths_escaped).expression.compile(compile_kwargs={"literal_binds": True}).string) ).group_by( RequestResult.path, func.strftime('%H', RequestResult.created_at) ).order_by( RequestResult.path, func.strftime('%H', RequestResult.created_at) ) previous_path = None url_throttled_reqs, url_total_reqs, url_slow_reqs = \ [0] * 24, [0] * 24, [0] * 24 for path, hour, url_throttled_req, url_total_req, url_slow_req in urls_data: if previous_path and previous_path != path: yield previous_path, dict( path=previous_path, url_throttled_reqs=url_throttled_reqs, url_total_reqs=url_total_reqs, url_slow_reqs=url_slow_reqs ) url_throttled_reqs, url_total_reqs, url_slow_reqs = \ [0] * 24, [0] * 24, [0] * 24 url_throttled_reqs[int(hour)] = url_throttled_req url_total_reqs[int(hour)] = url_total_req url_slow_reqs[int(hour)] = url_slow_req previous_path = path yield path, dict( path=path, url_throttled_reqs=url_throttled_reqs, url_total_reqs=url_total_reqs, url_slow_reqs=url_slow_reqs ) def get_url_durations(engine, domain_name) -> Dict[str, Tuple[int]]: """ Get information about durations of requests url-by-url. """ with session_scope(engine) as db: urls_data = db.query( RequestResult.path, RequestResult.duration ).filter( RequestResult.domain == domain_name ).order_by( RequestResult.path ) # Use iterator directly to avoid loading all data into RAM durations_by_path = itertools.groupby( urls_data, lambda item: item[0]) for key, group in durations_by_path: yield key, [duration for _, duration in group]	4.88 KB	2026-02-10 14:37

Terminal