Zy Filemanager

📁 File Browser

Name	Size	Permissions	Actions
📁 __pycache__	DIR	0755	🗑️
📁 ai	DIR	0755	🗑️
📁 crons	DIR	0755	🗑️
📁 integrations	DIR	0755	🗑️
📁 profiler	DIR	0755	🗑️
📄 __init__.py (1,410 bytes)	1.38 KB	0644	✍️ ⬇️ 🗑️
📄 _compat.py (3,116 bytes)	3.04 KB	0644	✍️ ⬇️ 🗑️
📄 _init_implementation.py (2,559 bytes)	2.5 KB	0644	✍️ ⬇️ 🗑️
📄 _log_batcher.py (5,500 bytes)	5.37 KB	0644	✍️ ⬇️ 🗑️
📄 _lru_cache.py (1,229 bytes)	1.2 KB	0644	✍️ ⬇️ 🗑️
📄 _metrics_batcher.py (5,042 bytes)	4.92 KB	0644	✍️ ⬇️ 🗑️
📄 _queue.py (11,250 bytes)	10.99 KB	0644	✍️ ⬇️ 🗑️
📄 _types.py (10,422 bytes)	10.18 KB	0644	✍️ ⬇️ 🗑️
📄 _werkzeug.py (3,734 bytes)	3.65 KB	0644	✍️ ⬇️ 🗑️
📄 api.py (15,282 bytes)	14.92 KB	0644	✍️ ⬇️ 🗑️
📄 attachments.py (3,109 bytes)	3.04 KB	0644	✍️ ⬇️ 🗑️
📄 client.py (42,099 bytes)	41.11 KB	0644	✍️ ⬇️ 🗑️
📄 consts.py (53,733 bytes)	52.47 KB	0644	✍️ ⬇️ 🗑️
📄 debug.py (1,019 bytes)	1019 B	0644	✍️ ⬇️ 🗑️
📄 envelope.py (10,473 bytes)	10.23 KB	0644	✍️ ⬇️ 🗑️
📄 feature_flags.py (2,232 bytes)	2.18 KB	0644	✍️ ⬇️ 🗑️
📄 hub.py (25,748 bytes)	25.14 KB	0644	✍️ ⬇️ 🗑️
📄 logger.py (2,797 bytes)	2.73 KB	0644	✍️ ⬇️ 🗑️
📄 metrics.py (2,044 bytes)	2 KB	0644	✍️ ⬇️ 🗑️
📄 monitor.py (3,639 bytes)	3.55 KB	0644	✍️ ⬇️ 🗑️
📄 py.typed (0 bytes)	0 B	0644	✍️ ⬇️ 🗑️
📄 scope.py (65,420 bytes)	63.89 KB	0644	✍️ ⬇️ 🗑️
📄 scrubber.py (6,064 bytes)	5.92 KB	0644	✍️ ⬇️ 🗑️
📄 serializer.py (13,534 bytes)	13.22 KB	0644	✍️ ⬇️ 🗑️
📄 session.py (5,589 bytes)	5.46 KB	0644	✍️ ⬇️ 🗑️
📄 sessions.py (9,172 bytes)	8.96 KB	0644	✍️ ⬇️ 🗑️
📄 spotlight.py (8,678 bytes)	8.47 KB	0644	✍️ ⬇️ 🗑️
📄 tracing.py (51,747 bytes)	50.53 KB	0644	✍️ ⬇️ 🗑️
📄 tracing_utils.py (40,540 bytes)	39.59 KB	0644	✍️ ⬇️ 🗑️
📄 transport.py (31,878 bytes)	31.13 KB	0644	✍️ ⬇️ 🗑️
📄 types.py (1,269 bytes)	1.24 KB	0644	✍️ ⬇️ 🗑️
📄 utils.py (63,900 bytes)	62.4 KB	0644	✍️ ⬇️ 🗑️
📄 worker.py (4,464 bytes)	4.36 KB	0644	✍️ ⬇️ 🗑️

✍️ Editing: monitor.py

            File: /opt/hc_python/lib/python3.12/site-packages/sentry_sdk/monitor.py

            Size: 3.55 KB | 
            Last Modified: 2025-12-03 14:34:44        

import os
import time
from threading import Thread, Lock

import sentry_sdk
from sentry_sdk.utils import logger

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from typing import Optional

MAX_DOWNSAMPLE_FACTOR = 10

class Monitor:
    """
    Performs health checks in a separate thread once every interval seconds
    and updates the internal state. Other parts of the SDK only read this state
    and act accordingly.
    """

name = "sentry.monitor"

def __init__(self, transport, interval=10):
        # type: (sentry_sdk.transport.Transport, float) -> None
        self.transport = transport  # type: sentry_sdk.transport.Transport
        self.interval = interval  # type: float

self._healthy = True
        self._downsample_factor = 0  # type: int

self._thread = None  # type: Optional[Thread]
        self._thread_lock = Lock()
        self._thread_for_pid = None  # type: Optional[int]
        self._running = True

def _ensure_running(self):
        # type: () -> None
        """
        Check that the monitor has an active thread to run in, or create one if not.

Note that this might fail (e.g. in Python 3.12 it's not possible to
        spawn new threads at interpreter shutdown). In that case self._running
        will be False after running this function.
        """
        if self._thread_for_pid == os.getpid() and self._thread is not None:
            return None

with self._thread_lock:
            if self._thread_for_pid == os.getpid() and self._thread is not None:
                return None

def _thread():
                # type: (...) -> None
                while self._running:
                    time.sleep(self.interval)
                    if self._running:
                        self.run()

thread = Thread(name=self.name, target=_thread)
            thread.daemon = True
            try:
                thread.start()
            except RuntimeError:
                # Unfortunately at this point the interpreter is in a state that no
                # longer allows us to spawn a thread and we have to bail.
                self._running = False
                return None

self._thread = thread
            self._thread_for_pid = os.getpid()

return None

def run(self):
        # type: () -> None
        self.check_health()
        self.set_downsample_factor()

def set_downsample_factor(self):
        # type: () -> None
        if self._healthy:
            if self._downsample_factor > 0:
                logger.debug(
                    "[Monitor] health check positive, reverting to normal sampling"
                )
            self._downsample_factor = 0
        else:
            if self.downsample_factor < MAX_DOWNSAMPLE_FACTOR:
                self._downsample_factor += 1
            logger.debug(
                "[Monitor] health check negative, downsampling with a factor of %d",
                self._downsample_factor,
            )

def check_health(self):
        # type: () -> None
        """
        Perform the actual health checks,
        currently only checks if the transport is rate-limited.
        TODO: augment in the future with more checks.
        """
        self._healthy = self.transport.is_healthy()

def is_healthy(self):
        # type: () -> bool
        self._ensure_running()
        return self._healthy

@property
    def downsample_factor(self):
        # type: () -> int
        self._ensure_running()
        return self._downsample_factor

def kill(self):
        # type: () -> None
        self._running = False

↩️ Back

Zy Filemanager

📁 File Browser

📤 Upload Files

Single File

Multiple Files

ZIP Extract

➕ Create New

New Folder

New File

✍️ Editing: monitor.py

Backup Management