Tentative first version

This commit is contained in:
Théophile Bastian 2022-12-12 14:12:40 +01:00
parent 5be5c4b444
commit 8ecc1f85b7
4 changed files with 90 additions and 0 deletions

1
.gitignore vendored
View file

@ -1,2 +1,3 @@
*.sqlite3 *.sqlite3
*.egg-info *.egg-info
__pycache__

66
netmon/entry.py Normal file
View file

@ -0,0 +1,66 @@
from .ping import Ping
import time
import typing as t
import sqlite3
import argparse
import logging
logger = logging.getLogger(__name__)
def record(target: str, interval: int, outage_interval: int):
"""Record errors while pinging `target` each `interval` seconds"""
ping = Ping(target)
db = sqlite3.connect("netmon.sqlite3")
outage: t.Optional[int] = None
try:
while True:
res = ping.ping()
if outage is None and not res: # beginning of outage
outage = int(time.time())
logger.info("Could not reach target.")
elif outage is not None and res: # end of outage
delta = int(time.time() - outage)
logger.info(
"Could reach target again (out for %d min %d s)",
delta // 60,
delta % 60,
)
with db:
db.execute(
"INSERT INTO errors (from_time, to_time, host) VALUES (?, ?, ?);",
(int(outage), int(time.time()), target),
)
time.sleep(interval if outage is None else outage_interval)
except KeyboardInterrupt as exn:
if outage is not None:
with db:
db.execute(
"INSERT INTO errors (from_time, to_time, host) VALUES (?, ?, ?);",
(int(outage), int(time.time()), target),
)
raise exn
def entry():
parser = argparse.ArgumentParser()
parser.add_argument("host", help="Remote host to ping (IP address preferred)")
parser.add_argument("-i", "--interval", default=30, help="Time between two pings")
parser.add_argument(
"-I",
"--outage-interval",
default=2,
help="Time between two pings while the network is down",
)
args = parser.parse_args()
record(args.host, args.interval, args.outage_interval)
if __name__ == "__main__":
entry()

23
netmon/ping.py Normal file
View file

@ -0,0 +1,23 @@
import subprocess
class Ping:
host: str
_cmd: list[str]
def __init__(self, host: str):
self.host = host
self._cmd = ["/usr/bin/ping", "-c", "1", "-W", "1", "-qn", self.host]
def ping(self) -> bool:
try:
subprocess.run(
self._cmd,
shell=False,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return True
except subprocess.CalledProcessError:
return False

0
netmon/py.typed Normal file
View file