[FEA] Lightweight event log runtime detector (#2082)

[sayedbilalbari]

Summary

Adds a lightweight Python event-log detector that classifies a single Spark application log as one of:

• PROFILING when a RAPIDS runtime signal is found.
• QUALIFICATION when startup properties indicate standard OSS Spark with no RAPIDS markers.
• UNKNOWN when the bounded scan cannot make a safe decision.

The detector is intended as an early routing check before invoking the full tools pipeline. It scans event logs directly from the Python layer and does not invoke the Scala tools runtime for this decision.

Resolves #2082.

Public API

from spark_rapids_tools.tools.eventlog_detector import (
    DetectionResult,
    SparkRuntime,
    ToolExecution,
    detect_spark_runtime,
)

result: DetectionResult = detect_spark_runtime(event_log_path)
# result.tool_execution: ToolExecution (QUALIFICATION | PROFILING | UNKNOWN)
# result.spark_runtime: Optional[SparkRuntime] (SPARK | SPARK_RAPIDS)
# result.app_id, result.spark_version, result.event_log_path,
# result.source_path, result.reason

Callers can use ToolExecution.UNKNOWN or any EventLogDetectionError subclass as a fallback signal for the full tools pipeline.

Detection Contract

The detector performs best-effort early routing for RAPIDS and standard OSS Spark event logs:

• RAPIDS logs are identified from SparkRapidsBuildInfoEvent or RAPIDS Spark configuration markers.
• OSS Spark logs are identified from SparkListenerEnvironmentUpdate when startup properties contain no RAPIDS markers.
• Logs remain UNKNOWN when the bounded scan reaches max_events_scanned before seeing enough information.

The scanner is streaming and bounded. It reads line-by-line through CspPath.open_input_stream(), stops as soon as a decision is available, and retains only runtime metadata plus accumulated Spark properties.

Supported Inputs

• Single event-log files.
• Apache Spark rolling event-log directories using the eventlog_v2_* / events_* layout.
• Plain text, .inprogress, .gz, .zst, and .zstd event logs.
• Local paths and remote paths supported by the existing CspPath storage layer.

Unsupported input shapes and codecs raise typed EventLogDetectionError subclasses so callers can fall back cleanly.

Implementation Notes

• README.md documents the detection flow, RAPIDS markers, CPU fast path, and streaming memory contract.
• markers.py keeps RAPIDS and Spark event constants pinned to the relevant Scala sources.
• resolver.py resolves a single-app input into an ordered file list.
• stream.py provides codec-aware streaming without loading the full log.
• scanner.py performs bounded event scanning across one or more files.
• detector.py exposes the top-level detect_spark_runtime entry point.

Dependencies

Adds zstandard==0.25.0 for .zstd event logs. .gz and .zst continue to use the existing PyArrow-backed I/O path.

Test Plan

• PYTHONPATH=src python -m pytest -q tests/spark_rapids_tools_ut/tools/eventlog_detector
• tox -e pylint,flake8
• Local validation against CPU, RAPIDS, OSS rolling, and .inprogress event logs.
• Remote validation against CPU and RAPIDS OSS rolling event logs using the existing storage configuration.

[greptile-apps]

Greptile Summary

This PR adds a lightweight, streaming Python detector that classifies a single Spark application event log as PROFILING, QUALIFICATION, or UNKNOWN without invoking the full Scala tools pipeline. The implementation is well-structured across six focused modules (resolver, stream, scanner, classifier, detector, markers) with comprehensive unit and integration tests.

• P1 — AttributeError escapes the error hierarchy in scanner.py: json.loads can succeed yet return a non-dict JSON value (array, number, string). Line 79's event.get(\"Event\") is outside the try/except (json.JSONDecodeError, ValueError) block and raises AttributeError for such values. This propagates through scan_events_across and detect_spark_runtime as a bare Python exception, bypassing the typed EventLogDetectionError hierarchy that callers use for graceful fallback.

Confidence Score: 4/5

Safe to merge after fixing the non-dict JSON guard in scanner.py; all other logic is correct and well-tested.

One P1 bug: a valid-JSON non-dict line (e.g. a bare JSON array) causes an unhandled AttributeError that bypasses the EventLogDetectionError hierarchy. All other components — classification, streaming, path resolution, decision logic — look correct and are covered by tests. Score is capped at 4 by the single P1.

user_tools/src/spark_rapids_tools/tools/eventlog_detector/scanner.py — the non-dict JSON guard fix is isolated to the scan_events function.

Important Files Changed

Filename	Overview
user_tools/src/spark_rapids_tools/tools/eventlog_detector/scanner.py	Bounded streaming scanner with correct merge-then-classify logic; P1 bug: non-dict JSON values cause unhandled AttributeError after the try/except block at line 70-76.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/classifier.py	Clean; _parse_bool correctly matches Scala's Try{s.toBoolean}.getOrElse(default) semantics; has_rapids_conf_markers intentionally broader than _is_spark_rapids.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/detector.py	Top-level entry point; decision logic (PROFILING/QUALIFICATION/UNKNOWN) is correct and consistent with scanner termination states.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/stream.py	Codec-aware streaming; OSError guard and ExitStack cleanup are correct; TextIOWrapper registered last in ExitStack causes harmless double-close of underlying byte_stream.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/resolver.py	Single-file and OSS rolling-dir resolution; numeric sort of events_ chunks is correct; clean error hierarchy.
user_tools/tests/spark_rapids_tools_ut/tools/eventlog_detector/test_scanner.py	Good coverage of termination paths and cross-file scanning; missing a test for valid-JSON-but-non-dict lines that would expose the AttributeError bug.
user_tools/tests/spark_rapids_tools_ut/tools/eventlog_detector/test_detector.py	Integration tests cover all ToolExecution outcomes, rolling dirs, string vs CspPath inputs, and source_path preservation.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/markers.py	Constants only; Scala source references are well-documented; no logic issues.
user_tools/src/spark_rapids_tools/tools/eventlog_detector/types.py	Clean enum and dataclass definitions; str-enum subclassing for ToolExecution/SparkRuntime is correct for wire-format compatibility.
user_tools/pyproject.toml	Adds zstandard==0.25.0 as a pinned dependency for .zstd event log support.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[detect_spark_runtime] --> B[resolve_event_log_files]
    B -->|single file| C[1-element file list]
    B -->|eventlog_v2_dir| D[sorted events_N_ files]
    B -->|other dir| E([UnsupportedInputError])
    C --> F[scan_events_across]
    D --> F
    F --> G[open_event_log_stream per file]
    G -->|.gz or .zst| H[PyArrow auto-decompress]
    G -->|.zstd| I[zstandard manual decompress]
    G -->|plain or .inprogress| J[raw bytes]
    G -->|other suffix| K([UnsupportedCompressionError])
    H --> L[scan_events JSON line loop]
    I --> L
    J --> L
    L -->|SparkRapidsBuildInfoEvent| M([DECISIVE SPARK_RAPIDS])
    L -->|EnvironmentUpdate classifies SPARK_RAPIDS| M
    L -->|EnvironmentUpdate no RAPIDS markers fast-path| N([CPU_FAST_PATH SPARK])
    L -->|budget exhausted| O([CAP_HIT])
    L -->|all lines read| P([EXHAUSTED])
    M --> Q{detector decision}
    N --> Q
    O --> Q
    P --> Q
    Q -->|SPARK_RAPIDS| R([PROFILING])
    Q -->|CPU_FAST_PATH and SPARK| S([QUALIFICATION])
    Q -->|EXHAUSTED and env_update_seen| S
    Q -->|otherwise| T([UNKNOWN])

Loading

_{Reviews (9): Last reviewed commit: "test: trim event log detector coverage o..." | Re-trigger Greptile}

[sayedbilalbari]

@greptile review

[sayedbilalbari]

@greptile review

[sayedbilalbari]

This logic is copied from how the Scala implementation in tools is

[cindyyuanjiang]

minor nit: can we combine test_disabled_rapids_plugin_still_counts_as_marker and test_rapids_enabled_key_alone_counts_as_marker here too? they seem to test the same thing

[sayedbilalbari]

updated. Actually updating to remove a bunch of other redundant or low-value tests as well.

[sayedbilalbari]

Removed these tests ->

  - test_detector.py::test_fast_path_does_not_fire_when_rapids_marker_present
    Removed because scanner already tests this exact branch: RAPIDS marker present in env update prevents CPU fast path, then later SQL config makes
    it decisive.
  - test_detector.py::test_reason_mentions_runtime_on_profiling
    Removed because it only checks a human-readable reason string. Routing behavior is already tested.
  - test_detector.py::test_reason_mentions_full_log_on_strict_qualification
    Removed for the same reason: brittle reason-string assertion, not core behavior.
  - test_resolver.py::test_generic_multi_app_dir_raises
    Removed because it hits the same resolver branch as test_non_oss_rolling_dir_raises: directory basename does not start with eventlog_v2_.
  - test_types.py::test_base_is_exception_not_value_error
    Removed because it asserts an implementation choice rather than an integration contract. We kept the important part: detector-specific errors
    subclass EventLogDetectionError

These tests have been consolidated ->

  - test_classifier.py marker tests
    Previously separate tests checked plain Spark, enabled-key-only, plugin-only, plugin-list, plugin-plus-enabled, and lookalike plugin. These are
    now one parametrized test:

    test_detects_rapids_marker_configs(props, expected)
    This keeps all cases but removes repeated test structure.
    This keeps all cases but removes repeated test structure.
  - test_stream.py unsupported compression tests
    Previously .lz4, .snappy, .lzf, and unknown suffix each had a separate test. These are now one parametrized test:

    test_unsupported_suffix_raises(suffix)

[cindyyuanjiang]

@sayedbilalbari To confirm the high-level goal of the parser:

1. For GPU event log, the detection is pretty decisive
2. For CPU event log, we can afford false positives because qualification tool will fail with GPU event log (then we can profile the event log)

[cindyyuanjiang]

Thanks @sayedbilalbari! LGTM!

[sayedbilalbari]

@sayedbilalbari To confirm the high-level goal of the parser:

1. For GPU event log, the detection is pretty decisive
2. For CPU event log, we can afford false positives because qualification tool will fail with GPU event log (then we can profile the event log)

@cindyyuanjiang Yes. With the current fast-path default, this is the behavior. In case false positives are to be avoided, toggle the fast-path off(trade off in some cases where running the qualification and failing takes longer than some extra time at detection)

[greptile-apps]

Non-dict JSON values escape the try/except and raise AttributeError

json.loads(raw) can succeed and return a non-dict JSON value (e.g., a JSON array […] or a bare number). The try/except (json.JSONDecodeError, ValueError) only guards the parse call on line 71; line 79's event.get("Event") is outside that block and raises AttributeError for any non-dict value. That AttributeError propagates all the way through scan_events_across and detect_spark_runtime as a bare Python exception — not an EventLogDetectionError — breaking callers that key their fallback on the detector's typed error hierarchy.

A minimal guard keeps the handling consistent with the existing "skip unrecognised lines" approach:

        try:
            event = json.loads(raw)
        except (json.JSONDecodeError, ValueError):
            result.events_scanned += 1
            continue

        if not isinstance(event, dict):
            result.events_scanned += 1
            continue

        result.events_scanned += 1
        name = event.get("Event")