python: add JobFrobnicator class and flux job-frobnicator cmd

Problem: There is a need for a command and Python class similar
to what exists for job validation -- implemented as a command which
can filter and modify jobspec, driven as a worker by the job-ingest
module.

Add a JobFrobnicator class, very similar to the JobValidator class,
which loads a set of plugins. However, instead of running all plugins
in parallel, jobspec is passed through a stack of plugins which can
modify jobspec in place.

A flux-job-frobnicator.py command is provided which, similar to
flux-job-validator.py, can act as a job-ingest worker. Given
jobspec, it will return the modified jobspec to the caller using the
job-ingest worker protocol by default.

Includes a `defaults` plugin which will set defaults in jobspec
based on defaults set by configuration policy.

Author: grondo

src/bindings/python/flux/Makefile.am

@@ -34,6 +34,9 @@ nobase_fluxpy_PYTHON = \
 	job/validator/plugins/schema.py \
 	job/validator/plugins/feasibility.py \
 	job/validator/plugins/require-instance.py \
+	job/frobnicator/__init__.py \
+	job/frobnicator/frobnicator.py \
+	job/frobnicator/plugins/defaults.py \
 	resource/Rlist.py \
 	resource/__init__.py \
 	resource/ResourceSetImplementation.py \

src/bindings/python/flux/job/frobnicator/__init__.py

@@ -0,0 +1,11 @@
+###############################################################
+# Copyright 2022 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+###############################################################
+
+from flux.job.frobnicator.frobnicator import JobFrobnicator, FrobnicatorPlugin

src/bindings/python/flux/job/frobnicator/frobnicator.py

@@ -0,0 +1,151 @@
+###############################################################
+# Copyright 2022 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+###############################################################
+
+import os
+import argparse
+from abc import abstractmethod
+
+import flux
+from flux.importer import import_plugins, import_path
+from flux.job import Jobspec
+
+
+class FrobnicatorPlugin:
+    """Base class for plugins which modify jobspec in place"""
+
+    def __init__(self, parser):
+        """Initialize a FrobnicatorPlugin"""
+
+    def configure(self, args, config):
+        """Configure a FrobnicatorPlugin. Run after arguments are parsed
+
+        Args:
+           args (:obj:`Namespace`): The resulting namespace after calling
+           argparse.parse_args()
+
+           config (:obj:`dict`): The current broker config, stored as a Python
+           dictionary.
+        """
+
+    @abstractmethod
+    def frob(self, jobspec, userid, urgency, flags):
+        """Modify jobspec. A FrobnicatorPlugin must implement this method.
+
+        The plugin should modify the jobspec parameter directly. Extra
+        job information (user, urgency, flags) are available in the
+        ``info`` parameter.
+
+        Args:
+            jobspec (:obj:`Jobspec`): The jobspec to modify
+
+            userid (:obj:`int`): Submitting user
+
+            urgency (:obj:`int`): Initial job urgency
+
+            flags (:obj:`int`): Job submission flags
+
+        Returns:
+            None or raises exception.
+        """
+        raise NotImplementedError
+
+
+# pylint: disable=too-many-instance-attributes
+class JobFrobnicator:
+    """A plugin-based job modification class
+
+    JobFrobnicator loads an ordered stack of plugins that implement the
+    FrobnicatorPlugin interface from the 'flux.job.frobnicator.plugins'
+    namespace.
+    """
+
+    plugin_namespace = "flux.job.frobnicator.plugins"
+
+    def __init__(self, argv, pluginpath=None, parser=None):
+
+        self.frobnicators = []
+        self.config = {}
+
+        if pluginpath is None:
+            pluginpath = []
+
+        if parser is None:
+            parser = argparse.ArgumentParser(
+                formatter_class=flux.util.help_formatter(), add_help=False
+            )
+
+        self.parser = parser
+        self.parser_group = self.parser.add_argument_group("Options")
+        self.plugins_group = self.parser.add_argument_group(
+            "Options provided by plugins"
+        )
+
+        self.parser_group.add_argument("--plugins", action="append", default=[])
+
+        args, self.remaining_args = self.parser.parse_known_args(argv)
+        if args.plugins:
+            args.plugins = [x for xs in args.plugins for x in xs.split(",")]
+
+        #  Load all available frobnicator plugins
+        self.plugins = import_plugins(self.plugin_namespace, pluginpath)
+        self.args = args
+
+    def start(self):
+        """Read broker config, select and configure frobnicator plugins"""
+
+        self.config = flux.Flux().rpc("config.get").get()
+
+        for name in self.args.plugins:
+            if name not in self.plugins:
+                try:
+                    self.plugins[name] = import_path(name)
+                except:
+                    raise ValueError(f"frobnicator plugin '{name}' not found")
+            plugin = self.plugins[name].Frobnicator(parser=self.plugins_group)
+            self.frobnicators.append(plugin)
+
+        # Parse remaining args and pass result to loaded plugins
+        args = self.parser.parse_args(self.remaining_args)
+        for frobnicator in self.frobnicators:
+            frobnicator.configure(args, config=self.config)
+
+    def frob(self, jobspec, user=None, flags=None, urgency=16):
+        """Modify jobspec using stack of loaded frobnicator plugins
+
+        Args:
+            jobspec (:obj:`Jobspec`): A Jobspec or JobspecV1 object
+                    which will be modified in place
+
+            userid (:obj:`int`): Submitting user
+
+            flags (:obj:`int`): Job submission flags
+
+            urgency (:obj:`int`): Initial job urgency
+
+        Returns:
+            :obj:`dict`: A dictionary containing a result object,
+            including keys::
+
+                {
+                    'errnum': 0,
+                    'errmsg': "An error message",
+                    'data': jobspec or None
+                }
+
+        """
+        if not isinstance(jobspec, Jobspec):
+            raise ValueError("jobspec not an instance of Jobspec")
+
+        if user is None:
+            user = os.getuid()
+
+        for frob in self.frobnicators:
+            frob.frob(jobspec, user, flags, urgency)
+        return {"errnum": 0, "data": jobspec}

