Feature/hooks track operation

.github/workflows/ci-oldest-reqs.txt

@@ -8,3 +8,4 @@ pytest-cov==3.0.0
 pytest==7.0.1
 ruamel.yaml==0.17.21
 tqdm==4.60.0
+GitPython==3.1.37

changelog.txt

@@ -48,6 +48,7 @@ Added
 - ``test-workflow`` CLI option for testing template environments/submission scripts (#747).
 - Frontier environment and template (#743).
 - Added ``-o`` / ``--operation`` flag to report project status information for specific operations (#725).
+- Added builtin `TrackOperations` execution hooks (#739).
 
 Changed
 +++++++
@@ -89,6 +90,7 @@ Added
 +++++
 
 - Added the OLCF Crusher environment (#708).
+- Added `TrackOperations` as a built in hook (#739).
 
 Changed
 +++++++

doc/api.rst

@@ -160,6 +160,12 @@ Aggregation
 
 .. autofunction:: flow.get_aggregate_id
 
+Hooks
+-----
+
+.. autoclass:: flow.hooks.TrackOperations
+   :members:
+
 Compute Environments
 --------------------
 

flow/hooks/__init__.py

@@ -1,7 +1,8 @@
-# Copyright (c) 2018 The Regents of the University of Michigan
+# Copyright (c) 2023 The Regents of the University of Michigan
 # All rights reserved.
 # This software is licensed under the BSD 3-Clause License.
 """Operation hooks."""
 from .hooks import _Hooks
+from .track_operations import TrackOperations
 
-__all__ = ["_Hooks"]
+__all__ = ["_Hooks", "TrackOperations"]

flow/hooks/git_util.py

@@ -0,0 +1,15 @@
+# Copyright (c) 2023 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+"""Define a function to collect metadata with git."""
+import git
+
+
+def collect_git_metadata(job):
+    """Collect git metadata for a given workspace.
+
+    The information includes the commit ID and a flag indicating if the
+    repository is dirty (has uncommitted changes).
+    """
+    repo = git.Repo(job.project.path)
+    return {"commit_id": str(repo.commit()), "dirty": repo.is_dirty()}

flow/hooks/track_operations.py

@@ -0,0 +1,205 @@
+# Copyright (c) 2023 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+"""Built in execution hook for basic tracking."""
+import json
+
+from .util import collect_metadata
+
+try:
+    from .git_util import collect_git_metadata
+except ImportError:
+    GIT = False
+else:
+    GIT = True
+
+
+_DEFAULT_FILENAME = "signac-execution-history.log"
+
+
+class TrackOperations:
+    """:class:`~.TrackOperations` tracks information about the execution of operations to a logfile.
+
+    This hook can provides information on the start, successful completion, and/or error of
+    one or more operations in a :class:`~.FlowProject` instance. The logs are stored in the file
+    given by ``log_filename`` within the job's path. The file  will be appended to if it already
+    exists.
+
+    The hooks stores metadata regarding the execution of the operation and the state of the
+    project at the time of execution, error, and/or completion. The data will also include
+    information about the git status if the project is detected as a git repo and
+    ``GitPython`` is installed in the environment.
+
+    Each call to the hook adds a single JSON line to the log file. These can be
+    read using the `json` builtin package or :meth:`~.TrackOperations.read_log`.
+
+    The current schema has the following structure:
+
+    - ``time``: The time of querying the metadata.
+    - ``stage``: Whether the hook was executed either "prior" or "after" the associated
+      operation's execution.
+    - ``error``: The error message on executing the operation if any.
+    - ``project``
+       - ``path``: Filepath to the project
+       - ``schema_version``: The project's schema version
+    - ``operation``: The operation name
+    - ``job_id``: The job id
+    - ``git``
+       - ``commit_id``: The current commit of the project's git repo.
+       - ``dirty``: Whether the project's repo has uncommitted changes or not.
+    - ``_schema_version``: The metadata storage's schema version. Schema is currently in version 1.
+
+
+    Warning
+    -------
+    This class will raise an exception when strict_git is set to ``True`` and either GitPython is
+    not available or the repository contains uncommitted changes (i.e. is "dirty").
+
+    Examples
+    --------
+    The following example will install :class:`~.TrackOperations` at the operation level.
+
+    .. code-block:: python
+
+        from flow import FlowProject
+        from flow.hooks import TrackOperations
+
+
+        class Project(FlowProject):
+            pass
+
+
+        track = TrackOperation()
+
+
+        @track.install_operation_hooks(Project)
+        @Project.operation
+        def foo(job):
+            pass
+
+
+    The code block below provides an example of how install :class:`~.TrackOperations` to a
+    instance of :class:`~.FlowProject`
+
+    .. code-block:: python
+
+        from flow import FlowProject
+        from flow.hooks import TrackOperations
+
+
+        class Project(FlowProject):
+            pass
+
+
+        if __name__ == "__main__":
+            project = Project()
+            project = TrackOperations().install_project_hooks(project)
+            project.main()
+
+
+    Parameters
+    ----------
+    log_filename : str, optional
+        The name of the log file in the job workspace. Defaults to "signac-execution-history.log".
+    strict_git : bool, optional
+        Whether to fail if ``GitPython`` cannot be imported or if there are uncommitted changes
+        to the project's git repo. Defaults to ``True``.
+    """
+
+    def __init__(self, log_filename=_DEFAULT_FILENAME, strict_git=True):
+        self.log_filename = log_filename
+        if strict_git and not GIT:
+            raise RuntimeError(
+                "Unable to collect git metadata from the repository, "
+                "because the GitPython package is not installed.\n\n"
+                "You can use '{}(strict_git=False)' to ignore this "
+                "error.".format(type(self).__name__)
+            )
+        self.strict_git = strict_git
+
+    def _write_metadata(self, job, metadata):
+        with open(job.fn(self.log_filename), "a") as logfile:
+            logfile.write(json.dumps(metadata) + "\n")
+
+    def _get_metadata(self, operation, job, stage, error=None):
+        """Define log_operation to collect metadata of job workspace and write to logfiles."""
+        # Add execution-related information to metadata.
+        metadata = {"stage": stage, "error": None if error is None else str(error)}
+        metadata.update(collect_metadata(operation, job))
+        if GIT:
+            git_metadata = collect_git_metadata(job)
+            if self.strict_git and git_metadata["dirty"]:
+                raise RuntimeError(
+                    "Unable to reliably log operation, because the git repository in "
+                    "the project root directory is dirty.\n\nMake sure to commit all "
+                    "changes or ignore this warning by setting '{}(strict_git=False)'.".format(
+                        type(self).__name__
+                    )
+                )
+            metadata["project"]["git"] = git_metadata
+        return metadata
+
+    def on_start(self, operation, job):
+        """Track the start of execution of an operation on a job."""
+        self._write_metadata(job, self._get_metadata(operation, job, stage="prior"))
+
+    def on_success(self, operation, job):
+        """Track the successful completion of an operation on a job."""
+        self._write_metadata(job, self._get_metadata(operation, job, stage="after"))
+
+    def on_exception(self, operation, error, job):
+        """Log errors raised during the execution of an operation on a job."""
+        self._write_metadata(
+            job, self._get_metadata(operation, job, stage="after", error=error)
+        )
+
+    def install_operation_hooks(self, op, project_cls=None):
+        """Decorate operation to track execution.
+
+        Parameters
+        ----------
+        op : function or type
+            An operation function to log or a subclass of :class:`~.FlowProject` if
+            ``project_cls`` is ``None``.
+        project_cls : type
+            A subclass of :class:`~.FlowProject`.
+        """
+        if project_cls is None:
+            return lambda func: self.install_operation_hooks(func, op)
+        project_cls.operation_hooks.on_start(self.on_start)(op)
+        project_cls.operation_hooks.on_success(self.on_success)(op)
+        project_cls.operation_hooks.on_exception(self.on_exception)(op)
+        return op
+
+    def install_project_hooks(self, project):
+        """Install hooks to track all operations in a `flow.FlowProject`.
+
+        Parameters
+        ----------
+        project : flow.FlowProject
+            The project to install hooks on.
+        """
+        project.project_hooks.on_start.append(self.on_start)
+        project.project_hooks.on_success.append(self.on_success)
+        project.project_hooks.on_exception.append(self.on_exception)
+        return project
+
+    @classmethod
+    def read_log(cls, job, log_filename=_DEFAULT_FILENAME):
+        """Return the execution log data as a list of dictionaries.
+
+        Parameters
+        ----------
+        job : signac.job.Job
+            The job to read the execution history of.
+        log_filename : str, optional
+            The name of the log file in the job workspace. Defaults to
+            "signac-execution-history.log".
+
+        Returns
+        -------
+        log : list[dict[str, any]]
+            Returns the job's current execution history for logged operations.
+        """
+        with open(job.fn(log_filename)) as fh:
+            return [json.loads(line) for line in fh.readlines() if line != ""]

flow/hooks/util.py

@@ -0,0 +1,25 @@
+# Copyright (c) 2023 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+"""Define a function to collect metadata on the operation and job."""
+from datetime import datetime, timezone
+
+
+def collect_metadata(operation, job):
+    """Collect metadata related to the operation and job.
+
+    Returns a directory including schema version, time, project, and job-operation.
+
+    """
+    return {
+        # the metadata schema version:
+        "_schema_version": "1",
+        "time": datetime.now(timezone.utc).isoformat(),
+        "project": {
+            "path": job.project.path,
+            # the project schema version:
+            "schema_version": job.project.config.get("schema_version"),
+        },
+        "operation": operation,
+        "job_id": job.id,
+    }

requirements/requirements-test.txt

@@ -1,5 +1,6 @@
 click==8.1.7
 coverage==7.3.2
+GitPython==3.1.37
 pytest-cov==4.1.0
 pytest==7.4.3
 ruamel.yaml==0.18.3

tests/define_hooks_track_operations_project.py

@@ -0,0 +1,34 @@
+from define_hooks_test_project import HOOKS_ERROR_MESSAGE
+
+from flow import FlowProject
+from flow.hooks import TrackOperations
+
+
+class _HooksTrackOperations(FlowProject):
+    pass
+
+
+LOG_FILENAME = "signac-execution-history.log"
+
+
+track_operations = TrackOperations(strict_git=False)
+
+
+@track_operations.install_operation_hooks(_HooksTrackOperations)
+@_HooksTrackOperations.operation
+def base(job):
+    if job.sp.raise_exception:
+        raise RuntimeError(HOOKS_ERROR_MESSAGE)
+
+
+@track_operations.install_operation_hooks(_HooksTrackOperations)
+@_HooksTrackOperations.operation(cmd=True, with_job=True)
+def cmd(job):
+    if job.sp.raise_exception:
+        return "exit 42"
+    else:
+        return "touch base_cmd.txt"
+
+
+if __name__ == "__main__":
+    _HooksTrackOperations().main()

tests/define_hooks_track_operations_strict_project.py

@@ -0,0 +1,34 @@
+from define_hooks_test_project import HOOKS_ERROR_MESSAGE
+
+from flow import FlowProject
+from flow.hooks import TrackOperations
+
+
+class _HooksTrackOperations(FlowProject):
+    pass
+
+
+LOG_FILENAME = "operations.log"
+
+
+track_operations = TrackOperations()
+
+
+@track_operations.install_operation_hooks(_HooksTrackOperations)
+@_HooksTrackOperations.operation
+def base(job):
+    if job.sp.raise_exception:
+        raise RuntimeError(HOOKS_ERROR_MESSAGE)
+
+
+@track_operations.install_operation_hooks(_HooksTrackOperations)
+@_HooksTrackOperations.operation(cmd=True, with_job=True)
+def cmd(job):
+    if job.sp.raise_exception:
+        return "exit 42"
+    else:
+        return "touch base_cmd.txt"
+
+
+if __name__ == "__main__":
+    _HooksTrackOperations().main()