From 477741fa41836350c60eb947e3a927502a0e8f3c Mon Sep 17 00:00:00 2001 From: Nina Date: Tue, 25 Aug 2026 19:29:32 -0400 Subject: [PATCH 1/4] initial (rought) attempt to report warnings Preliminary attempt at adding a warnigns section to the report dashboard. If any of the successful files contained warnings, this section will print (per task, the total num warnings, min, max, and average). Changes include: * FileMetics updated to have default values (revisit) * _parse_all_metrics updated to search for warnings (edge cases) * _compute_warning_summary added Things that need to be cleaned up: * Clarify if warnings should be counted if file failed * Add path to log file * Clean up printing (ie. formatting and decimal points) * Address edge cases in models.py code --- src/tigerflow/cli/report.py | 47 +++++++++++++++++++++++++++++++++++++ src/tigerflow/models.py | 45 +++++++++++++++++++---------------- 2 files changed, 72 insertions(+), 20 deletions(-) diff --git a/src/tigerflow/cli/report.py b/src/tigerflow/cli/report.py index 10e07cf..c798350 100644 --- a/src/tigerflow/cli/report.py +++ b/src/tigerflow/cli/report.py @@ -74,6 +74,37 @@ def _compute_metrics_summary(metrics: dict[str, list[FileMetrics]]) -> dict: } +def _compute_warning_summary(metrics: dict[str, list[FileMetrics]]) -> dict: + """Compute warning totals from metrcis""" + warnings = {} + for task, task_metrics in metrics.items(): + if task not in warnings.keys(): + warnings[task] = { + "total_num_warnings": 0, + "min_num_warnings": 0, + "max_num_warnings": 0, + "total_file_num": 0, + } + for m in task_metrics: + warnings[task]["total_num_warnings"] = ( + warnings[task]["total_num_warnings"] + m.num_warnings + ) + warnings[task]["min_num_warnings"] = min( + warnings[task]["min_num_warnings"], m.num_warnings + ) + warnings[task]["max_num_warnings"] = max( + warnings[task]["max_num_warnings"], m.num_warnings + ) + warnings[task]["total_file_num"] = warnings[task]["total_file_num"] + 1 + + for task in warnings.keys(): + warnings[task]["avg_num_warnings"] = ( + warnings[task]["total_num_warnings"] / warnings[task]["total_file_num"] + ) + + return warnings + + def _build_dashboard_panel(report: PipelineReport) -> Panel: """Build the dashboard panel.""" @@ -182,6 +213,22 @@ def fmt_duration(ms: float) -> str: lines.append(f" [dim]... +{total_errors - 5} more[/dim]") lines.append("") + # Warnings summary (for this run) + total_warnings = sum( + m.num_warnings + for metrics in report.metrics.values() + for m in metrics + if m.status == "success" # only counts warnings for successful files (?) + ) + if total_warnings > 0: + lines.append(f"[bold]Warnings:[/bold] {total_warnings}") + warning_summary = _compute_warning_summary(report.metrics) + for task_name, warning_data in warning_summary.items(): + if warning_data["total_num_warnings"] > 0: + lines.append( + f" [dim]{task_name}[/dim] [yellow]{warning_data['total_num_warnings']} warnings ({warning_data['min_num_warnings']}-{warning_data['max_num_warnings']}, avg={warning_data['avg_num_warnings']})[/yellow]" + ) + content = "\n".join(lines) return Panel(content, title="[bold]tigerflow report[/bold]", title_align="left") diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index 12504fb..eaf952e 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -38,11 +38,12 @@ def is_alive(self) -> bool: class FileMetrics(BaseModel): """Timing metrics for a single file processed by a task.""" - file: str + file: str = "" task: str - started_at: datetime - finished_at: datetime - status: Literal["success", "error"] + started_at: datetime = datetime.now() + finished_at: datetime = datetime.now() + status: Literal["success", "error", "pending"] = "pending" + num_warnings: int = 0 @property def duration_ms(self) -> float: @@ -570,26 +571,30 @@ def _parse_all_metrics(self) -> list[FileMetrics]: for log_file in log_files: try: with open(log_file) as f: + current_file = "" for line in f: - if "METRICS" not in line: + if "| INFO | Starting processing: " in line: + current_file = FileMetrics(task=task_dir.name) # Update + if not current_file: continue - start = line.find("{") - if start == -1: + if "| WARNING |" in line: + current_file.num_warnings += 1 continue - data = json.loads(line[start:]) - metrics.append( - FileMetrics( - file=data["file"], - task=task_dir.name, - started_at=datetime.fromisoformat( - data["started_at"] - ), - finished_at=datetime.fromisoformat( - data["finished_at"] - ), - status=data["status"], + if "| METRICS |" in line: + start = line.find("{") + if start == -1: + continue + data = json.loads(line[start:]) + current_file.file = data["file"] + current_file.started_at = datetime.fromisoformat( + data["started_at"] ) - ) + current_file.finished_at = datetime.fromisoformat( + data["finished_at"] + ) + current_file.status = data["status"] + metrics.append(current_file) + current_file = "" except (OSError, json.JSONDecodeError, KeyError): continue From c4df6c2d19fedaa251bbfa7ecae8ba851ba71c5e Mon Sep 17 00:00:00 2001 From: Nina Date: Wed, 26 Aug 2026 11:59:16 -0400 Subject: [PATCH 2/4] update tests to reflect new log parsing strategy --- tests/unit/cli/test_report.py | 39 +++++++++++++++++++++++++++-------- 1 file changed, 30 insertions(+), 9 deletions(-) diff --git a/tests/unit/cli/test_report.py b/tests/unit/cli/test_report.py index f842379..dec31f9 100644 --- a/tests/unit/cli/test_report.py +++ b/tests/unit/cli/test_report.py @@ -248,7 +248,10 @@ def test_parse_all_metrics(self, tmp_path: Path): all_lines = [] for i in range(5): all_lines.append( - f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" + f"2026-03-10 10:00:01 | INFO | Starting processing: file{i}.txt" + ) + all_lines.append( + f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) (log_dir / "task-12345.log").write_text("\n".join(all_lines)) @@ -302,7 +305,10 @@ def test_task_progress_with_symlinks(self, tmp_path: Path): for i in range(8): status = "success" if i < 6 else "error" task_lines.append( - f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" + f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" + ) + task_lines.append( + f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" ) (log_dir / "task-100.log").write_text("\n".join(task_lines)) @@ -351,7 +357,10 @@ def test_multi_task_dependency_chain(self, tmp_path: Path): for i in range(10): status = "success" if i < 8 else "error" lines1.append( - f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" + f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" + ) + lines1.append( + f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" ) (log_dir1 / "task-100.log").write_text("\n".join(lines1)) @@ -364,7 +373,10 @@ def test_multi_task_dependency_chain(self, tmp_path: Path): lines2 = [] for i in range(5): lines2.append( - f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" + f"2026-03-10 12:00:02 | INFO | Starting processing: file{i}.txt" + ) + lines2.append( + f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) (log_dir2 / "task-100.log").write_text("\n".join(lines2)) @@ -442,7 +454,10 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path): run1_lines = [] for i in range(5): run1_lines.append( - f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'run1_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" + f"2026-03-10 10:00:01 | INFO | Starting processing: file{i}.txt" + ) + run1_lines.append( + f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'run1_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) (finished / f"run1_file{i}.txt").touch() (log_dir1 / "task-100.log").write_text("\n".join(run1_lines)) @@ -453,7 +468,10 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path): run2_lines = [] for i in range(7): run2_lines.append( - f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'run2_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" + f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" + ) + run2_lines.append( + f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'run2_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) (finished / f"run2_file{i}.txt").touch() (log_dir2 / "task-200.log").write_text("\n".join(run2_lines)) @@ -488,14 +506,17 @@ def test_parse_metrics_from_multiple_worker_logs(self, tmp_path: Path): # Worker 1 processed file1 worker1_log = log_dir / "task-worker-12345.log" worker1_log.write_text( - f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': 'file1.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" + f"2026-03-10 12:00:01 | INFO | Starting processing: file1.txt\n" + f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': 'file1.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" ) # Worker 2 processed file2 and file3 worker2_log = log_dir / "task-worker-12346.log" worker2_log.write_text( - f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': 'file2.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" - f"2026-03-10 12:00:03 | METRICS | {json.dumps({'file': 'file3.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'error'})}\n" + f"2026-03-10 12:00:02 | INFO | Starting processing: file2.txt\n" + f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': 'file2.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" + f"2026-03-10 12:00:03 | INFO | Starting processing: file3.txt\n" + f"2026-03-10 12:00:03 | METRICS | {json.dumps({'file': 'file3.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'error'})}\n" ) output = PipelineOutput(tmp_path) From 2db9813a545ce8cb08419ed70d5cd9a7b6fff511 Mon Sep 17 00:00:00 2001 From: Nina Date: Wed, 26 Aug 2026 12:00:00 -0400 Subject: [PATCH 3/4] changes print format and includes failed files in total count --- src/tigerflow/cli/report.py | 27 ++++++++++++--------------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/src/tigerflow/cli/report.py b/src/tigerflow/cli/report.py index c798350..61f559a 100644 --- a/src/tigerflow/cli/report.py +++ b/src/tigerflow/cli/report.py @@ -89,18 +89,18 @@ def _compute_warning_summary(metrics: dict[str, list[FileMetrics]]) -> dict: warnings[task]["total_num_warnings"] = ( warnings[task]["total_num_warnings"] + m.num_warnings ) - warnings[task]["min_num_warnings"] = min( - warnings[task]["min_num_warnings"], m.num_warnings - ) - warnings[task]["max_num_warnings"] = max( - warnings[task]["max_num_warnings"], m.num_warnings - ) + # warnings[task]["min_num_warnings"] = min( + # warnings[task]["min_num_warnings"], m.num_warnings + # ) + # warnings[task]["max_num_warnings"] = max( + # warnings[task]["max_num_warnings"], m.num_warnings + # ) warnings[task]["total_file_num"] = warnings[task]["total_file_num"] + 1 - for task in warnings.keys(): - warnings[task]["avg_num_warnings"] = ( - warnings[task]["total_num_warnings"] / warnings[task]["total_file_num"] - ) + # for task in warnings.keys(): + # warnings[task]["avg_num_warnings"] = ( + # warnings[task]["total_num_warnings"] / warnings[task]["total_file_num"] + # ) return warnings @@ -215,10 +215,7 @@ def fmt_duration(ms: float) -> str: # Warnings summary (for this run) total_warnings = sum( - m.num_warnings - for metrics in report.metrics.values() - for m in metrics - if m.status == "success" # only counts warnings for successful files (?) + m.num_warnings for metrics in report.metrics.values() for m in metrics ) if total_warnings > 0: lines.append(f"[bold]Warnings:[/bold] {total_warnings}") @@ -226,7 +223,7 @@ def fmt_duration(ms: float) -> str: for task_name, warning_data in warning_summary.items(): if warning_data["total_num_warnings"] > 0: lines.append( - f" [dim]{task_name}[/dim] [yellow]{warning_data['total_num_warnings']} warnings ({warning_data['min_num_warnings']}-{warning_data['max_num_warnings']}, avg={warning_data['avg_num_warnings']})[/yellow]" + f" [dim]{task_name}[/dim] [yellow]{warning_data['total_num_warnings']} warnings across {warning_data['total_file_num']} file(s)[/yellow] See logs in .tigerflow/{task_name} for more details" ) content = "\n".join(lines) From 2a64bd4c1157efe6b5792fdeb70b4e81eda82a99 Mon Sep 17 00:00:00 2001 From: Nina Date: Wed, 26 Aug 2026 15:48:44 -0400 Subject: [PATCH 4/4] Moves warning info out of FileMetrics Now, the warnigng information is stored in the format: {"task_name":{"num_files_processed":0, "num_warnings":0}. Instead of the warnings being associated with specific files, they are simply summed at the task level. This means concurrently running files will not cause issues and is overall less brittle than the previous approach. --- src/tigerflow/cli/report.py | 40 ++-------------- src/tigerflow/models.py | 88 +++++++++++++++++++++++++---------- tests/unit/cli/test_report.py | 21 --------- 3 files changed, 67 insertions(+), 82 deletions(-) diff --git a/src/tigerflow/cli/report.py b/src/tigerflow/cli/report.py index 61f559a..1ea9e75 100644 --- a/src/tigerflow/cli/report.py +++ b/src/tigerflow/cli/report.py @@ -74,37 +74,6 @@ def _compute_metrics_summary(metrics: dict[str, list[FileMetrics]]) -> dict: } -def _compute_warning_summary(metrics: dict[str, list[FileMetrics]]) -> dict: - """Compute warning totals from metrcis""" - warnings = {} - for task, task_metrics in metrics.items(): - if task not in warnings.keys(): - warnings[task] = { - "total_num_warnings": 0, - "min_num_warnings": 0, - "max_num_warnings": 0, - "total_file_num": 0, - } - for m in task_metrics: - warnings[task]["total_num_warnings"] = ( - warnings[task]["total_num_warnings"] + m.num_warnings - ) - # warnings[task]["min_num_warnings"] = min( - # warnings[task]["min_num_warnings"], m.num_warnings - # ) - # warnings[task]["max_num_warnings"] = max( - # warnings[task]["max_num_warnings"], m.num_warnings - # ) - warnings[task]["total_file_num"] = warnings[task]["total_file_num"] + 1 - - # for task in warnings.keys(): - # warnings[task]["avg_num_warnings"] = ( - # warnings[task]["total_num_warnings"] / warnings[task]["total_file_num"] - # ) - - return warnings - - def _build_dashboard_panel(report: PipelineReport) -> Panel: """Build the dashboard panel.""" @@ -215,15 +184,14 @@ def fmt_duration(ms: float) -> str: # Warnings summary (for this run) total_warnings = sum( - m.num_warnings for metrics in report.metrics.values() for m in metrics + task_warnings["num_warnings"] for task_warnings in report.num_warnings.values() ) if total_warnings > 0: lines.append(f"[bold]Warnings:[/bold] {total_warnings}") - warning_summary = _compute_warning_summary(report.metrics) - for task_name, warning_data in warning_summary.items(): - if warning_data["total_num_warnings"] > 0: + for task_name, warning_data in report.num_warnings.items(): + if warning_data["num_warnings"] > 0: lines.append( - f" [dim]{task_name}[/dim] [yellow]{warning_data['total_num_warnings']} warnings across {warning_data['total_file_num']} file(s)[/yellow] See logs in .tigerflow/{task_name} for more details" + f" [dim]{task_name}[/dim] [yellow]{warning_data['num_warnings']} warnings across {warning_data['num_files_processed']} file(s)[/yellow] See logs in .tigerflow/{task_name} for more details" ) content = "\n".join(lines) diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index eaf952e..4110656 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -38,12 +38,11 @@ def is_alive(self) -> bool: class FileMetrics(BaseModel): """Timing metrics for a single file processed by a task.""" - file: str = "" + file: str task: str - started_at: datetime = datetime.now() - finished_at: datetime = datetime.now() - status: Literal["success", "error", "pending"] = "pending" - num_warnings: int = 0 + started_at: datetime + finished_at: datetime + status: Literal["success", "error"] @property def duration_ms(self) -> float: @@ -492,6 +491,7 @@ class PipelineReport(BaseModel): staged: int | None = None # None if stopped tasks: list[TaskProgress] = [] metrics: dict[str, list[FileMetrics]] = {} + num_warnings: dict[str, dict[str, int]] = {} errors: dict[str, list[FileError]] = {} @@ -571,35 +571,71 @@ def _parse_all_metrics(self) -> list[FileMetrics]: for log_file in log_files: try: with open(log_file) as f: - current_file = "" for line in f: - if "| INFO | Starting processing: " in line: - current_file = FileMetrics(task=task_dir.name) # Update - if not current_file: + if "METRICS" not in line: continue - if "| WARNING |" in line: - current_file.num_warnings += 1 + start = line.find("{") + if start == -1: continue - if "| METRICS |" in line: - start = line.find("{") - if start == -1: - continue - data = json.loads(line[start:]) - current_file.file = data["file"] - current_file.started_at = datetime.fromisoformat( - data["started_at"] - ) - current_file.finished_at = datetime.fromisoformat( - data["finished_at"] + data = json.loads(line[start:]) + metrics.append( + FileMetrics( + file=data["file"], + task=task_dir.name, + started_at=datetime.fromisoformat( + data["started_at"] + ), + finished_at=datetime.fromisoformat( + data["finished_at"] + ), + status=data["status"], ) - current_file.status = data["status"] - metrics.append(current_file) - current_file = "" + ) except (OSError, json.JSONDecodeError, KeyError): continue return metrics + def _parse_warnings(self) -> dict[str, dict[str, int]]: + """Parse WARNINGS from task log files. + + Reads from: + - {task}/logs/{pid}/task-{pid}.log (local/local_async tasks) + - {task}/logs/{pid}/task-worker-{job_id}.log (Slurm worker logs) + """ + all_warnings = {} + + for task_dir in self._get_task_dirs(): + log_files = list(task_dir.glob("logs/**/task*.log")) + + for log_file in log_files: + if task_dir.name not in all_warnings: + all_warnings[task_dir.name] = { + "num_files_processed": 0, + "num_warnings": 0, + } + try: + with open(log_file) as f: + for line in f: + if "METRICS" in line: + all_warnings[task_dir.name]["num_files_processed"] = ( + all_warnings[task_dir.name]["num_files_processed"] + + 1 + ) + if "WARNING" in line: + if re.search( + r"Received signal \d+, initiating shutdown", + line, + ): + continue + all_warnings[task_dir.name]["num_warnings"] = ( + all_warnings[task_dir.name]["num_warnings"] + 1 + ) + + except OSError: + continue + return all_warnings + def report(self) -> PipelineReport: """Generate a complete pipeline status report.""" self.validate() @@ -672,6 +708,7 @@ def report(self) -> PipelineReport: # === Per-Task Progress (from METRICS logs, all runs) === all_metrics = self._parse_all_metrics() + warnings = self._parse_warnings() task_meta = self._get_task_meta() # Group metrics by task @@ -726,5 +763,6 @@ def report(self) -> PipelineReport: staged=len(staged_stems) if is_running else None, tasks=tasks, metrics=metrics_by_task, + num_warnings=warnings, errors=errors, ) diff --git a/tests/unit/cli/test_report.py b/tests/unit/cli/test_report.py index dec31f9..2ec0700 100644 --- a/tests/unit/cli/test_report.py +++ b/tests/unit/cli/test_report.py @@ -247,9 +247,6 @@ def test_parse_all_metrics(self, tmp_path: Path): # All metrics appended to single task log all_lines = [] for i in range(5): - all_lines.append( - f"2026-03-10 10:00:01 | INFO | Starting processing: file{i}.txt" - ) all_lines.append( f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) @@ -304,9 +301,6 @@ def test_task_progress_with_symlinks(self, tmp_path: Path): task_lines = [] for i in range(8): status = "success" if i < 6 else "error" - task_lines.append( - f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" - ) task_lines.append( f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" ) @@ -356,9 +350,6 @@ def test_multi_task_dependency_chain(self, tmp_path: Path): lines1 = [] for i in range(10): status = "success" if i < 8 else "error" - lines1.append( - f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" - ) lines1.append( f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': status})}" ) @@ -372,9 +363,6 @@ def test_multi_task_dependency_chain(self, tmp_path: Path): log_dir2.mkdir(parents=True) lines2 = [] for i in range(5): - lines2.append( - f"2026-03-10 12:00:02 | INFO | Starting processing: file{i}.txt" - ) lines2.append( f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': f'file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) @@ -453,9 +441,6 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path): log_dir1.mkdir(parents=True) run1_lines = [] for i in range(5): - run1_lines.append( - f"2026-03-10 10:00:01 | INFO | Starting processing: file{i}.txt" - ) run1_lines.append( f"2026-03-10 10:00:01 | METRICS | {json.dumps({'file': f'run1_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) @@ -467,9 +452,6 @@ def test_task_progress_aggregates_across_runs(self, tmp_path: Path): log_dir2.mkdir(parents=True) run2_lines = [] for i in range(7): - run2_lines.append( - f"2026-03-10 12:00:01 | INFO | Starting processing: file{i}.txt" - ) run2_lines.append( f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': f'run2_file{i}.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}" ) @@ -506,16 +488,13 @@ def test_parse_metrics_from_multiple_worker_logs(self, tmp_path: Path): # Worker 1 processed file1 worker1_log = log_dir / "task-worker-12345.log" worker1_log.write_text( - f"2026-03-10 12:00:01 | INFO | Starting processing: file1.txt\n" f"2026-03-10 12:00:01 | METRICS | {json.dumps({'file': 'file1.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" ) # Worker 2 processed file2 and file3 worker2_log = log_dir / "task-worker-12346.log" worker2_log.write_text( - f"2026-03-10 12:00:02 | INFO | Starting processing: file2.txt\n" f"2026-03-10 12:00:02 | METRICS | {json.dumps({'file': 'file2.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'success'})}\n" - f"2026-03-10 12:00:03 | INFO | Starting processing: file3.txt\n" f"2026-03-10 12:00:03 | METRICS | {json.dumps({'file': 'file3.txt', 'started_at': now.isoformat(), 'finished_at': now.isoformat(), 'status': 'error'})}\n" )