New RietveldPollerWithCache is working stable. Removed old RietveldPoller and the flag to choose be…

scripts/master/try_job_rietveld.py

 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import datetime
 import hashlib
 import json
 import os
 import pytz
 import time
@@ skipping 86 matching lines @@
                           postdata=urllib.urlencode(params))
 
   def _MakeSet(self, data):
     """Converts the input data string into a set of email addresses.
     """
     emails = (email.strip() for email in data.splitlines())
     self._users = frozenset(email for email in emails if email)
     log.msg('Found %d users' % len(self._users))
 
 
-class _RietveldPoller(base.PollingChangeSource):
-  """Polls Rietveld for any pending patch sets to build.
-
-  Periodically polls Rietveld to see if any patch sets have been marked by
-  users to be tried.  If so, send them to the trybots.
-  """
-
-  def __init__(self, get_pending_endpoint, interval, cachepath=None):
-    """
-    Args:
-      get_pending_endpoint: Rietveld URL string used to retrieve jobs to try.
-      interval: Interval used to poll Rietveld, in seconds.
-      cachepath: Path to file where state to persist between master restarts
-                 will be stored.
-    """
-    # Set interval time in base class.
-    self.pollInterval = interval
-
-    # A string URL for the Rietveld endpoint to query for pending try jobs.
-    self._get_pending_endpoint = get_pending_endpoint
-
-    # Cursor used to keep track of next patchset(s) to try.  If the cursor
-    # is None, then try from the beginning.
-    self._cursor = None
-
-    # Try job parent of this poller.
-    self._try_job_rietveld = None
-
-    self._cachepath = cachepath
-
-    if self._cachepath:
-      if os.path.exists(self._cachepath):
-        with open(self._cachepath) as f:
-          # Using JSON allows us to be flexible and extend the format
-          # in a compatible way.
-          data = json.load(f)
-          self._cursor = data.get('cursor').encode('utf-8')
-
-  # base.PollingChangeSource overrides:
-  def poll(self):
-    """Polls Rietveld for any pending try jobs and submit them.
-
-    Returns:
-      A deferred objects to be called once the operation completes.
-    """
-    log.msg('RietveldPoller.poll')
-    d = defer.succeed(None)
-    d.addCallback(self._OpenUrl)
-    d.addCallback(self._ParseJson)
-    d.addErrback(log.err, 'error in RietveldPoller')  # eat errors
-    return d
-
-  def setServiceParent(self, parent):
-    base.PollingChangeSource.setServiceParent(self, parent)
-    self._try_job_rietveld = parent
-
-  def _OpenUrl(self, _):
-    """Downloads pending patch sets from Rietveld.
-
-    Returns: A string containing the pending patchsets from Rietveld
-        encoded as JSON.
-    """
-    endpoint = self._get_pending_endpoint
-    if self._cursor:
-      sep = '&' if '?' in endpoint else '?'
-      endpoint = endpoint + '%scursor=%s' % (sep, self._cursor)
-
-    log.msg('RietveldPoller._OpenUrl: %s' % endpoint)
-    return client.getPage(endpoint, agent='buildbot', timeout=2*60)
-
-  def _ParseJson(self, json_string):
-    """Parses the JSON pending patch set information.
-
-    Args:
-      json_string: A string containing the serialized JSON jobs.
-
-    Returns: A list of pending try jobs.  This is the list of all jobs returned
-        by Rietveld, not simply the ones we tried this time.
-    """
-    data = json.loads(json_string)
-    d = self._try_job_rietveld.SubmitJobs(data['jobs'])
-    def success_callback(value):
-      self._cursor = str(data['cursor'])
-      self._try_job_rietveld.processed_keys.clear()
-
-      if self._cachepath:
-        with open(self._cachepath, 'w') as f:
-          json.dump({'cursor': self._cursor}, f)
-
-      return value
-    d.addCallback(success_callback)
-    return d
-
-
 class _RietveldPollerWithCache(base.PollingChangeSource):
   """Polls Rietveld for any pending patch sets to build.
 
   Periodically polls Rietveld to see if any patch sets have been marked by
   users to be tried. If so, send them to the trybots. Uses cursor to download
   all pages within a single poll. To avoid sending same jobs, keeps a cache of
   the jobs that were already processed thus reducing chances of a duplicate job
   submission and increasing robustness to bugs in Rietveld. On the first poll
   this cache is initialized with jobs currently pending on the Buildbot.
   """
@@ skipping 133 matching lines @@
         num_removed += 1
     log.msg('[RPWC] Removed %d old jobs from the cache.' % num_removed)
     self._processed_keys = new_processed_keys
 
 
 class TryJobRietveld(TryJobBase):
   """A try job source that gets jobs from pending Rietveld patch sets."""
 
   def __init__(self, name, pools, properties=None, last_good_urls=None,
                code_review_sites=None, project=None, filter_master=False,
-               cachepath=None, cache_processed_jobs=False):
+               cachepath=None):
     """Creates a try job source for Rietveld patch sets.
 
     Args:
       name: Name of this scheduler.
       pools: No idea.
       properties: Extra build properties specific to this scheduler.
       last_good_urls: Dictionary of project to last known good build URL.
       code_review_sites: Dictionary of project to code review site.  This
           class care only about the 'chrome' project.
       project: The name of the project whose review site URL to extract.
           If the project is not found in the dictionary, an exception is
           raised.
       filter_master: Filter try jobs by master name. Necessary if several try
           masters share the same rietveld instance.
     """
     TryJobBase.__init__(self, name, pools, properties,
                         last_good_urls, code_review_sites)
     endpoint = self._GetRietveldEndPointForProject(
         code_review_sites, project, filter_master)
 
-    if cache_processed_jobs:
-      self._poller = _RietveldPollerWithCache(endpoint, interval=10)
-    else:
-      self._poller = _RietveldPoller(endpoint, interval=10, cachepath=cachepath)
+    self._poller = _RietveldPollerWithCache(endpoint, interval=10)
     self._valid_users = _ValidUserPoller(interval=12 * 60 * 60)
     self._project = project
 
     # Cleared by _RietveldPoller._ParseJson's success callback.
     self.processed_keys = set()
 
     log.msg('TryJobRietveld created, get_pending_endpoint=%s '
             'project=%s' % (endpoint, project))
 
   @staticmethod
@@ skipping 110 matching lines @@
         else:
           self.processed_keys.add(job['key'])
           log.err('Rietveld not updated: no corresponding service found.')
 
   # TryJobBase overrides:
   def setServiceParent(self, parent):
     TryJobBase.setServiceParent(self, parent)
     self._poller.setServiceParent(self)
     self._poller.master = self.master
     self._valid_users.setServiceParent(self)