Add CIPD pin reporting to swarming.

appengine/swarming/server/task_result.py

 # Copyright 2014 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Task execution result models.
 
 This module doesn't do the scheduling itself. It only describes the entities to
 store tasks results.
 
 - TaskResultSummary represents the overall result for the TaskRequest taking in
@@ skipping 40 matching lines @@
        |id=1 (not stored)| |id=1            |
        +-----------------+ +----------------+
                  ^      ^
                  |      |
     +---------------+  +---------------+
     |TaskOutputChunk|  |TaskOutputChunk| ...
     |id=1           |  |id=2           |
     +---------------+  +---------------+
 """
 
+import collections
 import datetime
 import logging
 import random
+import re
 
 from google.appengine.api import datastore_errors
 from google.appengine.datastore import datastore_query
 from google.appengine.ext import ndb
 
 from components import datastore_utils
 from components import utils
 from server import large
 from server import task_pack
 from server import task_request
 
+import cipd
+
 # Amount of time after which a bot is considered dead. In short, if a bot has
 # not ping in the last 5 minutes while running a task, it is considered dead.
 BOT_PING_TOLERANCE = datetime.timedelta(seconds=5*60)
 
 
 class State(object):
   """States in which a task can be.
 
   It's in fact an enum. Values should be in decreasing order of importance.
   """
@@ skipping 214 matching lines @@
   @property
   def is_valid(self):
     return self.bot_overhead is not None
 
   def _pre_put_hook(self):
     if self.bot_overhead is None:
       raise datastore_errors.BadValueError(
           'PerformanceStats.bot_overhead is required')
 
 
+class CipdPins(ndb.Model):
+  """Specifies which CIPD client and packages were actually installed.
+
+  A part of _TaskResultCommon.
+  """
+  # CIPD package of CIPD client to use.
+  # client_package.package_name and version are provided.
+  # client_package.path will be None.
+  client_package = ndb.LocalStructuredProperty(task_request.CipdPackage)
+
+  # List of packages to install in $CIPD_PATH prior task execution.
+  packages = ndb.LocalStructuredProperty(task_request.CipdPackage,
+                                         repeated=True)
+
+
 class _TaskResultCommon(ndb.Model):
   """Contains properties that is common to both TaskRunResult and
   TaskResultSummary.
 
   It is not meant to be instantiated on its own.
 
   TODO(maruel): Overhaul this entity:
   - Get rid of TaskOutput as it is not needed anymore (?)
   """
   # Bot that ran this task.
@@ skipping 47 matching lines @@
   # Children tasks that were triggered by this task. This is set when the task
   # reentrantly creates other Swarming tasks. Note that the task_id is to a
   # TaskResultSummary.
   children_task_ids = ndb.StringProperty(
       validator=_validate_task_summary_id, repeated=True)
 
   # File outputs of the task. Only set if TaskRequest.properties.sources_ref is
   # set. The isolateserver and namespace should match.
   outputs_ref = ndb.LocalStructuredProperty(task_request.FilesRef)
 
+  # The pinned versions of all the CIPD packages used in the task.
+  cipd_pins = ndb.LocalStructuredProperty(CipdPins)
+
   @property
   def can_be_canceled(self):
     """Returns True if the task is in a state that can be canceled."""
     # TOOD(maruel): To be able to add State.RUNNING, the following must be done:
     # task_scheduler.cancel_task() must be strictly a transaction relative to
     # task_scheduler.bot_kill_task() and task_scheduler.bot_update_task().
     #
     # The tricky part is to keep this code performant. On the other hand, all
     # the entities under the transaction (TaskToRun, TaskResultSummary and
     # TaskRunResult) are under the same entity root, so it's definitely
@@ skipping 171 matching lines @@
     present.
     """
     if not self.run_result_key or not self.stdout_chunks:
       # The task was not reaped or no output was streamed for this index yet.
       raise ndb.Return(None)
 
     output_key = _run_result_key_to_output_key(self.run_result_key)
     out = yield TaskOutput.get_output_async(output_key, self.stdout_chunks)
     raise ndb.Return(out)
 
+  def validate(self, request):
+    """Validation that requires the task_request.
+
+    Full validation includes calling this method, and the checks in
+    _pre_put_hook.
+
+    Raises ValueError if this is invalid, otherwise returns None.
+    """
+    props = request.properties
+
+    if props.cipd_input and self.cipd_pins:
+      with cipd.pin_check_fn(None, None) as check:
+        check(props.cipd_input.client_package, self.cipd_pins.client_package)
+        if len(props.cipd_input.packages) != len(self.cipd_pins.packages):
+          raise ValueError('Mismatched package lengths')
+        for a, b in zip(props.cipd_input.packages, self.cipd_pins.packages):
+          check(a, b)
+
   def _pre_put_hook(self):
     """Use extra validation that cannot be validated throught 'validator'."""
     super(_TaskResultCommon, self)._pre_put_hook()
     if self.state == State.EXPIRED:
       if self.failure or self.exit_code is not None:
         raise datastore_errors.BadValueError(
             'Unexpected State, a task can\'t fail if it hasn\'t started yet')
 
     if self.state == State.TIMED_OUT and not self.failure:
       raise datastore_errors.BadValueError('Timeout implies task failure')
@@ skipping 198 matching lines @@
       return None
     return task_pack.result_summary_key_to_run_result_key(
         self.key, self.try_number)
 
   @property
   def task_id(self):
     return task_pack.pack_result_summary_key(self.key)
 
   def reset_to_pending(self):
     """Resets this entity to pending state."""
+    self.cipd_pins = None
     self.duration = None
     self.exit_code = None
     self.internal_failure = False
     self.outputs_ref = None
     self.started_ts = None
     self.state = State.PENDING
 
   def set_from_run_result(self, run_result, request):
     """Copies all the relevant properties from a TaskRunResult into this
     TaskResultSummary.
@@ skipping 366 matching lines @@
   if tags:
     # Add TaskResultSummary indexes if desired.
     if sort != 'created_ts':
       raise ValueError(
           'Add needed indexes for sort:%s and tags if desired' % sort)
     tags_filter = TaskResultSummary.tags == tags[0]
     for tag in tags[1:]:
       tags_filter = ndb.AND(tags_filter, TaskResultSummary.tags == tag)
     query = query.filter(tags_filter)
   return _filter_query(TaskResultSummary, query, start, end, sort, state)