Skip to content

Index

Job monitor subpackage. Provides log utilities, failure pattern detection, and the JobMonitorServer.

Modules:

Name Description
failure_patterns

Defines known failure signatures and maps them to structured classifications.
log_utils

Utilities for reading logs, extracting tails, combining multiple logs, and classifying failures.
server

Implements the JobMonitorServer MCP interface and exposes summarize_status and related tools.

JobMonitorServer

Bases: BaseMCPServer

MCP server that provides tools for job log inspection, failure detection, and diagnostic summarization.

The server exposes tools that read stdout and stderr logs, extract tail segments, and classify failures using the regex-based patterns defined in the job_monitor.failure_patterns module.

Attributes:

Name Type Description
tail_bytes int

Maximum number of bytes to read from the end of each log file. Used to avoid sending excessive log content to the client. Default is obtained from the MONITOR_LOG_TAIL_BYTES environment variable or falls back to 50000.
status_depth int

Number of historical status entries to inspect when summarizing job state. Default is obtained from the MONITOR_DEFAULT_STATUS_DEPTH environment variable or falls back to 20.

Methods:

Name Description
summarize_status

Inspect logs, classify failures, and return a structured summary suitable for planning or diagnosis.
read_logs

Return the tailed stdout and stderr logs for a specific job.
Source code in src/mada_tools/monitor/job_monitor/server.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class JobMonitorServer(BaseMCPServer):
    """
    MCP server that provides tools for job log inspection, failure detection,
    and diagnostic summarization.

    The server exposes tools that read stdout and stderr logs, extract tail
    segments, and classify failures using the regex-based patterns defined in
    the job_monitor.failure_patterns module.

    Attributes:
        tail_bytes (int): Maximum number of bytes to read from the end of each
            log file. Used to avoid sending excessive log content to the client.
            Default is obtained from the MONITOR_LOG_TAIL_BYTES environment
            variable or falls back to 50000.

        status_depth (int): Number of historical status entries to inspect when
            summarizing job state. Default is obtained from the
            MONITOR_DEFAULT_STATUS_DEPTH environment variable or falls back
            to 20.

    Methods:
        summarize_status: Inspect logs, classify failures, and return a
            structured summary suitable for planning or diagnosis.
        read_logs: Return the tailed stdout and stderr logs for a specific job.
    """

    def __init__(self):
        super().__init__(
            "Job Monitor",
            "Tools for log inspection, failure detection, and job diagnostics.",
        )

        self.tail_bytes = int(self.get_env_var("MONITOR_LOG_TAIL_BYTES", 50000))
        self.status_depth = int(self.get_env_var("MONITOR_DEFAULT_STATUS_DEPTH", 20))

    def _register_tools(self):
        """
        Register all MCP tools exposed by the Job Monitor Server.

        Tools defined:
            read_logs: Return tailed stdout and stderr content.
            summarize_status: Produce a structured diagnostic summary.
        """

        @self.mcp.tool()
        def read_logs(run_location: str, stdout_file: str, stderr_file: str) -> str:
            """
            Read and return the tailed stdout and stderr logs for a job.

            Args:
                run_location (str): Path to the directory containing output logs.
                stdout_file (str): Name of the stdout file to read.
                stderr_file (str): Name of the stderr file to read.

            Returns:
                str: JSON-encoded object containing:
                    - stdout (str): Tail of the stdout log.
                    - stderr (str): Tail of the stderr log.

            Raises:
                ToolExecutionError: If log reading fails for any reason.
            """
            try:
                out_path = os.path.join(run_location, stdout_file)
                err_path = os.path.join(run_location, stderr_file)

                result = {
                    "stdout": tail_log(out_path, self.tail_bytes),
                    "stderr": tail_log(err_path, self.tail_bytes),
                }
                return json.dumps(result, indent=2)
            except Exception as e:
                raise ToolExecutionError(f"Failed to read logs: {e}")

        @self.mcp.tool()
        def summarize_status(
            run_location: str,
            stdout_file: str,
            stderr_file: str,
            exit_code: Optional[int] = None,
        ) -> str:
            """
            Produce a structured job status summary including failure analysis.

            This tool reads the tails of stdout and stderr, concatenates them,
            classifies any detected failures, and returns a JSON-encoded
            diagnostic bundle. Used by the Job Manager or other agents to
            determine job health and identify root causes of failures.

            Args:
                run_location (str): Directory containing output logs.
                stdout_file (str): Name of the stdout file to read.
                stderr_file (str): Name of the stderr file to read.
                exit_code (Optional[int]): Exit code from the scheduler or runtime,
                    if available.

            Returns:
                str: JSON-encoded dictionary containing:
                    - exit_code (int or None)
                    - stdout_tail (str): Tail of the stdout log.
                    - stderr_tail (str): Tail of the stderr log.
                    - detected_failures (List[Dict]):
                        List of matched failure signatures or a fallback
                        unclassified entry with excerpt context.

            Notes:
                The classification logic is defined in failure_patterns.py and
                supports matching patterns such as segmentation faults, MPI
                aborts, scheduler preemption signatures, missing output, and
                other known error indicators.
            """
            out_path = os.path.join(run_location, stdout_file)
            err_path = os.path.join(run_location, stderr_file)

            out_text = tail_log(out_path, self.tail_bytes)
            err_text = tail_log(err_path, self.tail_bytes)
            combined = out_text + "\n" + err_text

            findings = classify_failure(combined, self.tail_bytes)

            summary = {
                "exit_code": exit_code,
                "stdout_tail": out_text,
                "stderr_tail": err_text,
                "detected_failures": findings,
            }

            return json.dumps(summary, indent=2)

