40b2742456
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 1m7s
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import importlib.util
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT_ROOT = Path(__file__).resolve().parents[1]
|
|
EXPORTER_PATH = SCRIPT_ROOT / "systemd-units-textfile-exporter.py"
|
|
|
|
|
|
def load_exporter():
|
|
spec = importlib.util.spec_from_file_location("systemd_units_textfile_exporter", EXPORTER_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_stops_probe_after_systemctl_timeout_budget(monkeypatch) -> None:
|
|
exporter = load_exporter()
|
|
monkeypatch.setattr(exporter, "UNIT_NAMES", ["runner-a.service", "runner-b.service", "runner-c.service"])
|
|
monkeypatch.setenv("AIOPS_SYSTEMD_MAX_TIMEOUTS", "1")
|
|
calls: list[str] = []
|
|
|
|
def fake_show_unit(unit: str) -> dict[str, str]:
|
|
calls.append(unit)
|
|
raise subprocess.TimeoutExpired(cmd=["systemctl", "show", unit], timeout=2)
|
|
|
|
monkeypatch.setattr(exporter, "_show_unit", fake_show_unit)
|
|
|
|
metrics = exporter.collect()
|
|
|
|
assert calls == ["runner-a.service"]
|
|
assert 'unit="runner-a.service",active_state="scrape_error",sub_state="timeout"' in metrics
|
|
assert (
|
|
'unit="runner-b.service",active_state="scrape_skipped",'
|
|
'sub_state="systemctl_timeout_budget_exhausted"'
|
|
) in metrics
|
|
assert (
|
|
'unit="runner-c.service",active_state="scrape_skipped",'
|
|
'sub_state="systemctl_timeout_budget_exhausted"'
|
|
) in metrics
|
|
assert "Command '['" not in metrics
|
|
assert "systemd_unit_exporter_timeout_budget_exhausted" in metrics
|
|
assert metrics.rstrip().endswith(" 1")
|
|
|
|
|
|
def test_main_returns_when_exporter_lock_is_busy(tmp_path: Path, monkeypatch) -> None:
|
|
exporter = load_exporter()
|
|
lock_path = tmp_path / "systemd_units.prom.lock"
|
|
output_dir = tmp_path / "textfiles"
|
|
monkeypatch.setattr(exporter, "LOCK_PATH", lock_path)
|
|
monkeypatch.setattr(exporter, "TEXTFILE_DIR", output_dir)
|
|
monkeypatch.setattr(exporter, "UNIT_NAMES", ["runner-a.service"])
|
|
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with lock_path.open("w") as lock_handle:
|
|
fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
|
|
exporter.main()
|
|
|
|
assert not (output_dir / exporter.OUTPUT_NAME).exists()
|