pending_manager.py

   1 # coding=utf8
   2 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5 """Commit queue manager class.
   6
   7 Security implications:
   8
   9 The following hypothesis are made:
  10 - Commit queue:
  11   - Impersonate the same svn credentials that the patchset owner.
  12   - Can't impersonate a non committer.
  13   - SVN will check the committer write access.
  14 """
  15
  16 import logging
  17 import os
  18 import time
  19 import traceback
  20 import urllib2
  21
  22 import find_depot_tools  # pylint: disable=W0611
  23 import checkout
  24 import git_cl
  25 import patch
  26 import subprocess2
  27
  28 import errors
  29 import model
  30 from verification import base
  31
  32
  33 class PendingCommit(base.Verified):
  34   """Represents a pending commit that is being processed."""
  35   # Important since they tell if we need to revalidate and send try jobs
  36   # again or not if any of these value changes.
  37   issue = int
  38   patchset = int
  39   description = unicode
  40   files = list
  41   # Only a cache, these values can be regenerated.
  42   owner = unicode
  43   reviewers = list
  44   base_url = unicode
  45   messages = list
  46   relpath = unicode
  47   # Only used after a patch was committed. Keeping here for try job retries.
  48   revision = (None, int, unicode)
  49
  50   def __init__(self, **kwargs):
  51     super(PendingCommit, self).__init__(**kwargs)
  52     for message in self.messages:
  53       # Save storage, no verifier really need 'text', just 'approval'.
  54       if 'text' in message:
  55         del message['text']
  56
  57   def pending_name(self):
  58     """The name that should be used for try jobs.
  59
  60     It makes it possible to regenerate the try_jobs array if ever needed."""
  61     return '%d-%d' % (self.issue, self.patchset)
  62
  63   def prepare_for_patch(self, context_obj):
  64     self.revision = context_obj.checkout.prepare(self.revision)
  65     # Verify revision consistency.
  66     if not self.revision:
  67       raise base.DiscardPending(
  68           self, 'Internal error: failed to checkout. Please try again.')
  69
  70   def apply_patch(self, context_obj, prepare):
  71     """Applies the pending patch to the checkout and throws if it fails."""
  72     try:
  73       if prepare:
  74         self.prepare_for_patch(context_obj)
  75       patches = context_obj.rietveld.get_patch(self.issue, self.patchset)
  76       if not patches:
  77         raise base.DiscardPending(
  78             self, 'No diff was found for this patchset.')
  79       if self.relpath:
  80         patches.set_relpath(self.relpath)
  81       self.files = [p.filename for p in patches]
  82       if not self.files:
  83         raise base.DiscardPending(
  84             self, 'No file was found in this patchset.')
  85       context_obj.checkout.apply_patch(patches)
  86     except (checkout.PatchApplicationFailed, patch.UnsupportedPatchFormat) as e:
  87       raise base.DiscardPending(self, str(e))
  88     except subprocess2.CalledProcessError as e:
  89       out = 'Failed to apply the patch.'
  90       if e.stdout:
  91         out += '\n%s' % e.stdout
  92       raise base.DiscardPending(self, out)
  93     except urllib2.HTTPError as e:
  94       raise base.DiscardPending(
  95           self,
  96           ('Failed to request the patch to try. Please note that binary files '
  97           'are still unsupported at the moment, this is being worked on.\n\n'
  98           'Thanks for your patience.\n\n%s') % e)
  99
 100
 101 class PendingQueue(model.PersistentMixIn):
 102   """Represents the queue of pending commits being processed.
 103
 104   Each entry is keyed by the issue number as a string to be json-compatible.
 105   There can only be one pending commit per issue and they are fine to be
 106   processed out of order.
 107   """
 108   pending_commits = dict
 109
 110   def add(self, item):
 111     self.pending_commits[str(item.issue)] = item
 112
 113   def get(self, key):
 114     return self.pending_commits[str(key)]
 115
 116   def iterate(self):
 117     """Returns the items sorted by issue id to ease testability."""
 118     return sorted(self.pending_commits.itervalues(), key=lambda x: x.issue)
 119
 120   def remove(self, key):
 121     self.pending_commits.pop(str(key), None)
 122
 123
 124 class PendingManager(object):
 125   """Fetch new issues from rietveld, pass the issues through all of verifiers
 126   and then commit the patches with checkout.
 127   """
 128   FAILED_NO_MESSAGE = (
 129       'Commit queue patch verification failed without an error message.\n'
 130       'Something went wrong, probably a crash, a hickup or simply\n'
 131       'the monkeys went out for dinner.\n'
 132       'Please email commit-bot@chromium.org with the CL url.')
 133   INTERNAL_EXCEPTION = (
 134       'Commit queue had an internal error.\n'
 135       'Something went really wrong, probably a crash, a hickup or\n'
 136       'simply the monkeys went out for dinner.\n'
 137       'Please email commit-bot@chromium.org with the CL url.')
 138   DESCRIPTION_UPDATED = (
 139       'Commit queue rejected this change because the description was changed\n'
 140       'between the time the change entered the commit queue and the time it\n'
 141       'was ready to commit. You can safely check the commit box again.')
 142   TRYING_PATCH = 'CQ is trying da patch. Follow status at\n'
 143   # Maximum number of commits done in a burst.
 144   MAX_COMMIT_BURST = 4
 145   # Delay (secs) between commit bursts.
 146   COMMIT_BURST_DELAY = 10*60
 147
 148   def __init__(self, context_obj, pre_patch_verifiers, verifiers):
 149     """
 150     Args:
 151       pre_patch_verifiers: Verifiers objects that are run before applying the
 152                            patch.
 153       verifiers: Verifiers object run after applying the patch.
 154     """
 155     assert len(pre_patch_verifiers) or len(verifiers)
 156     self.context = context_obj
 157     self.pre_patch_verifiers = pre_patch_verifiers or []
 158     self.verifiers = verifiers or []
 159     self.all_verifiers = pre_patch_verifiers + verifiers
 160     self.queue = PendingQueue()
 161     # Keep the timestamps of the last few commits so that we can control the
 162     # pace (burstiness) of commits.
 163     self.recent_commit_timestamps = []
 164     # Assert names are unique.
 165     names = [x.name for x in pre_patch_verifiers + verifiers]
 166     assert len(names) == len(set(names))
 167     for verifier in self.pre_patch_verifiers:
 168       assert not isinstance(verifier, base.VerifierCheckout)
 169
 170   def look_for_new_pending_commit(self):
 171     """Looks for new reviews on self.context.rietveld with c+ set.
 172
 173     Calls _new_pending_commit() on all new review found.
 174     """
 175     try:
 176       new_issues = self._fetch_pending_issues()
 177
 178       # If there is an issue in processed_issues that is not in new_issues,
 179       # discard it.
 180       for pending in self.queue.iterate():
 181         # Note that pending.issue is a int but self.queue.pending_commits keys
 182         # are str due to json support.
 183         if pending.issue not in new_issues:
 184           logging.info('Flushing issue %d' % pending.issue)
 185           self.context.status.send(
 186               pending,
 187               { 'verification': 'abort',
 188                 'payload': {
 189                   'output': 'CQ bit was unchecked on CL. Ignoring.' }})
 190           pending.get_state = lambda: base.IGNORED
 191           self._discard_pending(pending, None)
 192
 193       # Find new issues.
 194       for issue_id in new_issues:
 195         if str(issue_id) not in self.queue.pending_commits:
 196           issue_data = self.context.rietveld.get_issue_properties(
 197               issue_id, True)
 198           # This assumption needs to hold.
 199           assert issue_id == issue_data['issue']
 200           if issue_data['patchsets'] and issue_data['commit']:
 201             logging.info('Found new issue %d' % issue_id)
 202             self.queue.add(
 203                 PendingCommit(
 204                     issue=issue_id,
 205                     owner=issue_data['owner_email'],
 206                     reviewers=issue_data['reviewers'],
 207                     patchset=issue_data['patchsets'][-1],
 208                     base_url=issue_data['base_url'],
 209                     description=issue_data['description'].replace('\r', ''),
 210                     messages=issue_data['messages']))
 211     except Exception as e:
 212       traceback.print_exc()
 213       # Swallow every exception in that code and move on. Make sure to send a
 214       # stack trace though.
 215       errors.send_stack(e)
 216
 217   def _fetch_pending_issues(self):
 218     """Returns the list of issue number for reviews on Rietveld with their last
 219     patchset with commit+ flag set.
 220     """
 221     return self.context.rietveld.get_pending_issues()
 222
 223   def process_new_pending_commit(self):
 224     """Starts verification on newly found pending commits."""
 225     expected = set(i.name for i in self.all_verifiers)
 226     for pending in self.queue.iterate():
 227       try:
 228         # Take in account the case where a verifier was removed.
 229         done = set(pending.verifications.keys())
 230         missing = expected - done
 231         if (not missing or pending.get_state() != base.PROCESSING):
 232           continue
 233         logging.info(
 234             'Processing issue %s (%s, %d)' % (
 235                 pending.issue, missing, pending.get_state()))
 236         self._verify_pending(pending)
 237       except base.DiscardPending as e:
 238         self._discard_pending(e.pending, e.status)
 239       except Exception as e:
 240         traceback.print_exc()
 241         # Swallow every exception in that code and move on. Make sure to send a
 242         # stack trace though.
 243         errors.send_stack(e)
 244
 245   def update_status(self):
 246     """Updates the status for each pending commit verifier."""
 247     why_nots = dict((p.issue, p.why_not()) for p in self.queue.iterate())
 248
 249     for verifier in self.all_verifiers:
 250       try:
 251         verifier.update_status(self.queue.iterate())
 252       except base.DiscardPending as e:
 253         # It's not efficient since it takes a full loop for each pending
 254         # commit to discard.
 255         self._discard_pending(e.pending, e.status)
 256       except Exception as e:
 257         traceback.print_exc()
 258         # Swallow every exception in that code and move on. Make sure to send
 259         # a stack trace though.
 260         errors.send_stack(e)
 261
 262     for pending in self.queue.iterate():
 263       why_not = pending.why_not()
 264       if why_nots[pending.issue] != why_not:
 265         self.context.status.send(
 266             pending,
 267             {'verification': 'why not',
 268              'payload': {'message': why_not}})
 269
 270
 271   def scan_results(self):
 272     """Scans pending commits that can be committed or discarded."""
 273     for pending in self.queue.iterate():
 274       state = pending.get_state()
 275       if state == base.FAILED:
 276         self._discard_pending(
 277             pending, pending.error_message() or self.FAILED_NO_MESSAGE)
 278       elif state == base.SUCCEEDED:
 279         if self._throttle(pending):
 280           continue
 281         try:
 282           # Runs checks. It's be nice to run the test before the postpone,
 283           # especially if the tree is closed for a long moment but at the same
 284           # time it would keep fetching the rietveld status constantly.
 285           self._last_minute_checks(pending)
 286           self.context.status.send(
 287               pending,
 288               {'verification': 'why not',
 289                'payload': {'message': ''}})
 290
 291           self._commit_patch(pending)
 292         except base.DiscardPending as e:
 293           self._discard_pending(e.pending, e.status)
 294         except Exception as e:
 295           traceback.print_exc()
 296           errors.send_stack(e)
 297           self._discard_pending(pending, self.INTERNAL_EXCEPTION)
 298       else:
 299         # When state is IGNORED, we need to keep this issue so it's not fetched
 300         # another time but we can't discard it since we don't want to remove the
 301         # commit bit for another project hosted on the same code review
 302         # instance.
 303         assert state in (base.PROCESSING, base.IGNORED)
 304
 305   def _verify_pending(self, pending):
 306     """Initiates all the verifiers on a pending change."""
 307     # Do not apply the patch if not necessary. It will be applied at commit
 308     # time anyway so if the patch doesn't apply, it'll be catch later.
 309     if not self._pending_run_verifiers(pending, self.pre_patch_verifiers):
 310       return
 311
 312     if self.verifiers:
 313       pending.prepare_for_patch(self.context)
 314
 315     # This CL is real business, alert the user that we're going to try his
 316     # patch.  Note that this is done *after* syncing but *before* applying the
 317     # patch.
 318     self.context.status.send(
 319         pending,
 320         { 'verification': 'initial',
 321           'payload': {'revision': pending.revision}})
 322     self.context.rietveld.add_comment(
 323         pending.issue,
 324         self.TRYING_PATCH + '%s/%s/%d/%d\n' % (
 325           self.context.status.url, pending.owner,
 326           pending.issue, pending.patchset))
 327
 328     if self.verifiers:
 329       pending.apply_patch(self.context, False)
 330       previous_cwd = os.getcwd()
 331       try:
 332         os.chdir(self.context.checkout.project_path)
 333         self._pending_run_verifiers(pending, self.verifiers)
 334       finally:
 335         os.chdir(previous_cwd)
 336
 337     # Send the initial 'why not' message.
 338     if pending.why_not():
 339       self.context.status.send(
 340           pending,
 341           {'verification': 'why not',
 342            'payload': {'message': pending.why_not()}})
 343
 344   @classmethod
 345   def _pending_run_verifiers(cls, pending, verifiers):
 346     """Runs verifiers on a pending change.
 347
 348     Returns True if all Verifiers were run.
 349     """
 350     for verifier in verifiers:
 351       if verifier.name in pending.verifications:
 352         logging.warning(
 353             'Re-running verififer %s for issue %s' % (
 354                 verifier.name, pending.issue))
 355       verifier.verify(pending)
 356       assert verifier.name in pending.verifications
 357       if pending.get_state() == base.IGNORED:
 358         assert pending.verifications[verifier.name].get_state() == base.IGNORED
 359         # Remove all the other verifiers since we need to keep it in the
 360         # 'datastore' to not retry this issue constantly.
 361         for key in pending.verifications.keys():
 362           if key != verifier.name:
 363             del pending.verifications[key]
 364         return False
 365       if pending.get_state() == base.FAILED:
 366         # Throw if it didn't pass, so the error message is not lost.
 367         raise base.DiscardPending(
 368             pending, pending.error_message() or cls.FAILED_NO_MESSAGE)
 369     return True
 370
 371   def _last_minute_checks(self, pending):
 372     """Does last minute checks on Rietvld before committing a pending patch."""
 373     pending_data = self.context.rietveld.get_issue_properties(
 374         pending.issue, True)
 375     if pending_data['commit'] != True:
 376       raise base.DiscardPending(pending, None)
 377     if pending_data['closed'] != False:
 378       raise base.DiscardPending(pending, None)
 379     if pending.description != pending_data['description'].replace('\r', ''):
 380       raise base.DiscardPending(pending, self.DESCRIPTION_UPDATED)
 381     commit_user = set([self.context.rietveld.email])
 382     expected = set(pending.reviewers) - commit_user
 383     actual  = set(pending_data['reviewers']) - commit_user
 384     # Try to be nice, if there was a drive-by review and the new reviewer left
 385     # a lgtm, don't abort.
 386     def is_approver(r):
 387       return any(
 388           m.get('approval') for m in pending_data['messages']
 389           if m['sender'] == r)
 390     drivers_by = [r for r in (actual - expected) if not is_approver(r)]
 391     if drivers_by:
 392       # That annoying driver-by.
 393       raise base.DiscardPending(
 394           pending,
 395           'List of reviewers changed. %s did a drive-by without LGTM\'ing!' %
 396           ','.join(drivers_by))
 397     if pending.patchset != pending_data['patchsets'][-1]:
 398       raise base.DiscardPending(pending,
 399           'Commit queue failed due to new patchset.')
 400
 401   def _discard_pending(self, pending, message):
 402     """Discards a pending commit. Attach an optional message to the review."""
 403     logging.debug('_discard_pending(%s, %s)', pending.issue, message)
 404     try:
 405       try:
 406         if pending.get_state() != base.IGNORED:
 407           self.context.rietveld.set_flag(
 408               pending.issue, pending.patchset, 'commit', 'False')
 409       except urllib2.HTTPError as e:
 410         logging.error(
 411             'Failed to set the flag to False for %s with message %s' % (
 412               pending.pending_name(), message))
 413         traceback.print_stack()
 414         errors.send_stack(e)
 415       if message:
 416         try:
 417           self.context.rietveld.add_comment(pending.issue, message)
 418         except urllib2.HTTPError as e:
 419           logging.error(
 420               'Failed to add comment for %s with message %s' % (
 421                 pending.pending_name(), message))
 422           traceback.print_stack()
 423           errors.send_stack(e)
 424         self.context.status.send(
 425             pending,
 426             { 'verification': 'abort',
 427               'payload': {
 428                 'output': message }})
 429     finally:
 430       # Most importantly, remove the PendingCommit from the queue.
 431       self.queue.remove(pending.issue)
 432
 433   def _commit_patch(self, pending):
 434     """Commits the pending patch to the repository.
 435
 436     Do the checkout and applies the patch.
 437     """
 438     try:
 439       try:
 440         # Make sure to apply on HEAD.
 441         pending.revision = None
 442         pending.apply_patch(self.context, True)
 443         # Commit it.
 444         commit_desc = git_cl.ChangeDescription(pending.description)
 445         if (self.context.server_hooks_missing and
 446             self.context.rietveld.email != pending.owner):
 447           commit_desc.update_reviewers(pending.reviewers)
 448           commit_desc.append_footer('Author: ' + pending.owner)
 449         commit_desc.append_footer('Review URL: %s/%s' % (
 450             self.context.rietveld.url,
 451             pending.issue))
 452         pending.revision = self.context.checkout.commit(
 453             commit_desc.description, pending.owner)
 454         if not pending.revision:
 455           raise base.DiscardPending(pending, 'Failed to commit patch.')
 456
 457         # Note that the commit succeeded for commit throttling.
 458         self.recent_commit_timestamps.append(time.time())
 459         self.recent_commit_timestamps = (
 460             self.recent_commit_timestamps[-(self.MAX_COMMIT_BURST + 1):])
 461
 462         viewvc_url = self.context.checkout.get_settings('VIEW_VC')
 463         issue_desc = git_cl.ChangeDescription(pending.description)
 464         msg = 'Committed: %s' % pending.revision
 465         if viewvc_url:
 466           viewvc_url = '%s%s' % (viewvc_url.rstrip('/'), pending.revision)
 467           msg = 'Committed: %s' % viewvc_url
 468           issue_desc.append_footer(msg)
 469
 470         # Update the CQ dashboard.
 471         self.context.status.send(
 472             pending,
 473             { 'verification': 'commit',
 474               'payload': {
 475                 'revision': pending.revision,
 476                 'output': msg,
 477                 'url': viewvc_url}})
 478
 479         # Closes the issue on Rietveld.
 480         # TODO(csharp): Retry if exceptions are encountered.
 481         try:
 482           self.context.rietveld.close_issue(pending.issue)
 483           self.context.rietveld.update_description(
 484               pending.issue, issue_desc.description)
 485           self.context.rietveld.add_comment(
 486               pending.issue, 'Change committed as %s' % pending.revision)
 487         except (urllib2.HTTPError, urllib2.URLError) as e:
 488           # Ignore AppEngine flakiness.
 489           logging.warning('Unable to fully close the issue')
 490         # And finally remove the issue. If the close_issue() call above failed,
 491         # it is possible the dashboard will be confused but it is harmless.
 492         try:
 493           self.queue.get(pending.issue)
 494         except KeyError:
 495           logging.error('Internal inconsistency for %d', pending.issue)
 496         self.queue.remove(pending.issue)
 497       except (
 498           checkout.PatchApplicationFailed, patch.UnsupportedPatchFormat) as e:
 499         raise base.DiscardPending(pending, str(e))
 500       except subprocess2.CalledProcessError as e:
 501         stdout = getattr(e, 'stdout', None)
 502         out = 'Failed to apply the patch.'
 503         if stdout:
 504           out += '\n%s' % stdout
 505         raise base.DiscardPending(pending, out)
 506     except base.DiscardPending as e:
 507       self._discard_pending(e.pending, e.status)
 508     except Exception as e:
 509       traceback.print_exc()
 510       # Swallow every exception in that code and move on. Make sure to send a
 511       # stack trace though.
 512       errors.send_stack(e)
 513
 514   def _throttle(self, pending):
 515     """Returns True if a commit should be delayed."""
 516     if pending.postpone():
 517       self.context.status.send(
 518           pending,
 519           {'verification': 'why not',
 520            'payload': {
 521                'message': pending.why_not()}})
 522       return True
 523     if not self.recent_commit_timestamps:
 524       return False
 525     cutoff = time.time() - self.COMMIT_BURST_DELAY
 526     bursted = len([True for i in self.recent_commit_timestamps if i > cutoff])
 527
 528     if bursted >= self.MAX_COMMIT_BURST:
 529       self.context.status.send(
 530           pending,
 531           {'verification': 'why not',
 532            'payload': {
 533                'message': ('Patch is ready to commit, but the CQ is delaying '
 534                            'it because CQ has already submitted %d patchs in '
 535                            'the last %d seconds' %
 536                            (self.MAX_COMMIT_BURST, self.COMMIT_BURST_DELAY))}})
 537       return True
 538
 539     return False
 540
 541   def load(self, filename):
 542     """Loads the commit queue state from a JSON file."""
 543     self.queue = model.load_from_json_file(filename)
 544
 545   def save(self, filename):
 546     """Save the commit queue state in a simple JSON file."""
 547     model.save_to_json_file(filename, self.queue)
 548
 549   def close(self):
 550     """Close all the active pending manager items."""
 551     self.context.status.close()