From adb9a75ea5edcb76d47b10bdb2c2cb931b4e1282 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 10 Aug 2026 13:14:34 -0600 Subject: [PATCH 1/5] add extraction analysis strategy Signed-off-by: Bob --- lib/ramble/ramble/analysis/__init__.py | 24 ++ lib/ramble/ramble/analysis/base.py | 20 ++ lib/ramble/ramble/analysis/default.py | 257 ++++++++++++++++ .../ramble/test/test_analysis_strategy.py | 29 ++ .../application-base/base_class.py | 278 +----------------- 5 files changed, 334 insertions(+), 274 deletions(-) create mode 100644 lib/ramble/ramble/analysis/__init__.py create mode 100644 lib/ramble/ramble/analysis/base.py create mode 100644 lib/ramble/ramble/analysis/default.py create mode 100644 lib/ramble/ramble/test/test_analysis_strategy.py diff --git a/lib/ramble/ramble/analysis/__init__.py b/lib/ramble/ramble/analysis/__init__.py new file mode 100644 index 000000000..667c2e071 --- /dev/null +++ b/lib/ramble/ramble/analysis/__init__.py @@ -0,0 +1,24 @@ +# Copyright 2022-2026 The Ramble Authors +# +# Licensed under the Apache License, Version 2.0 or the MIT license +# , at your +# option. This file may not be copied, modified, or distributed +# except according to those terms. + +"""Analysis package for Ramble""" + +# flake8: noqa: F401 +from ramble.analysis.base import AnalysisStrategyBase as AnalysisStrategyBase +from ramble.analysis.default import DefaultAnalysisStrategy + +_strategy_registry = { + "default": DefaultAnalysisStrategy, +} + + +def get_strategy(name, app_inst): + """Get the analysis strategy instance by name.""" + if name not in _strategy_registry: + raise ValueError(f"Unknown analysis strategy: {name}") + return _strategy_registry[name](app_inst) diff --git a/lib/ramble/ramble/analysis/base.py b/lib/ramble/ramble/analysis/base.py new file mode 100644 index 000000000..9fb0ee4e4 --- /dev/null +++ b/lib/ramble/ramble/analysis/base.py @@ -0,0 +1,20 @@ +# Copyright 2022-2026 The Ramble Authors +# +# Licensed under the Apache License, Version 2.0 or the MIT license +# , at your +# option. This file may not be copied, modified, or distributed +# except according to those terms. + +"""Define base classes for analysis strategies""" + +from ramble.language.application_language import ApplicationMeta + + +class AnalysisStrategyBase(metaclass=ApplicationMeta): + + def __init__(self, app_inst): + self.app_inst = app_inst + + def __call__(self, workspace): + raise NotImplementedError diff --git a/lib/ramble/ramble/analysis/default.py b/lib/ramble/ramble/analysis/default.py new file mode 100644 index 000000000..ec95bab47 --- /dev/null +++ b/lib/ramble/ramble/analysis/default.py @@ -0,0 +1,257 @@ +# Copyright 2022-2026 The Ramble Authors +# +# Licensed under the Apache License, Version 2.0 or the MIT license +# , at your +# option. This file may not be copied, modified, or distributed +# except according to those terms. + +"""Define the default analysis strategy""" + +import os +import string + +import ramble.success_criteria +import ramble.util.lock as lk +from ramble.analysis.base import AnalysisStrategyBase +from ramble.experiment_result import ExperimentStatus +from ramble.util.logger import logger + +_NULL_CONTEXT = "null" + + +def _get_context_display_name(context): + return ( + f"default ({_NULL_CONTEXT}) context" if context == _NULL_CONTEXT else f"{context} context" + ) + + +class DefaultAnalysisStrategy(AnalysisStrategyBase): + """Default regex-based analysis/extraction strategy.""" + + def __call__(self, workspace): + app = self.app_inst + + if app.get_status() == ExperimentStatus.UNKNOWN and not workspace.dry_run: + logger.warn(f"Experiment has status {app.get_status()}. Skipping analysis..\n") + app.result.finalize(workspace) + return + + def format_context(context_match, context_format): + context_val = {} + if isinstance(context_format, str): + for group in string.Formatter().parse(context_format): + if group[1]: + context_val[group[1]] = context_match[group[1]] + + context_string = context_format.format(**context_val) + return context_string + + # Exit early if read from cache works. + if app.result.read_cache(workspace, app): + app.result.finalize(workspace) + return + + criteria_list = app.success_list + if not criteria_list: + criteria_list = ramble.success_criteria.ScopedCriteriaList() + criteria_list.reset() + + files, f_defs, inmem_defs = app.analysis_dicts(criteria_list) + + exp_lock = app.experiment_lock + + fom_values = {} + context_metadata = {} + null_key = (_NULL_CONTEXT, _NULL_CONTEXT, frozenset()) + context_metadata[null_key] = { + "name": _NULL_CONTEXT, + "def_name": _NULL_CONTEXT, + "vars": {}, + } + + # Iterate over files. We already know they exist + with lk.ReadTransaction(exp_lock): + for file, file_conf in files.items(): + + # Start with no active contexts in a file. + active_contexts = {} + logger.debug(f"Reading log file: {file}") + + if not os.path.exists(file): + logger.debug(f"Skipping analysis of non-existent file: {file}") + continue + + per_file_crit_objs = [ + criteria_list.find_criteria(c) for c in file_conf["success_criteria"] + ] + + with open(file, encoding="utf-8", errors="replace") as f: + for line in f: + new_per_file_crit_objs = [] + for crit_obj in per_file_crit_objs: + if crit_obj.passed(line, app): + crit_obj.mark_found() + elif crit_obj.anti_matched(line): + crit_obj.mark_anti_found() + else: + new_per_file_crit_objs.append(crit_obj) + per_file_crit_objs = new_per_file_crit_objs + + # Iterate over contexts and add matched contexts to active_contexts + for context, foms in file_conf["contexts"].items(): + if context != _NULL_CONTEXT: + context_conf = f_defs[context]["definition"] + if context_conf.get("pre_filter", "") not in line: + context_match = None + else: + context_match = context_conf["regex"].match(line) + + if context_match: + context_name = format_context( + context_match, + context_conf["format"], + ) + logger.debug(f"Line was: {line}") + logger.debug(f" Context match {context} -- {context_name}") + + context_vars = context_match.groupdict() + context_key = ( + context_name, + context, + frozenset(context_vars.items()), + ) + + active_contexts[context] = context_key + + if context_key not in fom_values: + fom_values[context_key] = {} + context_metadata[context_key] = { + "name": context_name, + "def_name": context, + "vars": context_vars, + } + + for fom in foms: + fom_conf = f_defs[context]["foms"][fom] + if fom_conf.get("pre_filter", "") not in line: + fom_match = None + else: + fom_match = fom_conf["regex"].match(line) + + if fom_match: + fom_vars = fom_match.groupdict() + if fom_conf["fom_name_expanded"] is not None: + fom_name = fom_conf["fom_name_expanded"] + else: + fom_name = app.expander.expand_var( + fom, extra_vars=fom_vars + ) + + if fom_conf["group"] in fom_conf["regex"].groupindex: + logger.debug(f" --- Matched fom {fom_name}") + fom_contexts = [] + # if a FOM has contexts, check if each is active + if fom_conf["contexts"]: + for _ in fom_conf["contexts"]: + context_key = ( + active_contexts[context] + if context in active_contexts + else null_key + ) + fom_contexts.append(context_key) + else: + fom_contexts.append(null_key) + + for fom_context in fom_contexts: + if fom_context not in fom_values: + fom_values[fom_context] = {} + fom_val = fom_match.group(fom_conf["group"]) + if fom_val is None: + continue + if fom_conf["units_expanded"] is not None: + fom_unit = fom_conf["units"] + else: + fom_unit = app.expander.expand_var( + fom_conf["units"], + extra_vars=fom_vars, + ) + fom_values[fom_context][fom_name] = { + "value": fom_val, + "units": fom_unit, + "origin": fom_conf["origin"], + "origin_type": fom_conf["origin_type"], + "fom_type": fom_conf["fom_type"], + } + app.extract_inmem_foms(inmem_defs, fom_values, context_metadata) + + # Test all non-file based success criteria + for criteria_obj, _ in criteria_list.all_criteria(): + if criteria_obj.file is None: + if criteria_obj.passed(app_inst=app, fom_values=fom_values): + criteria_obj.mark_found() + + # If an app has no FOMs defined, don't fail it for that + success = (not f_defs and not inmem_defs) or False + for fom in fom_values.values(): + for value in fom.values(): + if "origin_type" in value and value["origin_type"] == "application": + success = True + success = success and criteria_list.passed() + + if success: + status = ExperimentStatus.SUCCESS + else: + preserved_terminal = { + ExperimentStatus.CANCELLED, + ExperimentStatus.TIMEOUT, + ExperimentStatus.FAILED, + } + current_status = app.get_status() + if current_status in preserved_terminal: + status = current_status + else: + status = ExperimentStatus.FAILED + + # When workflow_manager is present, only use app_status when workflow is completed or + # unresolved. + if app.workflow_manager is not None: + wm_status = app.workflow_manager.get_status(workspace) + if not ( + wm_status is None + or wm_status in [ExperimentStatus.COMPLETE, ExperimentStatus.UNRESOLVED] + ): + status = wm_status + + app.set_status(status) + app.result.finalize(workspace) + + for criteria_obj, criteria_scope in criteria_list.all_criteria(): + if criteria_obj.owner is not None: + criteria_name = f"{criteria_obj.owner.scoped_name}::{criteria_obj.name}" + else: + criteria_name = f"config::{criteria_scope}::{criteria_obj.name}" + if criteria_obj.ok(): + app.result.success_criteria[criteria_name] = "PASSED" + else: + app.result.success_criteria[criteria_name] = "FAILED" + + for context_key, fom_map in fom_values.items(): + metadata = context_metadata[context_key] + context_map = { + "name": metadata["name"], + "foms": [], + "display_name": _get_context_display_name(metadata["name"]), + "context_def_name": metadata["def_name"], + "context_vars": metadata["vars"], + } + + for fom_name, fom in fom_map.items(): + fom_copy = fom.copy() + fom_copy["name"] = fom_name + context_map["foms"].append(fom_copy) + + if metadata["name"] == _NULL_CONTEXT: + app.result.contexts.insert(0, context_map) + else: + app.result.contexts.append(context_map) diff --git a/lib/ramble/ramble/test/test_analysis_strategy.py b/lib/ramble/ramble/test/test_analysis_strategy.py new file mode 100644 index 000000000..794529094 --- /dev/null +++ b/lib/ramble/ramble/test/test_analysis_strategy.py @@ -0,0 +1,29 @@ +# Copyright 2022-2026 The Ramble Authors +# +# Licensed under the Apache License, Version 2.0 or the MIT license +# , at your +# option. This file may not be copied, modified, or distributed +# except according to those terms. + +"""Unit tests for analysis strategy pattern""" + +import pytest + +import ramble.analysis +from ramble.analysis.default import DefaultAnalysisStrategy + + +class DummyApp: + pass + + +def test_get_strategy(): + app = DummyApp() + + strategy = ramble.analysis.get_strategy("default", app) + assert isinstance(strategy, DefaultAnalysisStrategy) + assert strategy.app_inst is app + + with pytest.raises(ValueError, match="Unknown analysis strategy: invalid"): + ramble.analysis.get_strategy("invalid", app) diff --git a/var/ramble/repos/builtin/base_classes/application-base/base_class.py b/var/ramble/repos/builtin/base_classes/application-base/base_class.py index 79d67dcab..5bbbb9d5a 100644 --- a/var/ramble/repos/builtin/base_classes/application-base/base_class.py +++ b/var/ramble/repos/builtin/base_classes/application-base/base_class.py @@ -16,7 +16,6 @@ import shlex import shutil import stat -import string import time from html import escape from typing import Dict, List @@ -3464,280 +3463,11 @@ def _analyze_experiments(self, workspace, app_inst=None): Success criteria are defined within the application.py, but can also be injected in a workspace config. """ + import ramble.analysis - if ( - self.get_status() == ExperimentStatus.UNKNOWN - and not workspace.dry_run - ): - logger.warn( - f"Experiment has status {self.get_status()}. Skipping analysis..\n" - ) - self.result.finalize(workspace) - return - - def format_context(context_match, context_format): - - context_val = {} - if isinstance(context_format, str): - for group in string.Formatter().parse(context_format): - if group[1]: - context_val[group[1]] = context_match[group[1]] - - context_string = context_format.format(**context_val) - return context_string - - # Exit early if read from cache works. - if self.result.read_cache(workspace, self): - self.result.finalize(workspace) - return - - criteria_list = self.success_list - if not criteria_list: - criteria_list = ramble.success_criteria.ScopedCriteriaList() - criteria_list.reset() - - files, f_defs, inmem_defs = self.analysis_dicts(criteria_list) - - exp_lock = self.experiment_lock - - fom_values = {} - context_metadata = {} - null_key = (_NULL_CONTEXT, _NULL_CONTEXT, frozenset()) - context_metadata[null_key] = { - "name": _NULL_CONTEXT, - "def_name": _NULL_CONTEXT, - "vars": {}, - } - - # Iterate over files. We already know they exist - with lk.ReadTransaction(exp_lock): - for file, file_conf in files.items(): - - # Start with no active contexts in a file. - active_contexts = {} - logger.debug(f"Reading log file: {file}") - - if not os.path.exists(file): - logger.debug( - f"Skipping analysis of non-existent file: {file}" - ) - continue - - per_file_crit_objs = [ - criteria_list.find_criteria(c) - for c in file_conf["success_criteria"] - ] - - with open(file, encoding="utf-8", errors="replace") as f: - for line in f: - new_per_file_crit_objs = [] - for crit_obj in per_file_crit_objs: - if crit_obj.passed(line, self): - crit_obj.mark_found() - elif crit_obj.anti_matched(line): - crit_obj.mark_anti_found() - else: - new_per_file_crit_objs.append(crit_obj) - per_file_crit_objs = new_per_file_crit_objs - - # Iterate over contexts and add matched contexts to active_contexts - for context, foms in file_conf["contexts"].items(): - if context != _NULL_CONTEXT: - context_conf = f_defs[context]["definition"] - if ( - context_conf.get("pre_filter", "") - not in line - ): - context_match = None - else: - context_match = context_conf[ - "regex" - ].match(line) - - if context_match: - context_name = format_context( - context_match, - context_conf["format"], - ) - logger.debug(f"Line was: {line}") - logger.debug( - f" Context match {context} -- {context_name}" - ) - - context_vars = context_match.groupdict() - context_key = ( - context_name, - context, - frozenset(context_vars.items()), - ) - - active_contexts[context] = context_key - - if context_key not in fom_values: - fom_values[context_key] = {} - context_metadata[context_key] = { - "name": context_name, - "def_name": context, - "vars": context_vars, - } - - for fom in foms: - fom_conf = f_defs[context]["foms"][fom] - if fom_conf.get("pre_filter", "") not in line: - fom_match = None - else: - fom_match = fom_conf["regex"].match(line) - - if fom_match: - fom_vars = fom_match.groupdict() - if ( - fom_conf["fom_name_expanded"] - is not None - ): - fom_name = fom_conf[ - "fom_name_expanded" - ] - else: - fom_name = self.expander.expand_var( - fom, extra_vars=fom_vars - ) - - if ( - fom_conf["group"] - in fom_conf["regex"].groupindex - ): - logger.debug( - f" --- Matched fom {fom_name}" - ) - fom_contexts = [] - # if a FOM has contexts, check if each is active - if fom_conf["contexts"]: - for _ in fom_conf["contexts"]: - context_key = ( - active_contexts[context] - if context - in active_contexts - else null_key - ) - fom_contexts.append( - context_key - ) - else: - fom_contexts.append(null_key) - - for fom_context in fom_contexts: - if fom_context not in fom_values: - fom_values[fom_context] = {} - fom_val = fom_match.group( - fom_conf["group"] - ) - if fom_val is None: - continue - if ( - fom_conf["units_expanded"] - is not None - ): - fom_unit = fom_conf["units"] - else: - fom_unit = ( - self.expander.expand_var( - fom_conf["units"], - extra_vars=fom_vars, - ) - ) - fom_values[fom_context][ - fom_name - ] = { - "value": fom_val, - "units": fom_unit, - "origin": fom_conf["origin"], - "origin_type": fom_conf[ - "origin_type" - ], - "fom_type": fom_conf[ - "fom_type" - ], - } - self.extract_inmem_foms(inmem_defs, fom_values, context_metadata) - - # Test all non-file based success criteria - for criteria_obj, _ in criteria_list.all_criteria(): - if criteria_obj.file is None: - if criteria_obj.passed(app_inst=self, fom_values=fom_values): - criteria_obj.mark_found() - - # If an app has no FOMs defined, don't fail it for that - success = (not f_defs and not inmem_defs) or False - for fom in fom_values.values(): - for value in fom.values(): - if ( - "origin_type" in value - and value["origin_type"] == "application" - ): - success = True - success = success and criteria_list.passed() - - if success: - status = ExperimentStatus.SUCCESS - else: - preserved_terminal = { - ExperimentStatus.CANCELLED, - ExperimentStatus.TIMEOUT, - ExperimentStatus.FAILED, - } - current_status = self.get_status() - if current_status in preserved_terminal: - status = current_status - else: - status = ExperimentStatus.FAILED - - # When workflow_manager is present, only use app_status when workflow is completed or - # unresolved. - if self.workflow_manager is not None: - wm_status = self.workflow_manager.get_status(workspace) - if not ( - wm_status is None - or wm_status - in [ExperimentStatus.COMPLETE, ExperimentStatus.UNRESOLVED] - ): - status = wm_status - - self.set_status(status) - self.result.finalize(workspace) - - for criteria_obj, criteria_scope in criteria_list.all_criteria(): - if criteria_obj.owner is not None: - criteria_name = ( - f"{criteria_obj.owner.scoped_name}::{criteria_obj.name}" - ) - else: - criteria_name = ( - f"config::{criteria_scope}::{criteria_obj.name}" - ) - if criteria_obj.ok(): - self.result.success_criteria[criteria_name] = "PASSED" - else: - self.result.success_criteria[criteria_name] = "FAILED" - - for context_key, fom_map in fom_values.items(): - metadata = context_metadata[context_key] - context_map = { - "name": metadata["name"], - "foms": [], - "display_name": _get_context_display_name(metadata["name"]), - "context_def_name": metadata["def_name"], - "context_vars": metadata["vars"], - } - - for fom_name, fom in fom_map.items(): - fom_copy = fom.copy() - fom_copy["name"] = fom_name - context_map["foms"].append(fom_copy) - - if metadata["name"] == _NULL_CONTEXT: - self.result.contexts.insert(0, context_map) - else: - self.result.contexts.append(context_map) + strategy_name = getattr(self, "analysis_strategy", "default") + strategy = ramble.analysis.get_strategy(strategy_name, self) + strategy(workspace) register_phase( "append_results_to_workspace", From 2ee621338f7c8f4b0a6729fad64f3f7f6bb0ec40 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 17 Aug 2026 13:23:08 -0600 Subject: [PATCH 2/5] Add backwards-and-exit file strat Signed-off-by: Bob --- lib/ramble/ramble/analysis/__init__.py | 3 + lib/ramble/ramble/analysis/backwards.py | 269 ++++++++++++++++++ .../ramble/test/test_analysis_strategy.py | 175 ++++++++++++ .../application-base/base_class.py | 2 +- 4 files changed, 448 insertions(+), 1 deletion(-) create mode 100644 lib/ramble/ramble/analysis/backwards.py diff --git a/lib/ramble/ramble/analysis/__init__.py b/lib/ramble/ramble/analysis/__init__.py index 667c2e071..d6c285bae 100644 --- a/lib/ramble/ramble/analysis/__init__.py +++ b/lib/ramble/ramble/analysis/__init__.py @@ -8,12 +8,15 @@ """Analysis package for Ramble""" +from ramble.analysis.backwards import BackwardsAnalysisStrategy + # flake8: noqa: F401 from ramble.analysis.base import AnalysisStrategyBase as AnalysisStrategyBase from ramble.analysis.default import DefaultAnalysisStrategy _strategy_registry = { "default": DefaultAnalysisStrategy, + "backwards": BackwardsAnalysisStrategy, } diff --git a/lib/ramble/ramble/analysis/backwards.py b/lib/ramble/ramble/analysis/backwards.py new file mode 100644 index 000000000..54d1dc4dc --- /dev/null +++ b/lib/ramble/ramble/analysis/backwards.py @@ -0,0 +1,269 @@ +# Copyright 2022-2026 The Ramble Authors +# +# Licensed under the Apache License, Version 2.0 or the MIT license +# , at your +# option. This file may not be copied, modified, or distributed +# except according to those terms. + +"""Define the backwards-reading analysis strategy""" + +import os + +import ramble.success_criteria +import ramble.util.lock as lk +from ramble.analysis.base import AnalysisStrategyBase +from ramble.experiment_result import ExperimentStatus +from ramble.util.logger import logger + +_NULL_CONTEXT = "null" + + +def _get_context_display_name(context): + return ( + f"default ({_NULL_CONTEXT}) context" if context == _NULL_CONTEXT else f"{context} context" + ) + + +def _read_file_backwards(file_path, block_size=4096): + """Yield lines from a file backwards, matching the forward reader.""" + with open(file_path, "rb") as f: + f.seek(0, os.SEEK_END) + file_size = f.tell() + position = file_size + buffer = b"" + is_first_block = True + + while position > 0: + grab_size = min(block_size, position) + position -= grab_size + f.seek(position) + chunk = f.read(grab_size) + buffer = chunk + buffer + + lines = buffer.split(b"\n") + if is_first_block and lines and lines[-1] == b"": + lines.pop() + is_first_block = False + buffer = lines[0] + + for line in reversed(lines[1:]): + yield line.decode("utf-8", errors="replace") + "\n" + + if buffer: + yield buffer.decode("utf-8", errors="replace") + "\n" + + +class BackwardsAnalysisStrategy(AnalysisStrategyBase): + """Optimized analysis strategy that reads logs backwards and stops early.""" + + def __call__(self, workspace): + app = self.app_inst + + if app.get_status() == ExperimentStatus.UNKNOWN and not workspace.dry_run: + logger.warn(f"Experiment has status {app.get_status()}. Skipping analysis..\n") + app.result.finalize(workspace) + return + + # Exit early if read from cache works. + if app.result.read_cache(workspace, app): + app.result.finalize(workspace) + return + + criteria_list = app.success_list + if not criteria_list: + criteria_list = ramble.success_criteria.ScopedCriteriaList() + criteria_list.reset() + + files, f_defs, inmem_defs = app.analysis_dicts(criteria_list) + + # Validate that only the null context is used + for file_conf in files.values(): + for context in file_conf["contexts"]: + if context != _NULL_CONTEXT: + if getattr(app, "analysis_strategy", None) is None: + logger.debug( + "Falling back to default forward-reading strategy due to non-null " + "context." + ) + default_strategy = ramble.analysis.get_strategy("default", app) + return default_strategy(workspace) + else: + raise ValueError( + f"BackwardsAnalysisStrategy cannot be used because " + f"context '{context}' is not the null context. " + "This strategy only supports the null context." + ) + + # Validate that only static FOMs and static units are used + for context_dict in f_defs.values(): + for fom_name, fom_conf in context_dict.get("foms", {}).items(): + if ( + fom_conf.get("fom_name_expanded") is None + or fom_conf.get("units_expanded") is None + ): + if getattr(app, "analysis_strategy", None) is None: + logger.debug( + "Falling back to default forward-reading strategy due to dynamic " + "FOM name or units." + ) + default_strategy = ramble.analysis.get_strategy("default", app) + return default_strategy(workspace) + else: + raise ValueError( + f"BackwardsAnalysisStrategy cannot be used because " + f"FOM '{fom_name}' has dynamic name or units. " + "This strategy only supports static FOMs." + ) + + exp_lock = app.experiment_lock + + fom_values = {} + context_metadata = {} + null_key = (_NULL_CONTEXT, _NULL_CONTEXT, frozenset()) + context_metadata[null_key] = { + "name": _NULL_CONTEXT, + "def_name": _NULL_CONTEXT, + "vars": {}, + } + + # Iterate over files. We already know they exist + with lk.ReadTransaction(exp_lock): + for file, file_conf in files.items(): + + logger.debug(f"Reading log file backwards: {file}") + + if not os.path.exists(file): + logger.debug(f"Skipping analysis of non-existent file: {file}") + continue + + per_file_crit_objs = [ + criteria_list.find_criteria(c) for c in file_conf["success_criteria"] + ] + + foms_to_find = set(file_conf["contexts"].get(_NULL_CONTEXT, [])) + + for line in _read_file_backwards(file): + new_per_file_crit_objs = [] + for crit_obj in per_file_crit_objs: + if crit_obj.passed(line, app): + crit_obj.mark_found() + elif crit_obj.anti_matched(line): + crit_obj.mark_anti_found() + else: + new_per_file_crit_objs.append(crit_obj) + per_file_crit_objs = new_per_file_crit_objs + + for fom in list(foms_to_find): + fom_conf = f_defs[_NULL_CONTEXT]["foms"][fom] + if fom_conf.get("pre_filter", "") not in line: + fom_match = None + else: + fom_match = fom_conf["regex"].match(line) + + if fom_match: + fom_vars = fom_match.groupdict() + if fom_conf["fom_name_expanded"] is not None: + fom_name = fom_conf["fom_name_expanded"] + else: + fom_name = app.expander.expand_var(fom, extra_vars=fom_vars) + + if fom_conf["group"] in fom_conf["regex"].groupindex: + fom_val = fom_match.group(fom_conf["group"]) + if fom_val is not None: + if fom_conf["units_expanded"] is not None: + fom_unit = fom_conf["units"] + else: + fom_unit = app.expander.expand_var( + fom_conf["units"], + extra_vars=fom_vars, + ) + + if null_key not in fom_values: + fom_values[null_key] = {} + fom_values[null_key][fom_name] = { + "value": fom_val, + "units": fom_unit, + "origin": fom_conf["origin"], + "origin_type": fom_conf["origin_type"], + "fom_type": fom_conf["fom_type"], + } + foms_to_find.remove(fom) + + # Stop reading if everything is found + if not foms_to_find and not per_file_crit_objs: + logger.debug("Found all FOMs and success criteria, stopping early.") + break + + app.extract_inmem_foms(inmem_defs, fom_values, context_metadata) + + # Test all non-file based success criteria + for criteria_obj, _ in criteria_list.all_criteria(): + if criteria_obj.file is None: + if criteria_obj.passed(app_inst=app, fom_values=fom_values): + criteria_obj.mark_found() + + # If an app has no FOMs defined, don't fail it for that + success = (not f_defs and not inmem_defs) or False + for fom in fom_values.values(): + for value in fom.values(): + if "origin_type" in value and value["origin_type"] == "application": + success = True + success = success and criteria_list.passed() + + if success: + status = ExperimentStatus.SUCCESS + else: + preserved_terminal = { + ExperimentStatus.CANCELLED, + ExperimentStatus.TIMEOUT, + ExperimentStatus.FAILED, + } + current_status = app.get_status() + if current_status in preserved_terminal: + status = current_status + else: + status = ExperimentStatus.FAILED + + # When workflow_manager is present, only use app_status when workflow is completed or + # unresolved. + if app.workflow_manager is not None: + wm_status = app.workflow_manager.get_status(workspace) + if not ( + wm_status is None + or wm_status in [ExperimentStatus.COMPLETE, ExperimentStatus.UNRESOLVED] + ): + status = wm_status + + app.set_status(status) + app.result.finalize(workspace) + + for criteria_obj, criteria_scope in criteria_list.all_criteria(): + if criteria_obj.owner is not None: + criteria_name = f"{criteria_obj.owner.scoped_name}::{criteria_obj.name}" + else: + criteria_name = f"config::{criteria_scope}::{criteria_obj.name}" + if criteria_obj.ok(): + app.result.success_criteria[criteria_name] = "PASSED" + else: + app.result.success_criteria[criteria_name] = "FAILED" + + for context_key, fom_map in fom_values.items(): + metadata = context_metadata[context_key] + context_map = { + "name": metadata["name"], + "foms": [], + "display_name": _get_context_display_name(metadata["name"]), + "context_def_name": metadata["def_name"], + "context_vars": metadata["vars"], + } + + for fom_name, fom in fom_map.items(): + fom_copy = fom.copy() + fom_copy["name"] = fom_name + context_map["foms"].append(fom_copy) + + if metadata["name"] == _NULL_CONTEXT: + app.result.contexts.insert(0, context_map) + else: + app.result.contexts.append(context_map) diff --git a/lib/ramble/ramble/test/test_analysis_strategy.py b/lib/ramble/ramble/test/test_analysis_strategy.py index 794529094..1e9e54e27 100644 --- a/lib/ramble/ramble/test/test_analysis_strategy.py +++ b/lib/ramble/ramble/test/test_analysis_strategy.py @@ -25,5 +25,180 @@ def test_get_strategy(): assert isinstance(strategy, DefaultAnalysisStrategy) assert strategy.app_inst is app + from ramble.analysis.backwards import BackwardsAnalysisStrategy + + strategy_backwards = ramble.analysis.get_strategy("backwards", app) + assert isinstance(strategy_backwards, BackwardsAnalysisStrategy) + assert strategy_backwards.app_inst is app + with pytest.raises(ValueError, match="Unknown analysis strategy: invalid"): ramble.analysis.get_strategy("invalid", app) + + +def test_read_file_backwards(tmpdir): + from ramble.analysis.backwards import _read_file_backwards + + temp_file = tmpdir.join("test_backwards.txt") + lines = ["line 1\n", "line 2\n", "line 3\n"] + temp_file.write("".join(lines)) + + result = list(_read_file_backwards(str(temp_file))) + assert result == ["line 3\n", "line 2\n", "line 1\n"] + + +def test_backwards_strategy_validation(): + from ramble.analysis.backwards import BackwardsAnalysisStrategy + + class MockResult: + def read_cache(self, workspace, app): + return False + + class MockApp: + def __init__(self): + self.success_list = None + self.get_status = lambda: None + self.result = MockResult() + + def analysis_dicts(self, criteria_list): + files = {"test.log": {"contexts": {"some_context": ["fom1"]}, "success_criteria": []}} + return files, {}, {} + + app = MockApp() + app.analysis_strategy = "backwards" + strategy = BackwardsAnalysisStrategy(app) + + class MockWorkspace: + dry_run = False + + msg = "BackwardsAnalysisStrategy cannot be used because context" + with pytest.raises(ValueError, match=msg): + strategy(MockWorkspace()) + + +def test_backwards_strategy_fallback(monkeypatch): + from ramble.analysis.backwards import BackwardsAnalysisStrategy + + called_default = False + + class MockDefaultStrategy: + def __init__(self, app): + pass + + def __call__(self, workspace): + nonlocal called_default + called_default = True + + monkeypatch.setitem(ramble.analysis._strategy_registry, "default", MockDefaultStrategy) + + class MockResult: + def read_cache(self, workspace, app): + return False + + class MockApp: + def __init__(self): + self.success_list = None + self.get_status = lambda: None + self.result = MockResult() + self.analysis_strategy = None + + def analysis_dicts(self, criteria_list): + files = {"test.log": {"contexts": {"some_context": ["fom1"]}, "success_criteria": []}} + return files, {}, {} + + app = MockApp() + strategy = BackwardsAnalysisStrategy(app) + + class MockWorkspace: + dry_run = False + + strategy(MockWorkspace()) + assert called_default + + +def test_backwards_strategy_dynamic_foms_validation(): + from ramble.analysis.backwards import BackwardsAnalysisStrategy + + class MockResult: + def read_cache(self, workspace, app): + return False + + class MockApp: + def __init__(self): + self.success_list = None + self.get_status = lambda: None + self.result = MockResult() + + def analysis_dicts(self, criteria_list): + files = {"test.log": {"contexts": {"null": ["fom1"]}, "success_criteria": []}} + f_defs = { + "null": { + "foms": { + "fom1": { + "fom_name_expanded": None, + "units_expanded": "s", + } + } + } + } + return files, f_defs, {} + + app = MockApp() + app.analysis_strategy = "backwards" + strategy = BackwardsAnalysisStrategy(app) + + class MockWorkspace: + dry_run = False + + msg = "BackwardsAnalysisStrategy cannot be used because FOM 'fom1' has dynamic name or units." + with pytest.raises(ValueError, match=msg): + strategy(MockWorkspace()) + + +def test_backwards_strategy_dynamic_foms_fallback(monkeypatch): + from ramble.analysis.backwards import BackwardsAnalysisStrategy + + called_default = False + + class MockDefaultStrategy: + def __init__(self, app): + pass + + def __call__(self, workspace): + nonlocal called_default + called_default = True + + monkeypatch.setitem(ramble.analysis._strategy_registry, "default", MockDefaultStrategy) + + class MockResult: + def read_cache(self, workspace, app): + return False + + class MockApp: + def __init__(self): + self.success_list = None + self.get_status = lambda: None + self.result = MockResult() + self.analysis_strategy = None + + def analysis_dicts(self, criteria_list): + files = {"test.log": {"contexts": {"null": ["fom1"]}, "success_criteria": []}} + f_defs = { + "null": { + "foms": { + "fom1": { + "fom_name_expanded": "fom1", + "units_expanded": None, + } + } + } + } + return files, f_defs, {} + + app = MockApp() + strategy = BackwardsAnalysisStrategy(app) + + class MockWorkspace: + dry_run = False + + strategy(MockWorkspace()) + assert called_default diff --git a/var/ramble/repos/builtin/base_classes/application-base/base_class.py b/var/ramble/repos/builtin/base_classes/application-base/base_class.py index 5bbbb9d5a..b0cfe2d50 100644 --- a/var/ramble/repos/builtin/base_classes/application-base/base_class.py +++ b/var/ramble/repos/builtin/base_classes/application-base/base_class.py @@ -3465,7 +3465,7 @@ def _analyze_experiments(self, workspace, app_inst=None): """ import ramble.analysis - strategy_name = getattr(self, "analysis_strategy", "default") + strategy_name = getattr(self, "analysis_strategy", None) or "backwards" strategy = ramble.analysis.get_strategy(strategy_name, self) strategy(workspace) From 068a75b9424ceda16bf31f265548473bc740fb81 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 17 Aug 2026 13:32:07 -0600 Subject: [PATCH 3/5] Rename default analysis strategy to forward Signed-off-by: Bob --- lib/ramble/ramble/analysis/__init__.py | 4 +-- lib/ramble/ramble/analysis/backwards.py | 13 ++++---- .../analysis/{default.py => forward.py} | 6 ++-- .../ramble/test/test_analysis_strategy.py | 30 +++++++++---------- 4 files changed, 26 insertions(+), 27 deletions(-) rename lib/ramble/ramble/analysis/{default.py => forward.py} (98%) diff --git a/lib/ramble/ramble/analysis/__init__.py b/lib/ramble/ramble/analysis/__init__.py index d6c285bae..f8d88e67e 100644 --- a/lib/ramble/ramble/analysis/__init__.py +++ b/lib/ramble/ramble/analysis/__init__.py @@ -12,10 +12,10 @@ # flake8: noqa: F401 from ramble.analysis.base import AnalysisStrategyBase as AnalysisStrategyBase -from ramble.analysis.default import DefaultAnalysisStrategy +from ramble.analysis.forward import ForwardAnalysisStrategy _strategy_registry = { - "default": DefaultAnalysisStrategy, + "forward": ForwardAnalysisStrategy, "backwards": BackwardsAnalysisStrategy, } diff --git a/lib/ramble/ramble/analysis/backwards.py b/lib/ramble/ramble/analysis/backwards.py index 54d1dc4dc..67405c39c 100644 --- a/lib/ramble/ramble/analysis/backwards.py +++ b/lib/ramble/ramble/analysis/backwards.py @@ -83,11 +83,10 @@ def __call__(self, workspace): if context != _NULL_CONTEXT: if getattr(app, "analysis_strategy", None) is None: logger.debug( - "Falling back to default forward-reading strategy due to non-null " - "context." + "Falling back to forward-reading strategy due to non-null context." ) - default_strategy = ramble.analysis.get_strategy("default", app) - return default_strategy(workspace) + forward_strategy = ramble.analysis.get_strategy("forward", app) + return forward_strategy(workspace) else: raise ValueError( f"BackwardsAnalysisStrategy cannot be used because " @@ -104,11 +103,11 @@ def __call__(self, workspace): ): if getattr(app, "analysis_strategy", None) is None: logger.debug( - "Falling back to default forward-reading strategy due to dynamic " + "Falling back to forward-reading strategy due to dynamic " "FOM name or units." ) - default_strategy = ramble.analysis.get_strategy("default", app) - return default_strategy(workspace) + forward_strategy = ramble.analysis.get_strategy("forward", app) + return forward_strategy(workspace) else: raise ValueError( f"BackwardsAnalysisStrategy cannot be used because " diff --git a/lib/ramble/ramble/analysis/default.py b/lib/ramble/ramble/analysis/forward.py similarity index 98% rename from lib/ramble/ramble/analysis/default.py rename to lib/ramble/ramble/analysis/forward.py index ec95bab47..b1aff9b81 100644 --- a/lib/ramble/ramble/analysis/default.py +++ b/lib/ramble/ramble/analysis/forward.py @@ -6,7 +6,7 @@ # option. This file may not be copied, modified, or distributed # except according to those terms. -"""Define the default analysis strategy""" +"""Define the forward-reading analysis strategy""" import os import string @@ -26,8 +26,8 @@ def _get_context_display_name(context): ) -class DefaultAnalysisStrategy(AnalysisStrategyBase): - """Default regex-based analysis/extraction strategy.""" +class ForwardAnalysisStrategy(AnalysisStrategyBase): + """Forward regex-based analysis/extraction strategy.""" def __call__(self, workspace): app = self.app_inst diff --git a/lib/ramble/ramble/test/test_analysis_strategy.py b/lib/ramble/ramble/test/test_analysis_strategy.py index 1e9e54e27..27f31b29c 100644 --- a/lib/ramble/ramble/test/test_analysis_strategy.py +++ b/lib/ramble/ramble/test/test_analysis_strategy.py @@ -11,7 +11,7 @@ import pytest import ramble.analysis -from ramble.analysis.default import DefaultAnalysisStrategy +from ramble.analysis.forward import ForwardAnalysisStrategy class DummyApp: @@ -21,8 +21,8 @@ class DummyApp: def test_get_strategy(): app = DummyApp() - strategy = ramble.analysis.get_strategy("default", app) - assert isinstance(strategy, DefaultAnalysisStrategy) + strategy = ramble.analysis.get_strategy("forward", app) + assert isinstance(strategy, ForwardAnalysisStrategy) assert strategy.app_inst is app from ramble.analysis.backwards import BackwardsAnalysisStrategy @@ -78,17 +78,17 @@ class MockWorkspace: def test_backwards_strategy_fallback(monkeypatch): from ramble.analysis.backwards import BackwardsAnalysisStrategy - called_default = False + called_forward = False - class MockDefaultStrategy: + class MockForwardStrategy: def __init__(self, app): pass def __call__(self, workspace): - nonlocal called_default - called_default = True + nonlocal called_forward + called_forward = True - monkeypatch.setitem(ramble.analysis._strategy_registry, "default", MockDefaultStrategy) + monkeypatch.setitem(ramble.analysis._strategy_registry, "forward", MockForwardStrategy) class MockResult: def read_cache(self, workspace, app): @@ -112,7 +112,7 @@ class MockWorkspace: dry_run = False strategy(MockWorkspace()) - assert called_default + assert called_forward def test_backwards_strategy_dynamic_foms_validation(): @@ -157,17 +157,17 @@ class MockWorkspace: def test_backwards_strategy_dynamic_foms_fallback(monkeypatch): from ramble.analysis.backwards import BackwardsAnalysisStrategy - called_default = False + called_forward = False - class MockDefaultStrategy: + class MockForwardStrategy: def __init__(self, app): pass def __call__(self, workspace): - nonlocal called_default - called_default = True + nonlocal called_forward + called_forward = True - monkeypatch.setitem(ramble.analysis._strategy_registry, "default", MockDefaultStrategy) + monkeypatch.setitem(ramble.analysis._strategy_registry, "forward", MockForwardStrategy) class MockResult: def read_cache(self, workspace, app): @@ -201,4 +201,4 @@ class MockWorkspace: dry_run = False strategy(MockWorkspace()) - assert called_default + assert called_forward From 3ac9debc093cb1e57372d1178f738883371e7f4b Mon Sep 17 00:00:00 2001 From: rfbgo <109985755+rfbgo@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:37:45 -0600 Subject: [PATCH 4/5] Apply suggestions from code review Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Signed-off-by: Bob --- lib/ramble/ramble/analysis/backwards.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/ramble/ramble/analysis/backwards.py b/lib/ramble/ramble/analysis/backwards.py index 67405c39c..cdf5742fe 100644 --- a/lib/ramble/ramble/analysis/backwards.py +++ b/lib/ramble/ramble/analysis/backwards.py @@ -50,7 +50,7 @@ def _read_file_backwards(file_path, block_size=4096): for line in reversed(lines[1:]): yield line.decode("utf-8", errors="replace") + "\n" - if buffer: + if file_size > 0: yield buffer.decode("utf-8", errors="replace") + "\n" @@ -85,6 +85,7 @@ def __call__(self, workspace): logger.debug( "Falling back to forward-reading strategy due to non-null context." ) + import ramble.analysis forward_strategy = ramble.analysis.get_strategy("forward", app) return forward_strategy(workspace) else: @@ -106,6 +107,7 @@ def __call__(self, workspace): "Falling back to forward-reading strategy due to dynamic " "FOM name or units." ) + import ramble.analysis forward_strategy = ramble.analysis.get_strategy("forward", app) return forward_strategy(workspace) else: From 26dff11341574cbc686bec74ae52201238f9ece6 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 17 Aug 2026 13:46:11 -0600 Subject: [PATCH 5/5] fix incorrect gemini 'fix' Signed-off-by: Bob --- lib/ramble/ramble/analysis/backwards.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/lib/ramble/ramble/analysis/backwards.py b/lib/ramble/ramble/analysis/backwards.py index cdf5742fe..ff41d11ec 100644 --- a/lib/ramble/ramble/analysis/backwards.py +++ b/lib/ramble/ramble/analysis/backwards.py @@ -85,8 +85,9 @@ def __call__(self, workspace): logger.debug( "Falling back to forward-reading strategy due to non-null context." ) - import ramble.analysis - forward_strategy = ramble.analysis.get_strategy("forward", app) + import ramble.analysis as ra + + forward_strategy = ra.get_strategy("forward", app) return forward_strategy(workspace) else: raise ValueError( @@ -107,8 +108,9 @@ def __call__(self, workspace): "Falling back to forward-reading strategy due to dynamic " "FOM name or units." ) - import ramble.analysis - forward_strategy = ramble.analysis.get_strategy("forward", app) + import ramble.analysis as ra + + forward_strategy = ra.get_strategy("forward", app) return forward_strategy(workspace) else: raise ValueError(