diff --git a/aiopslab/orchestrator/orchestrator.py b/aiopslab/orchestrator/orchestrator.py index 055ff8ec..b9b723fa 100644 --- a/aiopslab/orchestrator/orchestrator.py +++ b/aiopslab/orchestrator/orchestrator.py @@ -1,235 +1,258 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - -"""Orchestrator class that interfaces with the agent and the environment.""" - -from aiopslab.service.helm import Helm -from aiopslab.service.kubectl import KubeCtl -from aiopslab.session import Session -from aiopslab.orchestrator.problems.registry import ProblemRegistry -from aiopslab.orchestrator.parser import ResponseParser -from aiopslab.utils.status import * -from aiopslab.utils.critical_section import CriticalSection -from aiopslab.service.telemetry.prometheus import Prometheus -import time -import inspect -import asyncio -import atexit -import os - - -class Orchestrator: - def __init__(self, results_dir=None): - self.agent = None - self.session = None - self.parser = ResponseParser() - self.probs = ProblemRegistry() - self.sprint = SessionPrint() - self.execution_start_time = None - self.execution_end_time = None - self.kubectl = KubeCtl() - self.use_wandb = os.getenv("USE_WANDB", "false").lower() == "true" - self.results_dir = results_dir - - def init_problem(self, problem_id: str): - """Initialize a problem instance for the agent to solve. - - Args: - problem_id (str): The problem instance identifier. - - Returns: - tuple: A tuple containing the problem description, task message, and session object. - """ - # Start timer - self.execution_start_time = time.time() - - self.session = Session(results_dir=self.results_dir) - print(f"Session ID: {self.session.session_id}") - prob = self.probs.get_problem_instance(problem_id) - deployment = self.probs.get_problem_deployment(problem_id) - self.session.set_problem(prob, pid=problem_id) - self.session.set_agent(self.agent_name) - - if deployment != "docker": - print("Setting up OpenEBS...") - - # Install OpenEBS - self.kubectl.exec_command( - "kubectl apply -f https://openebs.github.io/charts/openebs-operator.yaml" - ) - self.kubectl.exec_command( - "kubectl patch storageclass openebs-hostpath -p '{\"metadata\": {\"annotations\":{\"storageclass.kubernetes.io/is-default-class\":\"true\"}}}'" - ) - self.kubectl.wait_for_ready("openebs") - print("OpenEBS setup completed.") - - # Setup and deploy Prometheus - self.prometheus = Prometheus() - self.prometheus.deploy() - - # deploy service - prob.app.delete() - prob.app.deploy() - - # make sure is_fault_injected is correct to apply appropriate - # function with atexit to recover fault - with CriticalSection(): - # inject fault - prob.inject_fault() - atexit.register(exit_cleanup_fault, prob=prob) - - # Check if start_workload is async or sync - if inspect.iscoroutinefunction(prob.start_workload): - asyncio.create_task(prob.start_workload()) - else: - prob.start_workload() - - task_desc = prob.get_task_description() - instructions = prob.get_instructions() - actions = prob.get_available_actions() - - return task_desc, instructions, actions - - def register_agent(self, agent, name="agent"): - """Register the agent for the current session. - - Args: - agent: The agent to register. - name: The name of the agent (default: "agent"). - """ - self.agent = agent - self.agent_name = name - - async def ask_agent(self, input): - """Ask the agent for the next action given the current context.""" - assert self.session is not None - assert self.agent is not None - - agent_response = await self.agent.get_action(input) - self.session.add({"role": "assistant", "content": agent_response}) - - return agent_response - - async def ask_env(self, input): - """Ask the environment for the observation given the current action.""" - assert self.session is not None - - try: - resp = self.parser.parse(input) - except ResponseParsingError as e: - self.session.add({"role": "env", "content": str(e)}) - return str(e) - - api, args, kwargs = resp["api_name"], resp["args"], resp["kwargs"] - - # if submit, save solution for eval - if api == "submit": - self.session.set_solution(args[0] if len(args) == 1 else args) - - try: - env_response = self.session.problem.perform_action(api, *args, **kwargs) - - if hasattr(env_response, "error"): - env_response = str(env_response) - print("An error occurred:", env_response) - except InvalidActionError as e: - env_response = str(e) - except Exception as e: - env_response = str(e) - print("Unhandled exception:", e) - - self.session.add({"role": "env", "content": env_response}) - - return env_response - - async def start_problem(self, max_steps: int): - """Start the task and run for a specified number of steps. - - Args: - max_steps (int): The maximum number of steps to run the task. - - Returns: - dict: The final state of the session. - """ - assert self.session is not None - action_instr = "Please take the next action" - action, env_response, results = "", "", {} - self.session.start() - - # catch any exception and recover fault before the users catch it - try: - for step in range(max_steps): - action = await self.ask_agent(action_instr) - self.sprint.agent(action) - - env_response = await self.ask_env(action) - self.sprint.service(env_response) - - if env_response == SubmissionStatus.VALID_SUBMISSION: - break - elif env_response == SubmissionStatus.INVALID_SUBMISSION: - raise ValueError("Invalid submission!") # TODO (@manish): ask to retry? - - action_instr = env_response + "\n" + "Please take the next action" - except Exception as e: - # Make sure the fault cleanup function is unregistered - # after recovering fault ahead because of exceptions - with CriticalSection(): - print("Some exception happened. Recovering the injected fault...") - self.session.problem.recover_fault() - atexit.unregister(exit_cleanup_fault) - raise e - - self.session.end() - - # A valid submission was made (or) max_steps reached - if env_response != SubmissionStatus.INVALID_SUBMISSION: - results = self.session.problem.eval( - self.session.solution, self.session.history, self.session.get_duration() - ) - self.sprint.result(results) - - self.session.set_results(results) - self.session.to_json() - if self.use_wandb: - self.session.to_wandb() - - with CriticalSection(): - self.session.problem.recover_fault() - atexit.unregister(exit_cleanup_fault) - - # Beyond recovering from fault, - # I feel sometimes it is safer to delete the whole namespace. - # But this will take more time. - # if not self.session.problem.sys_status_after_recovery(): - self.session.problem.app.cleanup() - - if self.session.problem.namespace != "docker": - self.prometheus.teardown() - print("Uninstalling OpenEBS...") - self.kubectl.exec_command("kubectl delete sc openebs-hostpath openebs-device --ignore-not-found") - self.kubectl.exec_command( - "kubectl delete -f https://openebs.github.io/charts/openebs-operator.yaml" - ) - self.kubectl.wait_for_namespace_deletion("openebs") - - self.execution_end_time = time.time() - total_execution_time = self.execution_end_time - self.execution_start_time - time_keys = ["TTD", "TTL", "TTA", "TTM"] - key = next((k for k in time_keys if k in results), None) - framework_overhead = ( - total_execution_time - results[key] - ) # Time spent doing everything besides running the agent - print(f"Framework overhead: {framework_overhead}") - - return { - "history": self.session.history, - "final_state": env_response, - "results": results, - "framework_overhead": framework_overhead, - } - - -def exit_cleanup_fault(prob): - print("Recovering fault before exit...") - prob.recover_fault() +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Orchestrator class that interfaces with the agent and the environment.""" + +from aiopslab.service.helm import Helm +from aiopslab.service.kubectl import KubeCtl +from aiopslab.session import Session +from aiopslab.orchestrator.problems.registry import ProblemRegistry +from aiopslab.orchestrator.parser import ResponseParser +from aiopslab.utils.status import * +from aiopslab.utils.critical_section import CriticalSection +from aiopslab.service.telemetry.prometheus import Prometheus +from aiopslab.timing import FAULT_OCCURRED, MITIGATION_COMPLETED +import time +import inspect +import asyncio +import atexit +import os + + +class Orchestrator: + def __init__(self, results_dir=None): + self.agent = None + self.session = None + self.parser = ResponseParser() + self.probs = ProblemRegistry() + self.sprint = SessionPrint() + self.execution_start_time = None + self.execution_end_time = None + self.kubectl = KubeCtl() + self.use_wandb = os.getenv("USE_WANDB", "false").lower() == "true" + self.results_dir = results_dir + + def init_problem(self, problem_id: str): + """Initialize a problem instance for the agent to solve. + + Args: + problem_id (str): The problem instance identifier. + + Returns: + tuple: A tuple containing the problem description, task message, and session object. + """ + # Start timer + self.execution_start_time = time.time() + + self.session = Session(results_dir=self.results_dir) + print(f"Session ID: {self.session.session_id}") + prob = self.probs.get_problem_instance(problem_id) + deployment = self.probs.get_problem_deployment(problem_id) + self.session.set_problem(prob, pid=problem_id) + self.session.set_agent(self.agent_name) + if hasattr(prob, "set_timing"): + prob.set_timing(self.session.timing) + + if deployment != "docker": + print("Setting up OpenEBS...") + + # Install OpenEBS + self.kubectl.exec_command( + "kubectl apply -f https://openebs.github.io/charts/openebs-operator.yaml" + ) + self.kubectl.exec_command( + "kubectl patch storageclass openebs-hostpath -p '{\"metadata\": {\"annotations\":{\"storageclass.kubernetes.io/is-default-class\":\"true\"}}}'" + ) + self.kubectl.wait_for_ready("openebs") + print("OpenEBS setup completed.") + + # Setup and deploy Prometheus + self.prometheus = Prometheus() + self.prometheus.deploy() + + # deploy service + prob.app.delete() + prob.app.deploy() + + # make sure is_fault_injected is correct to apply appropriate + # function with atexit to recover fault + with CriticalSection(): + # inject fault + prob.inject_fault() + self.session.timing.mark(FAULT_OCCURRED) + atexit.register(exit_cleanup_fault, prob=prob) + + # Check if start_workload is async or sync + if inspect.iscoroutinefunction(prob.start_workload): + asyncio.create_task(prob.start_workload()) + else: + prob.start_workload() + + task_desc = prob.get_task_description() + instructions = prob.get_instructions() + actions = prob.get_available_actions() + + return task_desc, instructions, actions + + def register_agent(self, agent, name="agent"): + """Register the agent for the current session. + + Args: + agent: The agent to register. + name: The name of the agent (default: "agent"). + """ + self.agent = agent + self.agent_name = name + + async def ask_agent(self, input): + """Ask the agent for the next action given the current context.""" + assert self.session is not None + assert self.agent is not None + + agent_response = await self.agent.get_action(input) + self.session.add({"role": "assistant", "content": agent_response}) + + return agent_response + + async def ask_env(self, input): + """Ask the environment for the observation given the current action.""" + assert self.session is not None + + try: + resp = self.parser.parse(input) + except ResponseParsingError as e: + self.session.add({"role": "env", "content": str(e)}) + return str(e) + + api, args, kwargs = resp["api_name"], resp["args"], resp["kwargs"] + + # if submit, save solution for eval + if api == "submit": + self.session.set_solution(args[0] if len(args) == 1 else args) + + try: + env_response = self.session.problem.perform_action(api, *args, **kwargs) + + if hasattr(env_response, "error"): + env_response = str(env_response) + print("An error occurred:", env_response) + except InvalidActionError as e: + env_response = str(e) + except Exception as e: + env_response = str(e) + print("Unhandled exception:", e) + + if env_response == SubmissionStatus.VALID_SUBMISSION: + completion_event = getattr( + self.session.problem, "timing_completion_event", None + ) + # Mitigation completion is oracle-driven and is recorded after eval(). + if completion_event and completion_event != MITIGATION_COMPLETED: + self.session.timing.mark(completion_event) + + self.session.add({"role": "env", "content": env_response}) + + return env_response + + async def start_problem(self, max_steps: int): + """Start the task and run for a specified number of steps. + + Args: + max_steps (int): The maximum number of steps to run the task. + + Returns: + dict: The final state of the session. + """ + assert self.session is not None + action_instr = "Please take the next action" + action, env_response, results = "", "", {} + self.session.start() + + # catch any exception and recover fault before the users catch it + try: + for step in range(max_steps): + action = await self.ask_agent(action_instr) + self.sprint.agent(action) + + env_response = await self.ask_env(action) + self.sprint.service(env_response) + + if env_response == SubmissionStatus.VALID_SUBMISSION: + break + elif env_response == SubmissionStatus.INVALID_SUBMISSION: + raise ValueError("Invalid submission!") # TODO (@manish): ask to retry? + + action_instr = env_response + "\n" + "Please take the next action" + except Exception as e: + # Make sure the fault cleanup function is unregistered + # after recovering fault ahead because of exceptions + with CriticalSection(): + print("Some exception happened. Recovering the injected fault...") + self.session.problem.recover_fault() + atexit.unregister(exit_cleanup_fault) + raise e + + self.session.end() + + # A valid submission was made (or) max_steps reached + if env_response != SubmissionStatus.INVALID_SUBMISSION: + results = self.session.problem.eval( + self.session.solution, self.session.history, self.session.get_duration() + ) + + # Mitigation has a stronger completion boundary than submit(): the + # task-specific evaluator must first confirm successful recovery. + if ( + results.get("success") + and getattr(self.session.problem, "timing_completion_event", None) + == MITIGATION_COMPLETED + ): + self.session.timing.mark(MITIGATION_COMPLETED) + + results["timing_events"] = self.session.timing.to_dict() + self.sprint.result(results) + + self.session.set_results(results) + self.session.to_json() + if self.use_wandb: + self.session.to_wandb() + + with CriticalSection(): + self.session.problem.recover_fault() + atexit.unregister(exit_cleanup_fault) + + # Beyond recovering from fault, + # I feel sometimes it is safer to delete the whole namespace. + # But this will take more time. + # if not self.session.problem.sys_status_after_recovery(): + self.session.problem.app.cleanup() + + if self.session.problem.namespace != "docker": + self.prometheus.teardown() + print("Uninstalling OpenEBS...") + self.kubectl.exec_command("kubectl delete sc openebs-hostpath openebs-device --ignore-not-found") + self.kubectl.exec_command( + "kubectl delete -f https://openebs.github.io/charts/openebs-operator.yaml" + ) + self.kubectl.wait_for_namespace_deletion("openebs") + + self.execution_end_time = time.time() + total_execution_time = self.execution_end_time - self.execution_start_time + time_keys = ["TTD", "TTL", "TTA", "TTM"] + key = next((k for k in time_keys if k in results), None) + framework_overhead = ( + total_execution_time - results[key] + ) # Time spent doing everything besides running the agent + print(f"Framework overhead: {framework_overhead}") + + return { + "history": self.session.history, + "final_state": env_response, + "results": results, + "framework_overhead": framework_overhead, + } + + +def exit_cleanup_fault(prob): + print("Recovering fault before exit...") + prob.recover_fault() diff --git a/aiopslab/orchestrator/tasks/analysis.py b/aiopslab/orchestrator/tasks/analysis.py index 1186a61b..2f00cee2 100644 --- a/aiopslab/orchestrator/tasks/analysis.py +++ b/aiopslab/orchestrator/tasks/analysis.py @@ -10,6 +10,7 @@ from aiopslab.orchestrator.actions.analysis import AnalysisActions from aiopslab.service.apps.base import Application from aiopslab.session import SessionItem +from aiopslab.timing import ANALYSIS_COMPLETED from aiopslab.utils.actions import get_actions from aiopslab.utils.status import InvalidActionError @@ -17,6 +18,8 @@ class AnalysisTask(Task): """An AIOps root cause analysis task.""" + timing_completion_event = ANALYSIS_COMPLETED + def __init__(self, app: Application): super().__init__() self.app = app diff --git a/aiopslab/orchestrator/tasks/base.py b/aiopslab/orchestrator/tasks/base.py index 5c039921..3838bf48 100644 --- a/aiopslab/orchestrator/tasks/base.py +++ b/aiopslab/orchestrator/tasks/base.py @@ -6,6 +6,7 @@ from aiopslab.service.kubectl import KubeCtl from aiopslab.orchestrator.evaluators.quantitative import * from aiopslab.orchestrator.evaluators.qualitative import LLMJudge +from aiopslab.timing import EvaluationTiming config = Config(BASE_DIR / "config.yml") @@ -14,9 +15,21 @@ class Task: """Base class for all tasks.""" + timing_completion_event = None + def __init__(self): self.results = {} self.kubectl = KubeCtl() + self.timing: EvaluationTiming | None = None + + def set_timing(self, timing: EvaluationTiming): + """Attach the session timing recorder to this task.""" + self.timing = timing + + def mark_timing_completion(self): + """Record this task's completion event when an event boundary is defined.""" + if self.timing is not None and self.timing_completion_event: + self.timing.mark(self.timing_completion_event) def get_task_description(self): raise NotImplementedError("Subclasses must implement this method.") diff --git a/aiopslab/orchestrator/tasks/detection.py b/aiopslab/orchestrator/tasks/detection.py index 449a86cb..48a8d59b 100644 --- a/aiopslab/orchestrator/tasks/detection.py +++ b/aiopslab/orchestrator/tasks/detection.py @@ -1,78 +1,81 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - -"""Define and query information about an AIOps Detection task.""" - -import textwrap -from typing import Any - - -from aiopslab.orchestrator.tasks.base import Task -from aiopslab.orchestrator.actions.detection import DetectionActions -from aiopslab.service.apps.base import Application -from aiopslab.session import SessionItem -from aiopslab.utils.actions import get_actions -from aiopslab.utils.status import InvalidActionError - - -class DetectionTask(Task): - """An AIOps anomaly detection task.""" - - def __init__(self, app: Application): - super().__init__() - self.app = app - self.helm_configs = self.app.helm_configs - self.app_summary = self.app.get_app_summary() - self.actions = DetectionActions() - - self.task_desc = """\ - You are an expert DevOps engineer assigned to detect anomalies in a deployed service. - - Service Details: - {app_summary} - - You will begin by analyzing the service's state and telemetry, and then submit your solution: - 1. str: "Yes" or "No" to indicate whether there are anomalies detected - """ - - self.instructions = """\ - You will respond with one of the above APIs as your next action. - Please respond in the following format in a markdown code block: - ```\n(, ...)\n``` - - For instance, if you want to list files in current directory, your response must be exactly: - - ```\nexec_shell("ls -l")\n``` - - If you decide that there are no anomalies: - - ```\nsubmit(\"No\")\n``` - - Or, if anomalies are found: - - ```\nsubmit(\"Yes\")\n``` - - Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes. - """ - - def get_task_description(self): - return textwrap.dedent(self.task_desc).format(app_summary=self.app_summary) - - def get_instructions(self): - return textwrap.dedent(self.instructions) - - def get_available_actions(self): - return get_actions(task="detection") - - def perform_action(self, action_name, *args, **kwargs): - action_method = getattr(self.actions, action_name, None) - - if action_method is not None and callable(action_method): - return action_method(*args, **kwargs) - else: - raise InvalidActionError(action_name) - - def eval(self, soln: Any, trace: list[SessionItem], duration: float): - self.add_result("TTD", duration) - self.common_eval(trace) - return self.results +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Define and query information about an AIOps Detection task.""" + +import textwrap +from typing import Any + + +from aiopslab.orchestrator.tasks.base import Task +from aiopslab.orchestrator.actions.detection import DetectionActions +from aiopslab.service.apps.base import Application +from aiopslab.session import SessionItem +from aiopslab.timing import DETECTION_COMPLETED +from aiopslab.utils.actions import get_actions +from aiopslab.utils.status import InvalidActionError + + +class DetectionTask(Task): + """An AIOps anomaly detection task.""" + + timing_completion_event = DETECTION_COMPLETED + + def __init__(self, app: Application): + super().__init__() + self.app = app + self.helm_configs = self.app.helm_configs + self.app_summary = self.app.get_app_summary() + self.actions = DetectionActions() + + self.task_desc = """\ + You are an expert DevOps engineer assigned to detect anomalies in a deployed service. + + Service Details: + {app_summary} + + You will begin by analyzing the service's state and telemetry, and then submit your solution: + 1. str: "Yes" or "No" to indicate whether there are anomalies detected + """ + + self.instructions = """\ + You will respond with one of the above APIs as your next action. + Please respond in the following format in a markdown code block: + ```\n(, ...)\n``` + + For instance, if you want to list files in current directory, your response must be exactly: + + ```\nexec_shell("ls -l")\n``` + + If you decide that there are no anomalies: + + ```\nsubmit(\"No\")\n``` + + Or, if anomalies are found: + + ```\nsubmit(\"Yes\")\n``` + + Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes. + """ + + def get_task_description(self): + return textwrap.dedent(self.task_desc).format(app_summary=self.app_summary) + + def get_instructions(self): + return textwrap.dedent(self.instructions) + + def get_available_actions(self): + return get_actions(task="detection") + + def perform_action(self, action_name, *args, **kwargs): + action_method = getattr(self.actions, action_name, None) + + if action_method is not None and callable(action_method): + return action_method(*args, **kwargs) + else: + raise InvalidActionError(action_name) + + def eval(self, soln: Any, trace: list[SessionItem], duration: float): + self.add_result("TTD", duration) + self.common_eval(trace) + return self.results diff --git a/aiopslab/orchestrator/tasks/localization.py b/aiopslab/orchestrator/tasks/localization.py index 6b3af40d..a05a5446 100644 --- a/aiopslab/orchestrator/tasks/localization.py +++ b/aiopslab/orchestrator/tasks/localization.py @@ -11,6 +11,7 @@ from aiopslab.orchestrator.actions.localization import LocalizationActions from aiopslab.service.apps.base import Application from aiopslab.session import SessionItem +from aiopslab.timing import LOCALIZATION_COMPLETED from aiopslab.utils.actions import get_actions from aiopslab.utils.status import InvalidActionError @@ -18,6 +19,8 @@ class LocalizationTask(Task): """An AIOps fault localization task.""" + timing_completion_event = LOCALIZATION_COMPLETED + def __init__(self, app: Application): super().__init__() self.app = app @@ -54,7 +57,8 @@ def __init__(self, app: Application): Or, if no faults are found: - ```\nsubmit([])\n``` + ```\nsubmit([]) + ``` Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes. """ diff --git a/aiopslab/orchestrator/tasks/mitigation.py b/aiopslab/orchestrator/tasks/mitigation.py index d15b62dd..e53e0564 100644 --- a/aiopslab/orchestrator/tasks/mitigation.py +++ b/aiopslab/orchestrator/tasks/mitigation.py @@ -10,6 +10,7 @@ from aiopslab.orchestrator.actions.mitigation import MitigationActions from aiopslab.service.apps.base import Application from aiopslab.session import SessionItem +from aiopslab.timing import MITIGATION_COMPLETED from aiopslab.utils.actions import get_actions from aiopslab.utils.status import InvalidActionError @@ -17,6 +18,10 @@ class MitigationTask(Task): """An AIOps anomaly mitigation task.""" + # Mitigation completion is recorded after the task-specific recovery oracle + # confirms that the environment is healthy, not merely when submit() is called. + timing_completion_event = MITIGATION_COMPLETED + def __init__(self, app: Application): super().__init__() self.app = app diff --git a/aiopslab/session.py b/aiopslab/session.py index 05c5f710..dd93bdf9 100644 --- a/aiopslab/session.py +++ b/aiopslab/session.py @@ -10,6 +10,7 @@ from pydantic import BaseModel from aiopslab.paths import RESULTS_DIR +from aiopslab.timing import EvaluationTiming class SessionItem(BaseModel): @@ -29,6 +30,7 @@ def __init__(self, results_dir=None) -> None: self.end_time = None self.agent_name = None self.results_dir = results_dir + self.timing = EvaluationTiming() def set_problem(self, problem, pid=None): """Set the problem instance for the session. @@ -60,7 +62,7 @@ def set_agent(self, agent_name): """Set the agent name for the session. Args: - agent_name (str): The name of the agent. + agent_name (str): The name of the agent (default: "agent"). """ self.agent_name = agent_name @@ -108,6 +110,7 @@ def to_dict(self): "problem_id": self.pid, "start_time": self.start_time, "end_time": self.end_time, + "timing_events": self.timing.to_dict(), "trace": [item.model_dump() for item in self.history], "results": self.results, } @@ -137,4 +140,6 @@ def from_json(self, filename: str): self.start_time = data.get("start_time") self.end_time = data.get("end_time") self.results = data.get("results") - self.history = [SessionItem.model_validate(item) for item in data.get("trace")] + self.history = [SessionItem.model_validate(item) for item in data.get("trace", [])] + for event, timestamp in data.get("timing_events", {}).items(): + self.timing.mark(event, timestamp) diff --git a/aiopslab/timing.py b/aiopslab/timing.py new file mode 100644 index 00000000..cf7f56ad --- /dev/null +++ b/aiopslab/timing.py @@ -0,0 +1,45 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Event-boundary timing primitives for AIOps evaluation.""" + +from __future__ import annotations + +import time +from typing import Dict + + +FAULT_OCCURRED = "fault_occurred" +DETECTION_COMPLETED = "detection_completed" +LOCALIZATION_COMPLETED = "localization_completed" +ANALYSIS_COMPLETED = "analysis_completed" +MITIGATION_COMPLETED = "mitigation_completed" + + +class EvaluationTiming: + """Record benchmark lifecycle events and derive elapsed intervals.""" + + def __init__(self) -> None: + self.events: Dict[str, float] = {} + + def mark(self, event: str, timestamp: float | None = None) -> float: + """Record an event timestamp and return the stored value.""" + value = time.time() if timestamp is None else timestamp + self.events[event] = value + return value + + def get(self, event: str) -> float | None: + """Return an event timestamp, if recorded.""" + return self.events.get(event) + + def elapsed(self, start_event: str, end_event: str) -> float | None: + """Return elapsed time between two recorded events.""" + start = self.get(start_event) + end = self.get(end_event) + if start is None or end is None: + return None + return end - start + + def to_dict(self) -> Dict[str, float]: + """Return a JSON-serializable snapshot of recorded events.""" + return dict(self.events) diff --git a/docs/evaluation_timing.md b/docs/evaluation_timing.md new file mode 100644 index 00000000..a5cdbe45 --- /dev/null +++ b/docs/evaluation_timing.md @@ -0,0 +1,32 @@ +# Event-Boundary Evaluation Timing + +AIOpsLab currently exposes `TTD`, `TTL`, `TTA`, and `TTM` from the session duration. That duration is useful for backward compatibility, but it does not by itself identify the benchmark lifecycle boundaries described by the research-aligned timing model. + +This PoW adds event instrumentation without changing the historical metric fields yet. + +## Canonical boundaries + +| Event | Meaning | +|---|---| +| `fault_occurred` | Fault injection has completed and the fault is active. | +| `detection_completed` | A valid detection submission has been accepted. | +| `localization_completed` | A valid localization submission has been accepted. | +| `analysis_completed` | A valid RCA/analysis submission has been accepted. | +| `mitigation_completed` | Mitigation has reached an oracle-confirmed completion point. | + +## Derived intervals + +The intended event-boundary model is: + +- `TTD = detection_completed - fault_occurred` +- `TTL = localization_completed - fault_occurred` +- `TTA = analysis_completed - fault_occurred` +- `TTM = mitigation_completed - detection_completed` + +The last four formulas are treated as the PoW target model; the primary AIOpsLab research source explicitly establishes TTD and TTM, while TTL/TTA remain operational extensions until independently sourced. + +## Compatibility rule + +Existing result fields are not rewritten by this PoW. Historical session-duration values remain valid as historical/legacy measurements. New event timestamps are persisted separately so a later PR can migrate the metric fields only after regression coverage demonstrates that all benchmark task types expose the required boundaries. + +No-op problems must not be interpreted as instantaneous fault detection merely because a session duration exists; future metric migration should represent the absence of a fault boundary explicitly. diff --git a/tests/unit/test_timing.py b/tests/unit/test_timing.py new file mode 100644 index 00000000..98e2f5ed --- /dev/null +++ b/tests/unit/test_timing.py @@ -0,0 +1,16 @@ +from aiopslab.timing import EvaluationTiming + + +def test_elapsed_uses_explicit_event_boundaries(): + timing = EvaluationTiming() + timing.mark("fault_occurred", 10.0) + timing.mark("detection_completed", 12.5) + + assert timing.elapsed("fault_occurred", "detection_completed") == 2.5 + + +def test_missing_boundary_does_not_fabricate_duration(): + timing = EvaluationTiming() + timing.mark("fault_occurred", 10.0) + + assert timing.elapsed("fault_occurred", "mitigation_completed") is None