swarming: refactor cipd input

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 30 matching lines @@
                |
     <See task_to_run.py and task_result.py>
 
 TaskProperties is embedded in TaskRequest. TaskProperties is still declared as a
 separate entity to clearly declare the boundary for task request deduplication.
 """
 
 
 import datetime
 import hashlib
-import logging
 import random
 import re
 import urlparse
 
 from google.appengine.api import datastore_errors
 from google.appengine.ext import ndb
 
 from components import auth
 from components import datastore_utils
 from components import pubsub
 from components import utils
+
+from server import config
 from server import task_pack
 import cipd
 
 
 # Maximum acceptable priority value, which is effectively the lowest priority.
 MAXIMUM_PRIORITY = 255
 
 
 # Enforced on both task request and bots.
 DIMENSION_KEY_RE = ur'^[a-zA-Z\-\_\.]+$'
@@ skipping 14 matching lines @@
 # The world started on 2010-01-01 at 00:00:00 UTC. The rationale is that using
 # EPOCH (1970) means that 40 years worth of keys are wasted.
 #
 # Note: This creates a 'naive' object instead of a formal UTC object. Note that
 # datetime.datetime.utcnow() also return naive objects. That's python.
 _BEGINING_OF_THE_WORLD = datetime.datetime(2010, 1, 1, 0, 0, 0, 0)
 
 
 # Used for isolated files.
 _HASH_CHARS = frozenset('0123456789abcdef')
-NAMESPACE_RE = re.compile(r'^[a-z0-9A-Z\-._]+$')
 
 
 ### Properties validators must come before the models.
 
 
 def _validate_isolated(prop, value):
   if value:
     if not _HASH_CHARS.issuperset(value) or len(value) != 40:
       raise datastore_errors.BadValueError(
           '%s must be lowercase hex of length 40, not %s' % (prop._name, value))
 
 
-def _validate_hostname(prop, value):
+def _validate_url(prop, value):
   # pylint: disable=unused-argument
   if value:
     # It must be https://*.appspot.com or https?://*
     parsed = urlparse.urlparse(value)
     if not parsed.netloc:
       raise datastore_errors.BadValueError(
           '%s must be valid hostname, not %s' % (prop._name, value))
     if parsed.netloc.endswith('appspot.com'):
       if parsed.scheme != 'https':
         raise datastore_errors.BadValueError(
             '%s must be https://, not %s' % (prop._name, value))
     elif parsed.scheme not in ('http', 'https'):
         raise datastore_errors.BadValueError(
             '%s must be https:// or http://, not %s' % (prop._name, value))
 
 
 def _validate_namespace(prop, value):
-  if not NAMESPACE_RE.match(value):
+  if not config.NAMESPACE_RE.match(value):
     raise datastore_errors.BadValueError('malformed %s' % prop._name)
 
 
 def _validate_dict_of_strings(prop, value):
   """Validates TaskProperties.env."""
   if not all(
          isinstance(k, unicode) and isinstance(v, unicode)
          for k, v in value.iteritems()):
     # pylint: disable=W0212
     raise TypeError('%s must be a dict of strings' % prop._name)
@@ skipping 58 matching lines @@
 
 
 def _validate_tags(prop, value):
   """Validates and sorts TaskRequest.tags."""
   if not ':' in value:
     # pylint: disable=W0212
     raise datastore_errors.BadValueError(
         '%s must be key:value form, not %s' % (prop._name, value))
 
 
-def _validate_package_name(prop, value):
-  """Validates a CIPD package name."""
-  if not cipd.is_valid_package_name(value):
+def _validate_package_name_template(prop, value):
+  """Validates a CIPD package name template."""
+  if not cipd.is_valid_package_name_template(value):
     raise datastore_errors.BadValueError(
-      '%s must be a valid CIPD package name "%s"' % (prop._name, value))
+        '%s must be a valid CIPD package name template "%s"' % (
+              prop._name, value))
 
 
 def _validate_package_version(prop, value):
   """Validates a CIPD package version."""
   if not cipd.is_valid_version(value):
     raise datastore_errors.BadValueError(
-      '%s must be a valid package version "%s"' % (prop._name, value))
+        '%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."""
 
   # Specifies FilesRef mode.
   # if True, this instance is a reference to a tree, as class docstring says.
   # if False, this instance is isolate input/output settings in TaskProperties.
   # TODO(maruel): refactor this class, get rid of this.
   is_ref = True
 
   # if is_ref, the hash of an isolated archive.
   # otherwise, the hash of the input 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)
+      validator=_validate_url, 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 self.is_ref:
       if not self.isolated or not self.isolatedserver or not self.namespace:
         raise datastore_errors.BadValueError(
             'isolated requires server and namespace')
     else:
       if not self.isolatedserver or not self.namespace:
         raise datastore_errors.BadValueError(
             '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 template. May use cipd.ALL_PARAMS.
+  # Most users will specify ${platform} parameter.
   package_name = ndb.StringProperty(
-      indexed=False, validator=_validate_package_name)
+      indexed=False, validator=_validate_package_name_template)
+  # Package version that is valid for all packages matched by package_name.
+  # Most users will specify tags.
   version = ndb.StringProperty(
       indexed=False, validator=_validate_package_version)
 
   def _pre_put_hook(self):
     super(CipdPackage, self)._pre_put_hook()
     if not self.package_name:
       raise datastore_errors.BadValueError('CIPD package name is required')
     if not self.version:
       raise datastore_errors.BadValueError('CIPD package version is required')
 
 
+class CipdInput(ndb.Model):
+  """Specifies which CIPD client and packages to install, from which server.
+
+  A part of TaskProperties.
+  """
+  # URL of the CIPD server. Must start with "https://" or "http://".
+  server = ndb.StringProperty(indexed=False, validator=_validate_url)
+
+  # CIPD package of CIPD client to use.
+  # client_package.version is required.
+  client_package = ndb.LocalStructuredProperty(CipdPackage)
+
+  # List of packages to install in $CIPD_PATH prior task execution.
+  packages = ndb.LocalStructuredProperty(CipdPackage, repeated=True)
+
+  def _pre_put_hook(self):
+    if not self.server:
+      raise datastore_errors.BadValueError('cipd server is required')
+    if not self.client_package:
+      raise datastore_errors.BadValueError('client_package is required')
+    self.client_package._pre_put_hook()
+
+    if not self.packages:
+      raise datastore_errors.BadValueError(
+          'cipd_input cannot have an empty package list')
+
+    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)
+
+
 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
@@ skipping 21 matching lines @@
 
   # 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&data 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)
+  # CIPD packages to install.
+  cipd_input = ndb.LocalStructuredProperty(CipdInput)
 
   # 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)
@@ skipping 67 matching lines @@
       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.is_ref = False
         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')
-
+      if self.cipd_input:
+        self.cipd_input._pre_put_hook()
+        if self.idempotent:
+          pinned = lambda p: cipd.is_pinned_version(p.version)
+          assert self.cipd_input.packages  # checked by cipd_input._pre_put_hook
+          if any(not pinned(p) for p in self.cipd_input.packages):
+            raise datastore_errors.BadValueError(
+                'an idempotent task cannot have unpinned packages; '
+                'use tags or instance IDs 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.
@@ skipping 302 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))