1# Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5import logging 6import math 7import threading 8 9import common 10from autotest_lib.client.common_lib import env 11from autotest_lib.client.common_lib import error 12from autotest_lib.client.common_lib import utils 13from autotest_lib.client.common_lib.cros import retry 14from autotest_lib.frontend.afe.json_rpc import proxy 15from autotest_lib.server import frontend 16try: 17 from chromite.lib import retry_util 18 from chromite.lib import timeout_util 19except ImportError: 20 logging.warn('Unable to import chromite.') 21 retry_util = None 22 timeout_util = None 23 24try: 25 from chromite.lib import metrics 26except ImportError: 27 logging.warn('Unable to import metrics from chromite.') 28 metrics = utils.metrics_mock 29 30 31def convert_timeout_to_retry(backoff, timeout_min, delay_sec): 32 """Compute the number of retry attempts for use with chromite.retry_util. 33 34 @param backoff: The exponential backoff factor. 35 @param timeout_min: The maximum amount of time (in minutes) to sleep. 36 @param delay_sec: The amount to sleep (in seconds) between each attempt. 37 38 @return: The number of retry attempts in the case of exponential backoff. 39 """ 40 # Estimate the max_retry in the case of exponential backoff: 41 # => total_sleep = sleep*sum(r=0..max_retry-1, backoff^r) 42 # => total_sleep = sleep( (1-backoff^max_retry) / (1-backoff) ) 43 # => max_retry*ln(backoff) = ln(1-(total_sleep/sleep)*(1-backoff)) 44 # => max_retry = ln(1-(total_sleep/sleep)*(1-backoff))/ln(backoff) 45 total_sleep = timeout_min * 60 46 numerator = math.log10(1-(total_sleep/delay_sec)*(1-backoff)) 47 denominator = math.log10(backoff) 48 return int(math.ceil(numerator/denominator)) 49 50 51class RetryingAFE(frontend.AFE): 52 """Wrapper around frontend.AFE that retries all RPCs. 53 54 Timeout for retries and delay between retries are configurable. 55 """ 56 def __init__(self, timeout_min=30, delay_sec=10, **dargs): 57 """Constructor 58 59 @param timeout_min: timeout in minutes until giving up. 60 @param delay_sec: pre-jittered delay between retries in seconds. 61 """ 62 self.timeout_min = timeout_min 63 self.delay_sec = delay_sec 64 super(RetryingAFE, self).__init__(**dargs) 65 66 67 def set_timeout(self, timeout_min): 68 """Set timeout minutes for the AFE server. 69 70 @param timeout_min: The timeout minutes for AFE server. 71 """ 72 self.timeout_min = timeout_min 73 74 75 def run(self, call, **dargs): 76 if retry_util is None: 77 raise ImportError('Unable to import chromite. Please consider to ' 78 'run build_externals to build site packages.') 79 # exc_retry: We retry if this exception is raised. 80 # blacklist: Exceptions that we raise immediately if caught. 81 exc_retry = Exception 82 blacklist = (ImportError, error.RPCException, proxy.JSONRPCException, 83 timeout_util.TimeoutError) 84 backoff = 2 85 max_retry = convert_timeout_to_retry(backoff, self.timeout_min, 86 self.delay_sec) 87 88 def _run(self, call, **dargs): 89 return super(RetryingAFE, self).run(call, **dargs) 90 91 def handler(exc): 92 """Check if exc is an exc_retry or if it's blacklisted. 93 94 @param exc: An exception. 95 96 @return: True if exc is an exc_retry and is not 97 blacklisted. False otherwise. 98 """ 99 is_exc_to_check = isinstance(exc, exc_retry) 100 is_blacklisted = isinstance(exc, blacklist) 101 return is_exc_to_check and not is_blacklisted 102 103 # If the call is not in main thread, signal can't be used to abort the 104 # call. In that case, use a basic retry which does not enforce timeout 105 # if the process hangs. 106 @retry.retry(Exception, timeout_min=self.timeout_min, 107 delay_sec=self.delay_sec, 108 blacklist=[ImportError, error.RPCException, 109 proxy.ValidationError]) 110 def _run_in_child_thread(self, call, **dargs): 111 return super(RetryingAFE, self).run(call, **dargs) 112 113 if isinstance(threading.current_thread(), threading._MainThread): 114 # Set the keyword argument for GenericRetry 115 dargs['sleep'] = self.delay_sec 116 dargs['backoff_factor'] = backoff 117 # timeout_util.Timeout fundamentally relies on sigalrm, and doesn't 118 # work at all in wsgi environment (just emits logs spam). So, don't 119 # use it in wsgi. 120 try: 121 if env.IN_MOD_WSGI: 122 return retry_util.GenericRetry(handler, max_retry, _run, 123 self, call, **dargs) 124 with timeout_util.Timeout(self.timeout_min * 60): 125 return retry_util.GenericRetry(handler, max_retry, _run, 126 self, call, **dargs) 127 except timeout_util.TimeoutError: 128 c = metrics.Counter( 129 'chromeos/autotest/retrying_afe/retry_timeout') 130 # Reserve field job_details for future use. 131 f = {'destination_server': self.server.split(':')[0], 132 'call': call, 133 'job_details': ''} 134 c.increment(fields=f) 135 raise 136 else: 137 return _run_in_child_thread(self, call, **dargs) 138 139 140class RetryingTKO(frontend.TKO): 141 """Wrapper around frontend.TKO that retries all RPCs. 142 143 Timeout for retries and delay between retries are configurable. 144 """ 145 def __init__(self, timeout_min=30, delay_sec=10, **dargs): 146 """Constructor 147 148 @param timeout_min: timeout in minutes until giving up. 149 @param delay_sec: pre-jittered delay between retries in seconds. 150 """ 151 self.timeout_min = timeout_min 152 self.delay_sec = delay_sec 153 super(RetryingTKO, self).__init__(**dargs) 154 155 156 def run(self, call, **dargs): 157 @retry.retry(Exception, timeout_min=self.timeout_min, 158 delay_sec=self.delay_sec, 159 blacklist=[ImportError, error.RPCException, 160 proxy.ValidationError]) 161 def _run(self, call, **dargs): 162 return super(RetryingTKO, self).run(call, **dargs) 163 return _run(self, call, **dargs) 164