Support global-job and replicated-job modes

Fixes: docker#2829
Signed-off-by: Leonard Kinday <[email protected]>

docker/types/services.py

@@ -29,6 +29,7 @@ class TaskTemplate(dict):
         force_update (int): A counter that triggers an update even if no
             relevant parameters have been changed.
     """
+
     def __init__(self, container_spec, resources=None, restart_policy=None,
                  placement=None, log_driver=None, networks=None,
                  force_update=None):
@@ -115,6 +116,7 @@ class ContainerSpec(dict):
         cap_drop (:py:class:`list`): A list of kernel capabilities to drop from
             the default set for the container.
     """
+
     def __init__(self, image, command=None, args=None, hostname=None, env=None,
                  workdir=None, user=None, labels=None, mounts=None,
                  stop_grace_period=None, secrets=None, tty=None, groups=None,
@@ -231,6 +233,7 @@ class Mount(dict):
         tmpfs_size (int or string): The size for the tmpfs mount in bytes.
         tmpfs_mode (int): The permission mode for the tmpfs mount.
     """
+
     def __init__(self, target, source, type='volume', read_only=False,
                  consistency=None, propagation=None, no_copy=False,
                  labels=None, driver_config=None, tmpfs_size=None,
@@ -331,6 +334,7 @@ class Resources(dict):
           ``{ resource_name: resource_value }``. Alternatively, a list of
           of resource specifications as defined by the Engine API.
     """
+
     def __init__(self, cpu_limit=None, mem_limit=None, cpu_reservation=None,
                  mem_reservation=None, generic_resources=None):
         limits = {}
@@ -401,6 +405,7 @@ class UpdateConfig(dict):
         order (string): Specifies the order of operations when rolling out an
           updated task. Either ``start_first`` or ``stop_first`` are accepted.
     """
+
     def __init__(self, parallelism=0, delay=None, failure_action='continue',
                  monitor=None, max_failure_ratio=None, order=None):
         self['Parallelism'] = parallelism
@@ -511,6 +516,7 @@ class DriverConfig(dict):
         name (string): Name of the driver to use.
         options (dict): Driver-specific options. Default: ``None``.
     """
+
     def __init__(self, name, options=None):
         self['Name'] = name
         if options:
@@ -532,6 +538,7 @@ class EndpointSpec(dict):
           is ``(target_port [, protocol [, publish_mode]])``.
           Ports can only be provided if the ``vip`` resolution mode is used.
     """
+
     def __init__(self, mode=None, ports=None):
         if ports:
             self['Ports'] = convert_service_ports(ports)
@@ -574,37 +581,70 @@ def convert_service_ports(ports):
 
 class ServiceMode(dict):
     """
-        Indicate whether a service should be deployed as a replicated or global
-        service, and associated parameters
+        Indicate whether a service or a job should be deployed as a replicated
+        or global service, and associated parameters
 
         Args:
-            mode (string): Can be either ``replicated`` or ``global``
+            mode (string): Can be either ``replicated``, ``global``,
+              ``replicated-job`` or ``global-job``
             replicas (int): Number of replicas. For replicated services only.
+            concurrency (int): Number of concurrent jobs. For replicated job
+              services only.
     """
-    def __init__(self, mode, replicas=None):
-        if mode not in ('replicated', 'global'):
-            raise errors.InvalidArgument(
-                'mode must be either "replicated" or "global"'
-            )
-        if mode != 'replicated' and replicas is not None:
+
+    def __init__(self, mode, replicas=None, concurrency=None):
+        replicated_modes = ('replicated', 'replicated-job')
+        supported_modes = replicated_modes + ('global', 'global-job')
+
+        if mode not in supported_modes:
             raise errors.InvalidArgument(
-                'replicas can only be used for replicated mode'
+                'mode must be either "replicated", "global", "replicated-job"'
+                ' or "global-job"'
             )
-        self[mode] = {}
+
+        if mode not in replicated_modes:
+            if replicas is not None:
+                raise errors.InvalidArgument(
+                    'replicas can only be used for "replicated" or'
+                    ' "replicated-job" mode'
+                )
+
+            if concurrency is not None:
+                raise errors.InvalidArgument(
+                    'concurrency can only be used for "replicated-job" mode'
+                )
+
+        service_mode = self._convert_mode(mode)
+        self.mode = service_mode
+        self[service_mode] = {}
+
         if replicas is not None:
-            self[mode]['Replicas'] = replicas
+            if mode == 'replicated':
+                self[service_mode]['Replicas'] = replicas
 
-    @property
-    def mode(self):
-        if 'global' in self:
-            return 'global'
-        return 'replicated'
+            if mode == 'replicated-job':
+                self[service_mode]['MaxConcurrent'] = concurrency or 1
+                self[service_mode]['TotalCompletions'] = replicas
+
+    @staticmethod
+    def _convert_mode(original_mode):
+        if original_mode == 'global-job':
+            return 'GlobalJob'
+
+        if original_mode == 'replicated-job':
+            return 'ReplicatedJob'
+
+        return original_mode
 
     @property
     def replicas(self):
-        if self.mode != 'replicated':
-            return None
-        return self['replicated'].get('Replicas')
+        if 'replicated' in self:
+            return self['replicated'].get('Replicas')
+
+        if 'ReplicatedJob' in self:
+            return self['ReplicatedJob'].get('TotalCompletions')
+
+        return None
 
 
 class SecretReference(dict):
@@ -678,6 +718,7 @@ class Placement(dict):
             platforms (:py:class:`list` of tuple): A list of platforms
                 expressed as ``(arch, os)`` tuples
     """
+
     def __init__(self, constraints=None, preferences=None, platforms=None,
                  maxreplicas=None):
         if constraints is not None:
@@ -710,6 +751,7 @@ class PlacementPreference(dict):
                 the scheduler will try to spread tasks evenly over groups of
                 nodes identified by this label.
     """
+
     def __init__(self, strategy, descriptor):
         if strategy != 'spread':
             raise errors.InvalidArgument(
@@ -731,6 +773,7 @@ class DNSConfig(dict):
             options (:py:class:`list`): A list of internal resolver variables
                 to be modified (e.g., ``debug``, ``ndots:3``, etc.).
     """
+
     def __init__(self, nameservers=None, search=None, options=None):
         self['Nameservers'] = nameservers
         self['Search'] = search
@@ -761,6 +804,7 @@ class Privileges(dict):
             selinux_type (string): SELinux type label
             selinux_level (string): SELinux level label
     """
+
     def __init__(self, credentialspec_file=None, credentialspec_registry=None,
                  selinux_disable=None, selinux_user=None, selinux_role=None,
                  selinux_type=None, selinux_level=None):
@@ -803,6 +847,7 @@ class NetworkAttachmentConfig(dict):
             options (:py:class:`dict`): Driver attachment options for the
                 network target.
     """
+
     def __init__(self, target, aliases=None, options=None):
         self['Target'] = target
         self['Aliases'] = aliases