swarming: change meaning of inputs_ref

appengine/swarming/server/task_request.py

 # coding: utf-8
 # Copyright 2014 The LUCI Authors. All rights reserved.
 # Use of this source code is governed by the Apache v2.0 license that can be
 # found in the LICENSE file.
 
 """Tasks definition.
 
 Each user request creates a new TaskRequest. The TaskRequest instance saves the
 metadata of the request, e.g. who requested it, when why, etc. It links to the
 actual data of the request in a TaskProperties. The TaskProperties represents
@@ skipping 205 matching lines @@
   if not cipd.is_valid_version(value):
     raise datastore_errors.BadValueError(
       '%s must be a valid package version "%s"' % (prop._name, value))
 
 
 ### Models.
 
 
 class FilesRef(ndb.Model):
   """Defines a data tree reference, normally a reference to a .isolated file."""
+
+  # TODO(maruel): make this class have one responsibility. Currently it is used
+  # in two modes:
+  # - a reference to a tree, as class docstring says.
+  # - input/output settings in TaskProperties.
+
   # The hash of an isolated archive.
   isolated = ndb.StringProperty(validator=_validate_isolated, indexed=False)
   # The hostname of the isolated server to use.
   isolatedserver = ndb.StringProperty(
       validator=_validate_hostname, indexed=False)
   # Namespace on the isolate server.
   namespace = ndb.StringProperty(validator=_validate_namespace, indexed=False)
 
   def _pre_put_hook(self):
     super(FilesRef, self)._pre_put_hook()
-    if not self.isolated or not self.isolatedserver or not self.namespace:
+    if not self.isolatedserver or not self.namespace:
       raise datastore_errors.BadValueError(
-          'isolated requires server and namespace')
+          'isolate server and namespace are required')
 
 
 class CipdPackage(ndb.Model):
   """A CIPD package to install in $CIPD_PATH and $PATH before task execution.
 
   A part of TaskProperties.
   """
   package_name = ndb.StringProperty(
       indexed=False, validator=_validate_package_name)
   version = ndb.StringProperty(
@@ skipping 10 matching lines @@
 class TaskProperties(ndb.Model):
   """Defines all the properties of a task to be run on the Swarming
   infrastructure.
 
   This entity is not saved in the DB as a standalone entity, instead it is
   embedded in a TaskRequest.
 
   This model is immutable.
 
   New-style TaskProperties supports invocation of run_isolated. When this
-  behavior is desired, the member .inputs_ref must be suppled. .extra_args can
-  be supplied to pass extraneous arguments.
+  behavior is desired, the member .inputs_ref with an .isolated field value must
+  be supplied. .extra_args can be supplied to pass extraneous arguments.
   """
+
+  # TODO(maruel): convert inputs_ref and _TaskResultCommon.outputs_ref as:
+  # - input = String which is the isolated input, if any
+  # - isolated_server = <server, metadata e.g. namespace> which is a
+  #   simplified version of FilesRef
+  # - _TaskResultCommon.output = String which is isolated output, if any.
+
   # Hashing algorithm used to hash TaskProperties to create its key.
   HASHING_ALGO = hashlib.sha1
 
   # Commands to run. It is a list of 1 item, the command to run.
   # TODO(maruel): Remove after 2016-06-01.
   commands = datastore_utils.DeterministicJsonProperty(
       json_type=list, indexed=False)
-  # Command to run. This is only relevant when self._inputs_ref is None. This is
-  # what is called 'raw commands', in the sense that no inputs files are
+  # Command to run. This is only relevant when self.inputs_ref.isolated is None.
+  # This is what is called 'raw commands', in the sense that no inputs files are
   # declared.
   command = ndb.StringProperty(repeated=True, indexed=False)
 
-  # File inputs of the task. Only inputs_ref or command&data can be specified.
+  # Isolate server, namespace and input isolate hash.
+  #
+  # Despite its name, contains isolate server URL and namespace for isolated
+  # output too. See TODO at the top of this class.
+  # May be non-None even if task input is not isolated.
+  #
+  # Only inputs_ref.isolated or command can be specified.
   inputs_ref = ndb.LocalStructuredProperty(FilesRef)
 
   # A list of CIPD packages to install $CIPD_PATH and $PATH before task
   # execution.
   packages = ndb.LocalStructuredProperty(CipdPackage, repeated=True)
 
   # Filter to use to determine the required properties on the bot to run on. For
   # example, Windows or hostname. Encoded as json. Optional but highly
   # recommended.
   dimensions = datastore_utils.DeterministicJsonProperty(
       validator=_validate_dimensions, json_type=dict, indexed=False)
 
   # Environment variables. Encoded as json. Optional.
   env = datastore_utils.DeterministicJsonProperty(
       validator=_validate_dict_of_strings, json_type=dict, indexed=False)
 
   # Maximum duration the bot can take to run this task. It's named hard_timeout
   # in the bot.
   execution_timeout_secs = ndb.IntegerProperty(
       validator=_validate_timeout, required=True, indexed=False)
 
   # Extra arguments to supply to the command `python run_isolated ...`. Can only
-  # be set if inputs_ref is set.
+  # be set if inputs_ref.isolated is set.
   extra_args = ndb.StringProperty(repeated=True, indexed=False)
 
   # Grace period is the time between signaling the task it timed out and killing
   # the process. During this time the process should clean up itself as quickly
   # as possible, potentially uploading partial results back.
   grace_period_secs = ndb.IntegerProperty(
       validator=_validate_grace, default=30, indexed=False)
 
   # Bot controlled timeout for new bytes from the subprocess. If a subprocess
   # doesn't output new data to stdout for .io_timeout_secs, consider the command
   # timed out. Optional.
   io_timeout_secs = ndb.IntegerProperty(
       validator=_validate_timeout, indexed=False)
 
   # If True, the task can safely be served results from a previously succeeded
   # task.
   idempotent = ndb.BooleanProperty(default=False, indexed=False)
 
   @property
   def is_terminate(self):
     """If True, it is a terminate request."""
     return (
         not self.commands and
         not self.command and
         self.dimensions.keys() == [u'id'] and
-        not self.inputs_ref and
+        not (self.inputs_ref and self.inputs_ref.isolated) and
         not self.env and
         not self.execution_timeout_secs and
         not self.extra_args and
         not self.grace_period_secs and
         not self.io_timeout_secs and
         not self.idempotent)
 
   @property
   def properties_hash(self):
     """Calculates the hash for this entity IFF the task is idempotent.