classify_failure(log_text, tail_bytes=50000)

Classify failures in log data using regex-based patterns.

The function scans the provided log text for known error signatures defined in FAILURE_PATTERNS. If one or more patterns match, a list of findings is returned. Each finding includes the failure type, matched text, and a recommended action.

If no patterns match and the logs contain suspicious content (non-empty), an UNCLASSIFIED_FAILURE entry is returned.

If no patterns match and logs are empty or clean, the function returns an empty list, indicating a successful run.

Parameters:

Name Type Description Default
log_text str

Complete or combined log content to analyze.

 required
tail_bytes int

Maximum size of the excerpt included for unclassified failures.

 50000

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: A non-empty list of failure findings. Each finding
List[Dict[str, Any]]

includes keys such as: - failure_type (str) - matched_text (str or None) - recommendation (str) - log_excerpt (str) for unclassified failures
Source code in src/mada_tools/monitor/job_monitor/log_utils.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def classify_failure(log_text: str, tail_bytes: int = 50000) -> List[Dict[str, Any]]:
    """
    Classify failures in log data using regex-based patterns.

    The function scans the provided log text for known error signatures defined
    in FAILURE_PATTERNS. If one or more patterns match, a list of findings is
    returned. Each finding includes the failure type, matched text, and a
    recommended action.

    If no patterns match and the logs contain suspicious content (non-empty),
    an UNCLASSIFIED_FAILURE entry is returned.

    If no patterns match and logs are empty or clean, the function returns an
    empty list, indicating a successful run.

    Args:
        log_text (str): Complete or combined log content to analyze.
        tail_bytes (int): Maximum size of the excerpt included for unclassified
            failures.

    Returns:
        List[Dict[str, Any]]: A non-empty list of failure findings. Each finding
        includes keys such as:
            - failure_type (str)
            - matched_text (str or None)
            - recommendation (str)
            - log_excerpt (str) for unclassified failures
    """
    findings = []

    # Search for all known patterns
    for name, cfg in FAILURE_PATTERNS.items():
        pattern = cfg["pattern"]
        match = re.search(pattern, log_text, re.IGNORECASE)
        if match:
            findings.append(
                {
                    "failure_type": name,
                    "matched_text": match.group(0),
                    "recommendation": cfg["recommendation"],
                }
            )

    # Fallback case
    if not findings:
        # Detect if the log contains any generic failure signals
        generic_failure_signals = ["error", "exception", "fail", "killed", "abort"]
        text_lower = log_text.lower() if log_text else ""
        has_failure_signals = any(sig in text_lower for sig in generic_failure_signals)

        # Case 3: No failures → successful run → return empty list
        if not has_failure_signals:
            return []

        # Case 2: Failure occurred but does not match known patterns
        excerpt = log_text[-tail_bytes:] if log_text else ""
        findings.append({**UNCLASSIFIED_FAILURE, "matched_text": None, "log_excerpt": excerpt})

    return findings

