pi3c
/
mem_checker
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

sync

main
Сергей Ванюшкин 2024-04-13 04:08:48 +03:00
parent 30dbd7d117
commit 4ec6e097a6
3 changed files with 117 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.pre-commit-config.yaml
View File

@ -23,7 +23,7 @@ repos:
rev: v3.15.2
hooks:
- id: pyupgrade
args: [ --py311-plus ]
args: [ --py310-plus ]
# Форматирует код под PEP8
- repo: https://github.com/pre-commit/mirrors-autopep8
@ -46,7 +46,7 @@ repos:
rev: 24.3.0
hooks:
- id: black
language_version: python3.11
language_version: python3.10
args: [ "--line-length=120" ]
# Проверка статических типов с помощью mypy
@ -54,5 +54,4 @@ repos:
rev: v1.9.0
hooks:
- id: mypy
exclude: 'migrations'
args: [--no-strict-optional, --ignore-missing-imports]

140
mem_checker/mem_checker.py
View File

@ -1,10 +1,38 @@
"""
A script for monitoring available RAM and free disk space.
It uses the free and df utilites.
When the threshold is exceeded, it sends a post request to the api.
The request contains the following json format:
[
{
type_msg: str,
message: str
detail: str
}
]
:copyright: Sergey Vanyushkin <pi3c@yandex.ru>
:git: https://git.pi3c.ru/pi3c/mem_checker.git
:license: MIT
2024г.
"""
import json
import subprocess
import urllib.parse
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import dataclass
from time import sleep
ENDPOINT_URL = "http://127.0.0.1:8000/user" # endpoint for send alerts
FREE_IN_MB = 500 # free: the value of the trigger in megabytes
FREE_IN_GB = 0.5 # free: the value of the trigger in gigabytes
DF_IN_PERCENT = 90 # df: the value of the trigger in percents
SLEEPTIME = 5 # main_loop sleep time per seconds
@dataclass(frozen=True)
@dataclass()
class MessageDTO:
type_msg: str
message: str
@ -24,13 +52,21 @@ class Command(ABC):
def _stdout_parser(self, stdout: str) -> list[MessageDTO]:
raise NotImplementedError
@abstractmethod
def _filter_alerts(self, stat: list[MessageDTO]) -> list[MessageDTO]:
raise NotImplementedError
def _execute(self, command: list) -> tuple[str, str]:
try:
result = subprocess.run(
command,
capture_output=True,
check=True,
text=True,
)
except Exception as e:
return "", str(e)
return result.stdout, result.stderr
def get_alerts(self) -> list[MessageDTO]:
@ -43,8 +79,8 @@ class Command(ABC):
detail=stderr,
)
]
alerts = self._stdout_parser(stdout=stdout)
return alerts
stat = self._stdout_parser(stdout=stdout)
return self._filter_alerts(stat)
class FreeMem(Command):
@ -53,15 +89,36 @@ class FreeMem(Command):
) -> None:
self.command = ["free", "-h"]
def _filter_alerts(self, stat: list[MessageDTO]) -> list[MessageDTO]:
filtered_alerts = []
for obj in stat:
if obj.type_msg == "Error":
filtered_alerts.append(obj)
elif obj.type_msg == "Status":
val, unit = obj.detail.split()
if (float(val) < FREE_IN_MB and unit == "Mi") or (float(val) < FREE_IN_GB and unit == "Gi"):
obj.type_msg = "Alert"
filtered_alerts.append(obj)
return filtered_alerts
def _stdout_parser(self, stdout: str) -> list[MessageDTO]:
memory = stdout.split()[9]
try:
memory = stdout.split()[12]
memory_units = memory[-2:]
memory = memory[:-2].replace(",", ".")
memory_value = float(memory) if "." in memory else int(memory)
except Exception as e:
return [
MessageDTO(
type_msg="ParserError",
message="Stdout parsing runtime error",
detail=str(e),
),
]
return [
MessageDTO(
type_msg="Status",
message="free mem",
message="free mem available",
detail=f"{memory_value} {memory_units}",
)
]
@ -71,48 +128,53 @@ class FreeHdd(Command):
def __init__(self) -> None:
self.command = ["df", "-h"]
def _filter_alerts(self, stat: list[MessageDTO]) -> list[MessageDTO]:
filtered_alerts = list()
for obj in stat:
persents = int(obj.detail.split()[4][:-1])
if DF_IN_PERCENT < persents:
obj.type_msg = "Alert"
filtered_alerts.append(obj)
return filtered_alerts
def _stdout_parser(self, stdout) -> list[MessageDTO]:
return []
def row_to_dto(row: list):
return MessageDTO(type_msg="Status", message="hdd mem available", detail=" ".join(row))
stat = [row.split() for row in stdout.split("\n")]
stat = list(map(row_to_dto, stat[1:-1]))
return self._filter_alerts(stat)
def send_alert(alerts: list) -> None:
pass
# print(*alerts, sep="\n")
def send_alert(alerts: list[MessageDTO]) -> None:
data = [{"type_msg": item.type_msg, "message": item.message, "detail": item.detail} for item in alerts]
json_data = json.dumps(data).encode("utf-8")
req = urllib.request.Request(
ENDPOINT_URL,
data=json_data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req) as response:
response.read().decode("utf-8")
# def shell_command_result(command: list) -> tuple[str, str]:
# process = subprocess.run(
# command,
# capture_output=True,
# check=True,
# text=True,
# )
# return process.stdout, process.stderr
#
#
# def get_free_mem() -> tuple[int | float, str]:
# stdout, stderr = shell_command_result(command=["free", "-h"])
# if stderr:
# return
#
# return memory_value, memory_units
#
#
# def get_hdd_free_mem() -> list[tuple[str, str]]:
# stdout, stderr = shell_command_result(command=["df", "-h"])
# rows = [row.split() for row in stdout.split("\n")]
# mount_points = [(row[5], row[4]) for row in rows[1:] if row]
# return mount_points
#
#
def main_loop() -> None:
free_mem = FreeMem()
while True:
stats = free_mem.get_alerts()
hdd_mem = FreeHdd()
stats += []
# print(stats)
sleep(1)
while True:
alerts: list[MessageDTO] = list()
alerts.extend(free_mem.get_alerts())
alerts.extend(hdd_mem.get_alerts())
if alerts:
send_alert(alerts=alerts)
sleep(SLEEPTIME)
if __name__ == "__main__":

