swarming: add support for cipd on the server side

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 43 matching lines @@
 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 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 122 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):
+    raise datastore_errors.BadValueError(
+      '%s must be a valid CIPD package name "%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))
+
+
 ### Models.
 
 
 class FilesRef(ndb.Model):
   """Defines a data tree reference, normally a reference to a .isolated file."""
   # 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:
       raise datastore_errors.BadValueError(
           'isolated requires server and namespace')
 
 
+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(
+      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 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.
   """
   # 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
   # declared.
   command = ndb.StringProperty(repeated=True, indexed=False)
 
   # File inputs of the task. Only inputs_ref 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)
+
   # 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 62 matching lines @@
         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')
       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.
@@ skipping 212 matching lines @@
             root_entity_shard_id))
 
 
 def make_request(request, is_bot_or_admin):
   """Registers the request in the DB.
 
   Fills up some values.
 
   If parent_task_id is set, properties for the parent are used:
   - priority: defaults to parent.priority - 1
-  - user: overriden by parent.user
+  - user: overridden by parent.user
 
   """
   assert request.__class__ is TaskRequest
   if request.parent_task_id:
     run_result_key = task_pack.unpack_run_result_key(request.parent_task_id)
     result_summary_key = task_pack.run_result_key_to_result_summary_key(
         run_result_key)
     request_key = task_pack.result_summary_key_to_request_key(
         result_summary_key)
     parent = request_key.get()
@@ skipping 69 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))