Add CIPD pin reporting to swarming.

appengine/swarming/server/task_scheduler.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.
 
 """High level tasks execution scheduling API.
 
 This is the interface closest to the HTTP handlers.
 """
 
 import datetime
@@ skipping 590 matching lines @@
         user=request.user)
     return request, run_result
   if failures:
     logging.info(
         'Chose nothing (failed %d, skipped %d)', failures, total_skipped)
   return None, None
 
 
 def bot_update_task(
     run_result_key, bot_id, output, output_chunk_start, exit_code, duration,
-    hard_timeout, io_timeout, cost_usd, outputs_ref, performance_stats):
+    hard_timeout, io_timeout, cost_usd, outputs_ref, cipd_pins,
+    performance_stats):
   """Updates a TaskRunResult and TaskResultSummary, along TaskOutput.
 
   Arguments:
   - run_result_key: ndb.Key to TaskRunResult.
   - bot_id: Self advertised bot id to ensure it's the one expected.
   - output: Data to append to this command output.
   - output_chunk_start: Index of output in the stdout stream.
   - exit_code: Mark that this task completed.
   - duration: Time spent in seconds for this task, excluding overheads.
   - hard_timeout: Bool set if an hard timeout occured.
   - io_timeout: Bool set if an I/O timeout occured.
   - cost_usd: Cost in $USD of this task up to now.
   - outputs_ref: task_request.FilesRef instance or None.
+  - cipd_pins: None or task_result.CipdPins
   - performance_stats: task_result.PerformanceStats instance or None. Can only
         be set when the task is completing.
 
   Invalid states, these are flat out refused:
   - A command is updated after it had an exit code assigned to.
 
   Returns:
     TaskRunResult.state or None in case of failure.
   """
   assert output_chunk_start is None or isinstance(output_chunk_start, int)
   assert output is None or isinstance(output, str)
   if cost_usd is not None and cost_usd < 0.:
     raise ValueError('cost_usd must be None or greater or equal than 0')
   if duration is not None and duration < 0.:
     raise ValueError('duration must be None or greater or equal than 0')
   if (duration is None) != (exit_code is None):
     raise ValueError(
         'had unexpected duration; expected iff a command completes\n'
         'duration: %r; exit: %r' % (duration, exit_code))
   if performance_stats and duration is None:
     raise ValueError(
         'duration must be set when performance_stats is set\n'
         'duration: %s; performance_stats: %s' %
         (duration, performance_stats))
 
   packed = task_pack.pack_run_result_key(run_result_key)
   logging.debug(
-      'bot_update_task(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)',
+      'bot_update_task(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)',
       packed, bot_id, len(output) if output else output, output_chunk_start,
       exit_code, duration, hard_timeout, io_timeout, cost_usd, outputs_ref,
-      performance_stats)
+      cipd_pins, performance_stats)
 
   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)
   request_future = request_key.get_async()
   server_version = utils.get_app_version()
   request = request_future.get_result()
   now = utils.utcnow()
 
   def run():
@@ skipping 35 matching lines @@
           result_summary_future.wait()
           return None, None, 'got 2 different durations; %s then %s' % (
               run_result.duration, duration)
       else:
         run_result.duration = duration
         run_result.exit_code = exit_code
 
     if outputs_ref:
       run_result.outputs_ref = outputs_ref
 
+    if cipd_pins:
+      run_result.cipd_pins = cipd_pins
+
     if run_result.state in task_result.State.STATES_RUNNING:
       if hard_timeout or io_timeout:
         run_result.state = task_result.State.TIMED_OUT
         run_result.completed_ts = now
       elif run_result.exit_code is not None:
         run_result.state = task_result.State.COMPLETED
         run_result.completed_ts = now
 
     run_result.signal_server_version(server_version)
+    run_result.validate(request)
     to_put = [run_result]
     if output:
       # This does 1 multi GETs. This also modifies run_result in place.
       to_put.extend(run_result.append_output(output, output_chunk_start or 0))
     if performance_stats:
       performance_stats.key = task_pack.run_result_key_to_performance_stats_key(
           run_result.key)
       to_put.append(performance_stats)
 
     run_result.cost_usd = max(cost_usd, run_result.cost_usd or 0.)
     run_result.modified_ts = now
 
     result_summary = result_summary_future.get_result()
     if (result_summary.try_number and
         result_summary.try_number > run_result.try_number):
       # The situation where a shard is retried but the bot running the previous
       # try somehow reappears and reports success, the result must still show
       # the last try's result. We still need to update cost_usd manually.
       result_summary.costs_usd[run_result.try_number-1] = run_result.cost_usd
       result_summary.modified_ts = now
     else:
       result_summary.set_from_run_result(run_result, request)
 
+    result_summary.validate(request)
     to_put.append(result_summary)
     ndb.put_multi(to_put)
 
     return result_summary, run_result, None
 
   try:
     smry, run_result, error = datastore_utils.transaction(run)
   except datastore_utils.CommitError as e:
     logging.info('Got commit error: %s', e)
     # It is important that the caller correctly surface this error.
@@ skipping 193 matching lines @@
 ## Task queue tasks.
 
 
 def task_handle_pubsub_task(payload):
   """Handles task enqueued by _maybe_pubsub_notify_via_tq."""
   # Do not catch errors to trigger task queue task retry. Errors should not
   # happen in normal case.
   _pubsub_notify(
       payload['task_id'], payload['topic'],
       payload['auth_token'], payload['userdata'])