cyberes
/
icinga2-checks
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

check_systemd_timer: handle running timer

This commit is contained in:
Cyberes 2024-10-10 20:55:42 -06:00
parent 57a8b24fc3
commit 02164d257a
1 changed files with 17 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
check_systemd_timer.py
View File

@ -18,12 +18,12 @@ from checker.result import quit_check
sys.path.insert(0, "/usr/lib/python3/dist-packages") sys.path.insert(0, "/usr/lib/python3/dist-packages")
import dbus import dbus
SYSTEMCTL_STATUS_RE = r'Loaded:\s.*\.timer;\s(.*?);.*?\)\s*Active:\s(.*?) since.*?\s*Trigger:\s(.*?);' SYSTEMCTL_STATUS_RE = r'Loaded:\s.*\.timer;\s(.*?);.*?\)\s*Active:\s(.*?) since.*?\s*Trigger:\s(.*?;|n\/a)'
class TimerInfo(BaseModel): class TimerInfo(BaseModel):
next: datetime next: Union[datetime, None]
time_left: timedelta time_left: Union[timedelta, None]
last: Union[datetime, None] last: Union[datetime, None]
since_last: Union[timedelta, None] since_last: Union[timedelta, None]
unit: str unit: str
@ -49,18 +49,24 @@ def get_last_trigger(timer_name: str):
def get_next_elapse(timer_name: str) -> Tuple[TimerInfo | None, None | str]: def get_next_elapse(timer_name: str) -> Tuple[TimerInfo | None, None | str]:
now = datetime.now()
try: try:
output = subprocess.check_output(["systemctl", "status", timer_name], universal_newlines=True) output = subprocess.check_output(["systemctl", "status", timer_name], universal_newlines=True)
if timer_name in output.split('\n')[0]: if timer_name in output.split('\n')[0]:
try: try:
parts = re.search(SYSTEMCTL_STATUS_RE, output) parts = re.search(SYSTEMCTL_STATUS_RE, output)
next_trigger = parse(parts.group(3)) next_trigger_str = parts.group(3)
now = datetime.now(tz=next_trigger.tzinfo) if next_trigger_str.lower() != 'n/a':
time_left = next_trigger - now next_trigger = parse(next_trigger_str)
time_left = next_trigger - now.replace(tzinfo=next_trigger.tzinfo)
else:
next_trigger = None
time_left = None
last_trigger = get_last_trigger(timer_name) last_trigger = get_last_trigger(timer_name)
if last_trigger is not None: if last_trigger is not None:
since_last = now - last_trigger since_last = now.replace(tzinfo=last_trigger.tzinfo) - last_trigger
else: else:
since_last = None since_last = None
@ -104,18 +110,18 @@ def check_timer(timer_name: str, expected_interval: int = None):
# if (next_elapse.left is not None and next_elapse.passed is not None) and (next_elapse.left < 0 or next_elapse.passed < 0): # if (next_elapse.left is not None and next_elapse.passed is not None) and (next_elapse.left < 0 or next_elapse.passed < 0):
# quit_check(f'Timer is negative??? Left: {next_elapse["left"]}. Passed: {next_elapse["passed"]}', nagios.STATE_UNKNOWN) # quit_check(f'Timer is negative??? Left: {next_elapse["left"]}. Passed: {next_elapse["passed"]}', nagios.STATE_UNKNOWN)
next_elapse_human = next_elapse.next.replace(tzinfo=tz.tzlocal()).strftime('%a %Y-%m-%d %H:%M %Z') next_elapse_human = next_elapse.next.replace(tzinfo=tz.tzlocal()).strftime('%a %Y-%m-%d %H:%M %Z') if next_elapse.next else 'N/A'
remaining_time_human = humanfriendly.format_timespan(next_elapse.time_left) remaining_time_human = humanfriendly.format_timespan(next_elapse.time_left) if next_elapse.time_left else 'N/A'
since_last_human = humanfriendly.format_timespan(next_elapse.since_last) if next_elapse.since_last else 'N/A' since_last_human = humanfriendly.format_timespan(next_elapse.since_last) if next_elapse.since_last else 'N/A'
perfdata_dict = { perfdata_dict = {
'remaining_time': { 'remaining_time': {
'value': next_elapse.time_left.seconds, 'value': next_elapse.time_left.seconds if next_elapse.time_left else 0,
'unit': 's', 'unit': 's',
'min': 0 'min': 0
}, },
'time_since_last': { 'time_since_last': {
'value': next_elapse.since_last.seconds if next_elapse.since_last is not None else 0, 'value': next_elapse.since_last.seconds if next_elapse.since_last else 0,
'unit': 's', 'unit': 's',
'min': 0 'min': 0
} }