Merge pull request #5409 from grondo/job-update-service

add job update service and new job-update(1) command

doc/Makefile.am

@@ -53,7 +53,8 @@ MAN1_FILES_PRIMARY = \
 	man1/flux-resource.1 \
 	man1/flux-pgrep.1 \
 	man1/flux-cancel.1 \
-	man1/flux-watch.1
+	man1/flux-watch.1 \
+	man1/flux-update.1
 
 # These files are generated as clones of a primary page.
 # Sphinx handles this automatically if declared in the conf.py

doc/man1/flux-update.rst

@@ -0,0 +1,102 @@
+.. flux-help-section: jobs
+
+==============
+flux-update(1)
+==============
+
+SYNOPSIS
+========
+
+**flux** **update** [*OPTIONS*] JOBID KEY=VALUE [KEY=VALUE...]
+
+DESCRIPTION
+===========
+
+flux-update(1) requests an update of one or more attributes for an active
+(pending or running) job. Updates are permitted and validated by the job
+manager before being applied to the job.
+
+Keys are expressed as period-delimited strings corresponding to an attribute
+path in jobspec. If a key does not start with ``attributes.``, ``tasks.``,
+or ``.resources`` (the top-level jobspec keys allowed by RFC 14), then
+the key is assumed to be prefixed with ``attributes.system.``, such that::
+
+  $ flux update f12345 myattr="value"
+
+would request an update of ``attributes.system.myattr`` to the string value
+``"value"``.
+
+The flux-update(1) command may also support other convenient key aliases.
+Key aliases are listed in the SPECIAL KEYS section below.
+
+Updates will be sent to the job manager update service, which checks that
+the current user is permitted to apply all updates, and that all updates
+are valid. If multiple updates are specified on the command line, then
+all updates are either applied or the request fails.
+
+.. note::
+   Job updates are allowed in the job manager by special plugins on
+   a case by case basis. At this time, the set of keys that can actually
+   be updated for a job may be very limited.
+
+The instance owner may be allowed to update specific attributes of jobs
+and bypass validation checks. For example, the duration of a guest job may
+be increased beyond currently configured limits if the update request is
+performed by the instance owner. When a job is modified in this way, future
+updates to the job by the guest user are denied with an error message::
+
+   job is immutable due to previous instance owner update
+
+This is necessary to prevent possible unintended bypass of limits or
+other checks on a job by a guest.
+
+The flux-update(1) command may also support special handling of values
+for specific keys. Those special cases are documented in the SPECIAL KEYS
+section below.
+
+OPTIONS
+=======
+
+**-n, --dry-run**
+  Do not send update to job manager, but print the updates in JSON to
+  stdout.
+
+**-v, --verbose**
+  Print updated keys on success.
+
+SPECIAL KEYS
+============
+
+*attributes.system.duration*, *duration*
+  Updates of the job ``duration`` can take the form of of *[+-]FSD*, where
+  ``+`` or ``-`` indicate an adjustment of the existing duration, and *FSD*
+  is any string or number in RFC 23 Flux Standard Duration. Examples include
+  ``60``, ``1m``, ``1.5h``, ``+10m``, ``-1h``.
+
+*name*
+  Alias for job name, i.e. ``attributes.system.job.name``
+
+EXIT STATUS
+===========
+
+0
+  All updates were successful
+
+1
+  Updates were invalid or not permitted, or the provided JOBID was invalid
+  or inactive, or the user does not have permission to modify the job
+
+2
+  Syntax or other command line error
+
+RESOURCES
+=========
+
+Flux: http://flux-framework.org
+
+RFC 14: Canonical Job Specification: https://flux-framework.readthedocs.io/projects/flux-rfc/en/latest/spec_14.html
+
+SEE ALSO
+========
+
+:man1:`flux-jobs`, :man1:`flux-submit`, :man1:`flux-bulksubmit`

doc/manpages.py

@@ -68,6 +68,7 @@
     ('man1/flux', 'flux', 'the Flux resource management framework', [author], 1),
     ('man1/flux-shell', 'flux-shell', 'the Flux job shell', [author], 1),
     ('man1/flux-watch', 'flux-watch', 'monitor one or more Flux jobs', [author], 1),
+    ('man1/flux-update', 'flux-update', 'update active Flux jobs', [author], 1),
     ('man3/flux_attr_get', 'flux_attr_set', 'get/set Flux broker attributes', [author], 3),
     ('man3/flux_attr_get', 'flux_attr_get', 'get/set Flux broker attributes', [author], 3),
     ('man3/flux_aux_set', 'flux_aux_get', 'get/set auxiliary handle data', [author], 3),

doc/test/spell.en.pws

@@ -717,3 +717,4 @@ MemoryHigh
 MemoryMax
 MemoryLow
 MemoryMin
+myattr

etc/completions/flux.pre

@@ -159,6 +159,29 @@ _flux_cancel()
     return 0
 }
 
+#  flux-update(1) completions
+_flux_update()
+{
+    local cmd=$1
+    OPTS="\
+        -v, --verbose \
+        -n, --dry-run \
+    "
+    if [[ $cur != -* ]]; then
+        #  Attempt to substitute a pending jobid
+        compopt +o filenames
+        pending_jobs=$(flux jobs -f pending -no {id})
+        COMPREPLY=( $(compgen -W "${pending_jobs}" -- "$cur") )
+        return 0
+    fi
+    COMPREPLY=( $(compgen -W "${OPTS} -h --help" -- "$cur") )
+    if [[ "${COMPREPLY[@]}" == *= ]]; then
+        # Add space if there is not a '=' in suggestions
+        compopt -o nospace
+    fi
+    return 0
+}
+
 #  flux-mini(1) completions
 _flux_mini()
 {
@@ -1474,6 +1497,9 @@ _flux_core()
     watch)
         _flux_watch $subcmd
         ;;
+    update)
+        _flux_update $subcmd
+        ;;
     -*)
         COMPREPLY=( $(compgen -W "${FLUX_OPTS}" -- "$cur") )
         ;;

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

@@ -533,12 +533,29 @@ def add_file(self, path, data, perms=0o0600, encoding=None):
         files[path] = Fileref(data, perms=perms, encoding=encoding)
         self.jobspec["attributes"]["system"]["files"] = files
 
-    def setattr(self, key, val):
+    def getattr(self, key):
+        """
+        get attribute from jobspec using dotted key notation, e.g.
+        system.duration or optionally attributes.system.duration.
 
+        Raises KeyError if a component of key does not exist.
+        """
+        if not key.startswith("attributes."):
+            key = "attributes." + key
+        value = self.jobspec
+        for attr in key.split("."):
+            value = value.get(attr)
+            if value is None:
+                raise KeyError
+        return value
+
+    def setattr(self, key, val):
         """
         set job attribute
         """
-        set_treedict(self.jobspec, "attributes." + key, val)
+        if not key.startswith("attributes."):
+            key = "attributes." + key
+        set_treedict(self.jobspec, key, val)
 
     def setattr_shell_option(self, key, val):
         """

src/cmd/Makefile.am

@@ -115,6 +115,7 @@ dist_fluxcmd_SCRIPTS = \
 	flux-queue.py \
 	flux-cancel.py \
 	flux-watch.py \
+	flux-update.py \
 	flux-imp-exec-helper
 
 fluxcmd_PROGRAMS = \