asyncprocess/tests/grep_tests.py

0001 # Description: Unit tests for the handler of the async grep requests. They run rg on log files
0002 # written to a temporary directory, so they are skipped where rg is not installed
0003 import gzip
0004 import json
0005 import os
0006 import shutil
0007 import tempfile
0008 import unittest
0009 from types import SimpleNamespace
0010 from typing import Any
0011 from unittest import mock
0012
0013 from pandaserver.asyncprocess import processor
0014 from pandaserver.config import daemon_config, panda_config
0015 from pandaserver.taskbuffer.db_proxy_mods.async_request_module import (
0016     SERVICE_JEDI,
0017     SERVICE_SERVER,
0018 )
0019
0020 # the lines of a log, a record of several lines among them
0021 RECORDS = [
0022     "2026-10-08 00:00:01,001 adder: DEBUG start job=7353014619\n",
0023     "2026-10-08 00:00:02,001 adder: ERROR failed job=7353014619\n",
0024     "Traceback (most recent call last):\n",
0025     '  File "/opt/ddm.py", line 522, in f\n',
0026     "    raise X\n",
0027     "ValueError: boom\n",
0028     "2026-10-08 00:00:03,001 adder: DEBUG other\n",
0029     "2026-10-08 00:00:04,001 adder: DEBUG done job=7353014619\n",
0030 ]
0031
0032
0033 def offset(first: int, lines: list[str] = RECORDS) -> int:
0034     """Where line first of the lines starts in their file."""
0035     return len("".join(lines[:first]))
0036
0037
0038 def record(first: int, end: int, lines: list[str] = RECORDS, head: bool = True, cut: bool = False) -> dict[str, Any]:
0039     """A record of whole_records made of lines first to end - 1 of the lines."""
0040     return {"offset": offset(first, lines), "head": head, "lines": [line.removesuffix("\n") for line in lines[first:end]], "cut": cut}
0041
0042
0043 @unittest.skipUnless(shutil.which("rg"), "rg is not installed")
0044 class TestHandleGrep(unittest.TestCase):
0045     def setUp(self) -> None:
0046         self.logdir = tempfile.mkdtemp()
0047         self.addCleanup(shutil.rmtree, self.logdir)
0048         patcher = mock.patch.object(panda_config, "logdir", self.logdir, create=True)
0049         patcher.start()
0050         self.addCleanup(patcher.stop)
0051
0052     def _write(self, name: str, text: str) -> None:
0053         with open(os.path.join(self.logdir, name), "w") as f:
0054             f.write(text)
0055
0056     def _grep(self, **params: Any) -> dict[str, Any]:
0057         """Run the handler on a request with these parameters, and give what it stored."""
0058         task_buffer = mock.MagicMock()
0059         row = {"request_id": "r1", "parameters": json.dumps(params)}
0060         processor._handle_grep(row, task_buffer, mock.MagicMock(), "host")
0061         args, kwargs = task_buffer.finish_async_result.call_args
0062         return {"status": args[2], **kwargs}
0063
0064     def test_pattern_starting_with_a_dash_is_searched_for(self) -> None:
0065         # in the place of an argument of its own, rg would read it as --pre and run the command
0066         # on every file under the working directory
0067         self._write("panda-a.log", "2026-10-08 00:00:01 opts --pre=true\n2026-10-08 00:00:02 other\n")
0068         out = self._grep(pattern="--pre=true", log_filename="panda-a.log")
0069         self.assertEqual(out["status"], "done")
0070         self.assertEqual(out["result"], "2026-10-08 00:00:01 opts --pre=true\n")
0071
0072     def test_pattern_means_the_same_in_a_gz_file(self) -> None:
0073         # "+" repeats in rg; zgrep read it as a plain character in its basic regular expressions
0074         text = "2026-10-08 00:00:01 aaab\n2026-10-08 00:00:02 a+b\n"
0075         self._write("panda-a.log", text)
0076         with gzip.open(os.path.join(self.logdir, "panda-a.log-20261008.gz"), "wt") as f:
0077             f.write(text)
0078         for name in ("panda-a.log", "panda-a.log-20261008.gz"):
0079             out = self._grep(pattern="a+b", log_filename=name)
0080             self.assertEqual(out["result"], "2026-10-08 00:00:01 aaab\n", name)
0081
0082     def test_gz_being_compressed(self) -> None:
0083         # logrotate removes the file it compresses once the .gz is complete
0084         self._write("panda-a.log-20261014", "".join(RECORDS))
0085         with gzip.open(os.path.join(self.logdir, "panda-a.log-20261014.gz"), "wt") as f:
0086             f.write("".join(RECORDS[:2]))
0087         out = self._grep(pattern="done", log_filename="panda-a.log-20261014.gz")
0088         self.assertEqual(out["status"], "failed")
0089         self.assertTrue(out["error_msg"].endswith(": Being compressed"), out["error_msg"])
0090         # the processor does not try again, as a compression takes longer than its attempts
0091         self.assertFalse(out["retriable"])
0092         os.remove(os.path.join(self.logdir, "panda-a.log-20261014"))
0093         self.assertEqual(self._grep(pattern="start", log_filename="panda-a.log-20261014.gz")["status"], "done")
0094
0095     def test_fixed_strings(self) -> None:
0096         # any of the strings matches, each as it is: "." and "+" are no wildcards here
0097         lines = [
0098             "2026-10-08 00:00:01 getJobs < SITE 2026-10-08T00:00:01 > start\n",
0099             "2026-10-08 00:00:02 a.b+c/d\n",
0100             "2026-10-08 00:00:03 axbbc/d\n",
0101             "2026-10-08 00:00:04 --pre=true\n",
0102         ]
0103         self._write("panda-a.log", "".join(lines))
0104         out = self._grep(fixed_strings=["getJobs < SITE 2026-10-08T00:00:01 >", "a.b+c/d", "--pre=true"], log_filename="panda-a.log")
0105         self.assertEqual(out["result"], lines[0] + lines[1] + lines[3])
0106
0107     def test_matching_lines_alone(self) -> None:
0108         self._write("panda-a.log", "".join(RECORDS))
0109         out = self._grep(pattern="7353014619", log_filename="panda-a.log")
0110         self.assertEqual(out["result"], RECORDS[0] + RECORDS[1] + RECORDS[7])
0111         self.assertEqual((out["return_code"], out["truncated"]), (0, False))
0112
0113     def test_whole_records(self) -> None:
0114         # each matching line comes with the lines of its record after it, up to the next stamp
0115         self._write("panda-a.log", "".join(RECORDS))
0116         with gzip.open(os.path.join(self.logdir, "panda-a.log-20261008.gz"), "wt") as f:
0117             f.write("".join(RECORDS))
0118         for name in ("panda-a.log", "panda-a.log-20261008.gz"):
0119             out = self._grep(pattern="7353014619", log_filename=name, whole_records=True)
0120             self.assertEqual(json.loads(out["result"]), {"records": [record(0, 1), record(1, 6), record(7, 8)], "window": None}, name)
0121             self.assertEqual((out["return_code"], out["truncated"]), (0, False), name)
0122
0123     def test_whole_records_matching_within_a_record(self) -> None:
0124         # a match in the traceback of a kept record adds no line twice, and a match in the
0125         # traceback of a record that does not match brings the lines after it, without a head
0126         self._write("panda-a.log", "".join(RECORDS))
0127         out = self._grep(pattern="ddm.py|ERROR failed", log_filename="panda-a.log", whole_records=True)
0128         self.assertEqual(json.loads(out["result"])["records"], [record(1, 6)])
0129         out = self._grep(pattern="raise X", log_filename="panda-a.log", whole_records=True)
0130         self.assertEqual(json.loads(out["result"])["records"], [record(4, 6, head=False)])
0131
0132     def test_whole_records_part_after_a_record(self) -> None:
0133         # a part of a record that does not match, right after one that does, stays apart from it
0134         lines = [
0135             "2026-10-08 00:00:01,001 adder: DEBUG start job=1\n",
0136             "2026-10-08 00:00:02,001 adder: ERROR failed\n",
0137             "Traceback (most recent call last):\n",
0138             "  File x, in f(job=1)\n",
0139             "ValueError: boom\n",
0140             "2026-10-08 00:00:03,001 adder: DEBUG other\n",
0141         ]
0142         self._write("panda-a.log", "".join(lines))
0143         out = self._grep(pattern="job=1", log_filename="panda-a.log", whole_records=True)
0144         self.assertEqual(
0145             json.loads(out["result"])["records"],
0146             [record(0, 1, lines), record(3, 5, lines, head=False)],
0147         )
0148
0149     def test_whole_records_cut(self) -> None:
0150         # a record longer than the most lines taken after its match is cut, and says so
0151         self._write("panda-a.log", "".join(RECORDS))
0152         with mock.patch.object(processor, "_MAX_CONTINUATION_LINES", 2):
0153             out = self._grep(pattern="ERROR failed", log_filename="panda-a.log", whole_records=True)
0154             self.assertEqual(json.loads(out["result"])["records"], [record(1, 4, cut=True)])
0155             self.assertTrue(out["truncated"])
0156         # a record of exactly as many lines as are taken is whole
0157         with mock.patch.object(processor, "_MAX_CONTINUATION_LINES", 4):
0158             out = self._grep(pattern="ERROR failed", log_filename="panda-a.log", whole_records=True)
0159             self.assertEqual(json.loads(out["result"])["records"], [record(1, 6)])
0160             self.assertFalse(out["truncated"])
0161
0162     def test_whole_records_in_a_tail_window(self) -> None:
0163         # the window cuts the line of ddm.py, so it starts with the next line; offsets are
0164         # those in the file, and the window says from which stamp records were searched whole
0165         self._write("panda-a.log", "".join(RECORDS))
0166         tail_bytes = len("".join(RECORDS[3:])) - 5
0167         out = self._grep(pattern="raise X|ddm.py|7353014619", log_filename="panda-a.log", whole_records=True, tail_bytes=tail_bytes)
0168         self.assertEqual(
0169             json.loads(out["result"]),
0170             {"records": [record(4, 6, head=False), record(7, 8)], "window": {"offset": offset(4), "first_stamp": "2026-10-08 00:00:03"}},
0171         )
0172
0173     def test_tail_window_starts_with_a_whole_line(self) -> None:
0174         self._write("panda-a.log", "".join(RECORDS))
0175         out = self._grep(pattern="ddm.py|raise X", log_filename="panda-a.log", tail_bytes=len("".join(RECORDS[3:])) - 5)
0176         self.assertEqual(out["result"], RECORDS[4])
0177         # a window that starts right after a line break takes the line there
0178         out = self._grep(pattern="ddm.py|raise X", log_filename="panda-a.log", tail_bytes=len("".join(RECORDS[3:])))
0179         self.assertEqual(out["result"], RECORDS[3] + RECORDS[4])
0180
0181     def test_whole_records_result_size_cap(self) -> None:
0182         # records are kept while they fit, and the result stays valid JSON
0183         lines = [f"2026-10-08 00:00:01 line {i:06d}\n" for i in range(1000)]
0184         self._write("panda-a.log", "".join(lines))
0185         with mock.patch.object(processor, "_MAX_RESULT_BYTES", 3000):
0186             out = self._grep(pattern="line", log_filename="panda-a.log", whole_records=True)
0187         self.assertTrue(out["truncated"])
0188         self.assertLessEqual(len(out["result"]), 3000)
0189         records = json.loads(out["result"])["records"]
0190         self.assertEqual(records, [record(i, i + 1, lines) for i in range(len(records))])
0191
0192     def test_result_size_cap(self) -> None:
0193         # the matcher is stopped once no more lines fit; the kept lines are whole, and the
0194         # return code still says that lines matched
0195         lines = [f"2026-10-08 00:00:01 line {i:06d}\n" for i in range(10000)]
0196         self._write("panda-a.log", "".join(lines))
0197         with mock.patch.object(processor, "_MAX_RESULT_BYTES", 1000):
0198             out = self._grep(pattern="line", log_filename="panda-a.log")
0199         self.assertTrue(out["truncated"])
0200         self.assertEqual(out["return_code"], 0)
0201         self.assertEqual(out["result"], "".join(lines[: 1000 // len(lines[0])]))
0202
0203     def test_max_matches(self) -> None:
0204         self._write("panda-a.log", "".join(RECORDS))
0205         out = self._grep(pattern="7353014619", log_filename="panda-a.log", max_matches=2)
0206         self.assertEqual(out["result"], RECORDS[0] + RECORDS[1])
0207         self.assertTrue(out["truncated"])
0208
0209     def test_no_match(self) -> None:
0210         self._write("panda-a.log", "".join(RECORDS))
0211         out = self._grep(pattern="nothing", log_filename="panda-a.log")
0212         self.assertEqual((out["status"], out["result"], out["return_code"], out["truncated"]), ("done", "", 1, False))
0213         out = self._grep(pattern="nothing", log_filename="panda-a.log", whole_records=True)
0214         self.assertEqual(json.loads(out["result"]), {"records": [], "window": None})
0215         self.assertEqual((out["return_code"], out["truncated"]), (1, False))
0216
0217     def test_timeout(self) -> None:
0218         self._write("panda-a.log", "".join(RECORDS))
0219         with (
0220             mock.patch.object(processor, "_bounded_matcher", return_value=["sleep", "10"]),
0221             mock.patch.object(processor, "_SUBPROCESS_TIMEOUT", 0.2),
0222         ):
0223             out = self._grep(pattern="x", log_filename="panda-a.log")
0224         self.assertEqual((out["status"], out["error_msg"]), ("failed", "timeout"))
0225
0226
0227 class TestHandleLogList(unittest.TestCase):
0228     def setUp(self) -> None:
0229         self.logdir = tempfile.mkdtemp()
0230         self.addCleanup(shutil.rmtree, self.logdir)
0231         patcher = mock.patch.object(panda_config, "logdir", self.logdir, create=True)
0232         patcher.start()
0233         self.addCleanup(patcher.stop)
0234
0235     def _list(self, parameters: dict[str, Any] | None = None) -> dict[str, Any]:
0236         """Run the handler, and give what it stored."""
0237         task_buffer = mock.MagicMock()
0238         processor._handle_log_list({"request_id": "r1", "parameters": json.dumps(parameters or {})}, task_buffer, mock.MagicMock(), "host")
0239         args, kwargs = task_buffer.finish_async_result.call_args
0240         return {"status": args[2], **kwargs}
0241
0242     def test_list(self) -> None:
0243         # the files a grep may read, rotated or not, and nothing else
0244         for name, text in (("panda-a.log", "xx"), ("panda-a.log-20261012", "x"), ("panda-a.log-20261008.gz", ""), ("httpd.log", "x")):
0245             path = os.path.join(self.logdir, name)
0246             with open(path, "w") as f:
0247                 f.write(text)
0248             # 2026-10-12 23:59:59 UTC
0249             os.utime(path, (1791849599, 1791849599))
0250         os.mkdir(os.path.join(self.logdir, "panda-dir"))
0251         out = self._list()
0252         self.assertEqual((out["status"], out["truncated"]), ("done", False))
0253         self.assertEqual(
0254             json.loads(out["result"]),
0255             [
0256                 {"name": "panda-a.log", "size": 2, "mtime": "2026-10-12 23:59:59"},
0257                 {"name": "panda-a.log-20261008.gz", "size": 0, "mtime": "2026-10-12 23:59:59"},
0258                 {"name": "panda-a.log-20261012", "size": 1, "mtime": "2026-10-12 23:59:59"},
0259             ],
0260         )
0261
0262     def test_log_names(self) -> None:
0263         # the files of the logs named and those rotated from them, and no log whose name only starts the same
0264         names = (
0265             "panda-a.log",
0266             "panda-a.log-20261012",
0267             "panda-a.log-20261008.gz",
0268             "panda-ab.log",
0269             "panda-ab.log-20261012.gz",
0270             "panda-b.log",
0271             "panda-b.log-20261012.gz",
0272         )
0273         for name in names:
0274             open(os.path.join(self.logdir, name), "w").close()
0275         out = self._list({"requester": "alice", "access": "owner", "log_names": ["panda-a.log", "panda-b.log-20261012.gz", "panda-missing.log"]})
0276         self.assertEqual((out["status"], out["truncated"]), ("done", False))
0277         self.assertEqual(
0278             [file["name"] for file in json.loads(out["result"])],
0279             ["panda-a.log", "panda-a.log-20261008.gz", "panda-a.log-20261012", "panda-b.log-20261012.gz"],
0280         )
0281
0282     def test_gz_being_compressed(self) -> None:
0283         # the .gz is left out until the file it is compressed from is gone, which is listed
0284         for name in ("panda-a.log", "panda-a.log-20261013.gz", "panda-a.log-20261014", "panda-a.log-20261014.gz"):
0285             open(os.path.join(self.logdir, name), "w").close()
0286         listed = [file["name"] for file in json.loads(self._list()["result"])]
0287         self.assertEqual(listed, ["panda-a.log", "panda-a.log-20261013.gz", "panda-a.log-20261014"])
0288         # also for the log named
0289         listed = [file["name"] for file in json.loads(self._list({"log_names": ["panda-a.log"]})["result"])]
0290         self.assertEqual(listed, ["panda-a.log", "panda-a.log-20261013.gz", "panda-a.log-20261014"])
0291         # and the .gz named alone, for which the file being compressed is listed
0292         listed = [file["name"] for file in json.loads(self._list({"log_names": ["panda-a.log-20261014.gz"]})["result"])]
0293         self.assertEqual(listed, ["panda-a.log-20261014"])
0294         os.remove(os.path.join(self.logdir, "panda-a.log-20261014"))
0295         listed = [file["name"] for file in json.loads(self._list({"log_names": ["panda-a.log-20261014.gz"]})["result"])]
0296         self.assertEqual(listed, ["panda-a.log-20261014.gz"])
0297
0298     def test_log_names_cut_short(self) -> None:
0299         # the list of the logs named is cut as the whole list is
0300         for i in range(100):
0301             open(os.path.join(self.logdir, f"panda-a.log-{20260101 + i}.gz"), "w").close()
0302         open(os.path.join(self.logdir, "panda-0.log"), "w").close()
0303         with mock.patch.object(processor, "_MAX_RESULT_BYTES", 1000):
0304             out = self._list({"log_names": ["panda-a.log"]})
0305         self.assertTrue(out["truncated"])
0306         files = json.loads(out["result"])
0307         self.assertEqual([file["name"] for file in files], [f"panda-a.log-{20260101 + i}.gz" for i in range(len(files))])
0308
0309     def test_list_cut_short(self) -> None:
0310         # files are listed while they fit, and the list stays valid JSON
0311         for i in range(100):
0312             open(os.path.join(self.logdir, f"panda-{i:03d}.log"), "w").close()
0313         with mock.patch.object(processor, "_MAX_RESULT_BYTES", 1000):
0314             out = self._list()
0315         self.assertTrue(out["truncated"])
0316         files = json.loads(out["result"])
0317         self.assertLessEqual(len(out["result"]), 1000)
0318         self.assertEqual([file["name"] for file in files], [f"panda-{i:03d}.log" for i in range(len(files))])
0319
0320     def test_no_logdir(self) -> None:
0321         with mock.patch.object(panda_config, "logdir", os.path.join(self.logdir, "missing")):
0322             out = self._list()
0323         self.assertEqual(out["status"], "failed")
0324
0325
0326 class TestGrepInCycle(unittest.TestCase):
0327     """A grep starts only when the cycle has time left for its longest run."""
0328
0329     DAEMON = {"module": "async_request_daemon", "period": 10, "timeout": 300}
0330
0331     def test_cycle_seconds(self) -> None:
0332         with mock.patch.object(daemon_config, "find_daemon", return_value=self.DAEMON):
0333             # the timeout less the seconds a result is stored in
0334             self.assertEqual(processor._cycle_seconds(SERVICE_SERVER), 270)
0335             # nothing kills a cycle of JEDI's watchdog
0336             self.assertIsNone(processor._cycle_seconds(SERVICE_JEDI))
0337         with mock.patch.object(daemon_config, "find_daemon", return_value=None):
0338             self.assertIsNone(processor._cycle_seconds(SERVICE_SERVER))
0339
0340     def test_fits_in_cycle(self) -> None:
0341         now = [1000.0]
0342         with mock.patch.object(processor, "time", SimpleNamespace(monotonic=lambda: now[0])):
0343             # 240 s from 30 s into a cycle of 270 s ends with it, a second later does not
0344             now[0] = 1030.0
0345             self.assertTrue(processor._fits_in_cycle("grep", 270, 1000.0))
0346             now[0] = 1031.0
0347             self.assertFalse(processor._fits_in_cycle("grep", 270, 1000.0))
0348             # a type that does not run long, a cycle nothing kills, and a cycle too short
0349             # for a grep at all start it as before
0350             self.assertTrue(processor._fits_in_cycle("log_list", 270, 1000.0))
0351             self.assertTrue(processor._fits_in_cycle("grep", None, 1000.0))
0352             self.assertTrue(processor._fits_in_cycle("grep", 30, 1000.0))
0353
0354     def test_run_leaves_a_grep_for_the_next_cycle(self) -> None:
0355         # the first grep takes 100 s, so the second would start 100 s into a cycle of 270 s
0356         now = [1000.0]
0357         ran: list[str] = []
0358
0359         def grep(row: dict[str, Any], tb: Any, tmp_logger: Any, result_machine: str) -> None:
0360             ran.append(row["request_id"])
0361             now[0] += 100
0362
0363         task_buffer = mock.MagicMock()
0364         task_buffer.get_pending_requests_for_machine.return_value = [
0365             {"request_id": request_id, "request_type": "grep", "machine_name": None, "parameters": "{}"} for request_id in ("r1", "r2")
0366         ]
0367         task_buffer.claim_async_result.return_value = True
0368         with (
0369             mock.patch.object(processor, "time", SimpleNamespace(monotonic=lambda: now[0])),
0370             mock.patch.dict(processor.HANDLERS, {"grep": grep}),
0371             mock.patch.object(daemon_config, "find_daemon", return_value=self.DAEMON),
0372             # run sets the start of the cycle, on the fake clock, where db_read reads it
0373             mock.patch.object(processor, "_cycle_started_at", None),
0374         ):
0375             processor.run(SERVICE_SERVER, task_buffer)
0376         self.assertEqual(ran, ["r1"])
0377         # not claimed, so it stays pending
0378         self.assertEqual([c.args[0] for c in task_buffer.claim_async_result.call_args_list], ["r1"])
0379
0380     def test_run_takes_a_grep_after_the_other_requests(self) -> None:
0381         # a db_read submitted after a grep still runs first, with the whole cycle before it,
0382         # rather than after the grep with what the grep left
0383         now = [1000.0]
0384         ran: list[tuple[str, float]] = []
0385
0386         def handle(row: dict[str, Any], tb: Any, tmp_logger: Any, result_machine: str) -> None:
0387             ran.append((row["request_id"], now[0]))
0388             now[0] += 100
0389
0390         task_buffer = mock.MagicMock()
0391         task_buffer.get_pending_requests_for_machine.return_value = [
0392             {"request_id": "r1", "request_type": "grep", "machine_name": None, "parameters": "{}"},
0393             {"request_id": "r2", "request_type": "db_read", "machine_name": "any", "parameters": "{}"},
0394             {"request_id": "r3", "request_type": "log_list", "machine_name": None, "parameters": "{}"},
0395         ]
0396         task_buffer.claim_async_result.return_value = True
0397         with (
0398             mock.patch.object(processor, "time", SimpleNamespace(monotonic=lambda: now[0])),
0399             mock.patch.dict(processor.HANDLERS, {"grep": handle, "db_read": handle, "log_list": handle}),
0400             mock.patch.object(daemon_config, "find_daemon", return_value=self.DAEMON),
0401             mock.patch.object(processor, "_cycle_started_at", None),
0402         ):
0403             processor.run(SERVICE_SERVER, task_buffer)
0404         # the others in the order they came, and then no grep, since 200 s of the cycle are gone
0405         self.assertEqual(ran, [("r2", 1000.0), ("r3", 1100.0)])
0406
0407
0408 if __name__ == "__main__":
0409     unittest.main()