combine_logs(paths, tail_bytes)

Combine multiple logs into a single formatted text block.

Each log file is tailed up to tail_bytes, then wrapped with a header naming the file. Missing files are included but reported with empty content.

Parameters:

Name Type Description Default
paths List[str]

Paths to log files.

 required
tail_bytes int

Number of bytes to read from the end of each file.

 required

Returns:

Name Type Description
str str

Combined and formatted log text for all paths.
Source code in src/mada_tools/monitor/job_monitor/log_utils.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def combine_logs(paths: List[str], tail_bytes: int) -> str:
    """
    Combine multiple logs into a single formatted text block.

    Each log file is tailed up to `tail_bytes`, then wrapped with a header naming
    the file. Missing files are included but reported with empty content.

    Args:
        paths (List[str]): Paths to log files.
        tail_bytes (int): Number of bytes to read from the end of each file.

    Returns:
        str: Combined and formatted log text for all paths.
    """
    logs = []
    for p in paths:
        logs.append(f"===== {os.path.basename(p)} =====\n")
        logs.append(tail_log(p, tail_bytes))
        logs.append("\n")
    return "".join(logs)

read_log(path)

Read the full contents of a log file.

If the file cannot be opened or read, the function returns an empty string.

Parameters:

Name Type Description Default
path str

Path to the log file to read.

 required

Returns:

Name Type Description
str str

Full contents of the log file, or an empty string if the file is missing
str

or unreadable.
Source code in src/mada_tools/monitor/job_monitor/log_utils.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def read_log(path: str) -> str:
    """
    Read the full contents of a log file.

    If the file cannot be opened or read, the function returns an empty string.

    Args:
        path (str): Path to the log file to read.

    Returns:
        str: Full contents of the log file, or an empty string if the file is missing
        or unreadable.
    """
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            return f.read()
    except Exception:
        return ""

tail_log(path, bytes_to_read)

Return the last bytes_to_read bytes of a log file.

Handles missing or unreadable files by returning an empty string.

Parameters:

Name Type Description Default
path str

Path to the log file.

 required
bytes_to_read int

Number of bytes from the end of the file to return.

 required

Returns:

Name Type Description
str str

The requested tail of the log file decoded as UTF-8, or an empty string
str

if the file is missing or cannot be read.
Source code in src/mada_tools/monitor/job_monitor/log_utils.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def tail_log(path: str, bytes_to_read: int) -> str:
    """
    Return the last `bytes_to_read` bytes of a log file.

    Handles missing or unreadable files by returning an empty string.

    Args:
        path (str): Path to the log file.
        bytes_to_read (int): Number of bytes from the end of the file to return.

    Returns:
        str: The requested tail of the log file decoded as UTF-8, or an empty string
        if the file is missing or cannot be read.
    """
    try:
        with open(path, "rb") as f:
            f.seek(0, os.SEEK_END)
            size = f.tell()
            f.seek(max(0, size - bytes_to_read))
            return f.read().decode("utf-8", errors="replace")
    except Exception:
        return ""