src/bindings/python/flux/job/frobnicator/plugins/defaults.py

@@ -0,0 +1,88 @@
+##############################################################
+# Copyright 2021 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+##############################################################
+
+"""Apply defaults to incoming jobspec based on broker config.
+
+"""
+
+from flux.job.frobnicator import FrobnicatorPlugin
+
+
+class DefaultsConfig:
+    """Convenience class for handling jobspec defaults configuration"""
+
+    def __init__(self, config={}):
+        self.defaults = {}
+        self.queues = {}
+        self.default_queue = None
+
+        try:
+            self.defaults = config["policy"]["jobspec"]["defaults"]["system"]
+            self.default_queue = self.defaults["queue"]
+            self.queues = config["queues"]
+        except KeyError:
+            pass
+
+        self.validate_config()
+
+    def validate_config(self):
+        if self.queues and not isinstance(self.queues, dict):
+            raise ValueError("queues must be a table")
+
+        if self.default_queue and self.default_queue not in self.queues:
+            raise ValueError(
+                f"default queue '{self.default_queue}' must be in [queues]"
+            )
+
+        for queue in self.queues:
+            self.queue_defaults(queue)
+
+    def queue_defaults(self, name):
+        if name and self.queues:
+            if name not in self.queues:
+                raise ValueError(f"invalid queue {name} specified")
+            qconf = self.queues[name]
+            try:
+                return qconf["policy"]["jobspec"]["defaults"]["system"]
+            except KeyError:
+                return None
+        return None
+
+    def setattr_default(self, jobspec, attr, value):
+        if attr == "duration" and jobspec.duration == 0:
+            jobspec.duration = value
+        elif attr not in jobspec.attributes["system"]:
+            jobspec.setattr(f"system.{attr}", value)
+
+    def apply_defaults(self, jobspec):
+        """Apply general defaults then queue-specific defaults to jobspec"""
+
+        for attr in self.defaults:
+            self.setattr_default(jobspec, attr, self.defaults[attr])
+
+        if jobspec.queue:
+            if jobspec.queue not in self.queues:
+                raise ValueError(f"Invalid queue '{jobspec.queue}' specified")
+            queue_defaults = self.queue_defaults(jobspec.queue)
+            if queue_defaults:
+                for attr in queue_defaults:
+                    self.setattr_default(jobspec, attr, queue_defaults[attr])
+
+
+class Frobnicator(FrobnicatorPlugin):
+    def __init__(self, parser):
+        self.config = DefaultsConfig()
+        super().__init__(parser)
+
+    def configure(self, args, config):
+        self.config = DefaultsConfig(config)
+
+    def frob(self, jobspec, user, urgency, flags):
+        self.config.apply_defaults(jobspec)

src/cmd/Makefile.am

@@ -77,6 +77,7 @@ dist_fluxcmd_SCRIPTS = \
 	flux-admin.py \
 	flux-jobtap.py \
 	flux-job-validator.py \
+	flux-job-frobnicator.py \
 	flux-job-exec-override.py \
 	flux-perilog-run.py \
 	flux-uri.py \