11
poetry.lock generated
View File

@ -124,7 +124,6 @@ files = [
{file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"},
{file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"},
{file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"},
{file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"},
{file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"},
{file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"},
{file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"},
@ -161,18 +160,18 @@ files = [
[[package]]
name = "setuptools"
version = "69.2.0"
version = "69.4.0"
description = "Easily download, build, install, upgrade, and uninstall Python packages"
optional = false
python-versions = ">=3.8"
files = [
{file = "setuptools-69.2.0-py3-none-any.whl", hash = "sha256:c21c49fb1042386df081cb5d86759792ab89efca84cf114889191cd09aacc80c"},
{file = "setuptools-69.2.0.tar.gz", hash = "sha256:0ff4183f8f42cd8fa3acea16c45205521a4ef28f73c6391d8a25e92893134f2e"},
{file = "setuptools-69.4.0-py3-none-any.whl", hash = "sha256:b6df12d754b505e4ca283c61582d5578db83ae2f56a979b3bc9a8754705ae3bf"},
{file = "setuptools-69.4.tar.gz", hash = "sha256:659e902e587e77fab8212358f5b03977b5f0d18d4724310d4a093929fee4ca1a"},
]
[package.extras]
docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "pygments-github-lexers (==0.0.5)", "rst.linker (>=1.9)", "sphinx (<7.2.5)", "sphinx (>=3.5)", "sphinx-favicon", "sphinx-inline-tabs", "sphinx-lint", "sphinx-notfound-page (>=1,<2)", "sphinx-reredirects", "sphinxcontrib-towncrier"]
testing = ["build[virtualenv]", "filelock (>=3.4.0)", "importlib-metadata", "ini2toml[lite] (>=0.9)", "jaraco.develop (>=7.21)", "jaraco.envs (>=2.2)", "jaraco.path (>=3.2.0)", "mypy (==1.9)", "packaging (>=23.2)", "pip (>=19.1)", "pytest (>=6)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-home (>=0.5)", "pytest-mypy (>=0.9.1)", "pytest-perf", "pytest-ruff (>=0.2.1)", "pytest-timeout", "pytest-xdist (>=3)", "tomli", "tomli-w (>=1.0.0)", "virtualenv (>=13.0.0)", "wheel"]
docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "pygments-github-lexers (==0.0.5)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-favicon", "sphinx-inline-tabs", "sphinx-lint", "sphinx-notfound-page (>=1,<2)", "sphinx-reredirects", "sphinxcontrib-towncrier"]
testing = ["build[virtualenv]", "filelock (>=3.4.0)", "importlib-metadata", "ini2toml[lite] (>=0.9)", "jaraco.develop (>=7.21)", "jaraco.envs (>=2.2)", "jaraco.path (>=3.2.0)", "mypy (==1.9)", "packaging (>=23.2)", "pip (>=19.1)", "pytest (>=6,!=8.1.1)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-home (>=0.5)", "pytest-mypy", "pytest-perf", "pytest-ruff (>=0.2.1)", "pytest-timeout", "pytest-xdist (>=3)", "tomli", "tomli-w (>=1.0.0)", "virtualenv (>=13.0.0)", "wheel"]
testing-integration = ["build[virtualenv] (>=1.0.3)", "filelock (>=3.4.0)", "jaraco.envs (>=2.2)", "jaraco.path (>=3.2.0)", "packaging (>=23.2)", "pytest", "pytest-enabler", "pytest-xdist", "tomli", "virtualenv (>=13.0.0)", "wheel"]
[[package]]