@@ skipping 12 matching lines @@
     out = super(TaskProperties, self).to_dict(exclude=['commands'])
     out['command'] = self.commands[0] if self.commands else self.command
     return out
 
   def _pre_put_hook(self):
     super(TaskProperties, self)._pre_put_hook()
     if self.commands:
         raise datastore_errors.BadValueError(
             'commands is not supported anymore')
     if not self.is_terminate:
-      if bool(self.command) == bool(self.inputs_ref):
-        raise datastore_errors.BadValueError('use one of command or inputs_ref')
-      if self.extra_args and not self.inputs_ref:
-        raise datastore_errors.BadValueError('extra_args require inputs_ref')
+      isolated_input = self.inputs_ref and self.inputs_ref.isolated
+      if bool(self.command) == bool(isolated_input):
+        raise datastore_errors.BadValueError(
+            'use one of command or inputs_ref.isolated')
+      if self.extra_args and not isolated_input:
+        raise datastore_errors.BadValueError(
+            'extra_args require inputs_ref.isolated')
       if self.inputs_ref:
         self.inputs_ref._pre_put_hook()
 
       package_names = set()
       for p in self.packages:
         p._pre_put_hook()
         if p.package_name in package_names:
           raise datastore_errors.BadValueError(
               'package %s is specified more than once' % p.package_name)
         package_names.add(p.package_name)
       self.packages.sort(key=lambda p: p.package_name)
 
       if self.idempotent:
         pinned = lambda p: cipd.is_pinned_version(p.version)
         if self.packages and any(not pinned(p) for p in self.packages):
           raise datastore_errors.BadValueError(
             'an idempotent task cannot have unpinned packages; '
             'use instance IDs or tags as package versions')
 
 
-
 class TaskRequest(ndb.Model):
   """Contains a user request.
 
   Key id is a decreasing integer based on time since utils.EPOCH plus some
   randomness on lower order bits. See _new_request_key() for the complete gory
   details.
 
   There is also "old style keys" which inherit from a fake root entity
   TaskRequestShard.
   TODO(maruel): Remove support 2015-10-01 once entities are deleted.
@@ skipping 301 matching lines @@
   _put_request(request)
   return request
 
 
 def validate_priority(priority):
   """Throws ValueError if priority is not a valid value."""
   if 0 > priority or MAXIMUM_PRIORITY < priority:
     raise datastore_errors.BadValueError(
         'priority (%d) must be between 0 and %d (inclusive)' %
         (priority, MAXIMUM_PRIORITY))