REDROOM

Navigation

Breadcrumbs

/ / opt / cloudlinux / venv / lib64 / python3.11 / site-packages / ssa / modules

Create

Create File | Create Folder | Upload Files | Run Command

Edit File Size: 11.65 KB

//opt/cloudlinux/venv/lib64/python3.11/site-packages/ssa/modules/processor.py

Text Base64

# -*- 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, Full
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,
    ensure_indexes,
    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
    # Bound the ingest queue so a tenant telemetry flood cannot grow the
    # root agent's RSS without limit; scales with the flush batch.
    QUEUE_MAXSIZE = BUFFER_SIZE * 10
    # Rate-limit the drop warning so a sustained flood does not also spam the log.
    DROP_LOG_INTERVAL = 1000

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(maxsize=self.QUEUE_MAXSIZE)
        self._dropped_requests = 0
        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.
        """
        while True:
            tick = datetime.now(timezone.utc)
            if tick.minute == 0:
                if tick.hour == 0:
                    self._daily_routine(tick)
                self._simple_sleep(60)
            else:
                self._sleep_till_next_hour(tick.minute)

def _daily_routine(self, tick) -> None:
        """
        Once-a-day maintenance: flush buffer, prune retention, run
        AutoTracer and DecisionMaker, then send stats.

Cleanup runs FIRST and is wrapped in ``_safe_exec`` so that
        retention is always enforced even if a later step
        (AutoTracer / DecisionMaker) is killed by the cgroup OOM killer
        on a large ``ssa.db`` (CLPRO-3077). Otherwise the DB keeps
        growing across restarts and the OOM becomes self-reinforcing.
        """
        if is_malformed_database(self.engine):
            # Pass the callable and its arg separately: calling it inline
            # would run restore_database OUTSIDE _safe_exec (and then invoke
            # _safe_exec(None) -> TypeError), so a restore failure would
            # escape into background_routine's `while True` (CLPRO-3077).
            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 cleanup (%s)", current_thread().name, tick
        )
        self._safe_exec(cleanup_old_data, self.engine)
        # Build the (domain, path) index on the post-cleanup (smaller)
        # table, before the AutoTracer / DecisionMaker scans that rely on
        # it to stream in index order instead of an external ORDER BY sort
        # whose temp spill could be charged against the cgroup on tmpfs
        # (CLPRO-3077).
        self.logger.info(
            "[%s] Routine thread ensuring indexes (%s)", current_thread().name, tick
        )
        self._safe_exec(ensure_indexes, self.engine)
        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._safe_exec(self.send_stats, report)
        # attempt to enable throttling detection kernel mechanism
        # in case it was accidentally switched off. Wrapped in _safe_exec
        # like every other daily step so a failure here can't escape into
        # background_routine's unguarded `while True` and kill the thread.
        self._safe_exec(switch_schedstats, True)

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"],
        )

# Non-blocking enqueue: if the single flush_worker can't keep up and
        # the queue is full, drop the request rather than block the socket
        # handler thread (which would propagate the stall into the root agent)
        # or grow the queue without bound.
        try:
            self._queue.put_nowait(request)
        except Full:
            self._dropped_requests += 1
            if self._dropped_requests % self.DROP_LOG_INTERVAL == 1:
                self.logger.warning(
                    "Request queue full (maxsize=%s); dropped %s request(s) so far",
                    self.QUEUE_MAXSIZE,
                    self._dropped_requests,
                )

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

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Directory Contents

Dirs: 1 × Files: 7

Sort:

Enable drag-to-move

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-07-22 02:20:04	Edit Download recursive recursive
autotracer.py	21.26 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
common.py	1.97 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
decision_maker.py	11.40 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
processor.py	11.65 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
stat_sender.py	7.90 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
storage.py	8.18 KB	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive
__init__.py	0 B	lrw-r--r--	2026-07-07 15:08:54	Edit Download recursive recursive