autotest: Make crash tests resilient to builds with high ambient crash rates

client/bin/site_crash_test.py

 # Copyright (c) 2010 The Chromium OS 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 logging, os, re
 from autotest_lib.client.bin import site_log_reader, site_utils, test
 from autotest_lib.client.common_lib import error, utils
 
 
 class CrashTest(test.test):
 
     _CONSENT_FILE = '/home/chronos/Consent To Send Stats'
+    _CORE_PATTERN = '/proc/sys/kernel/core_pattern'
     _CRASH_REPORTER_PATH = '/sbin/crash_reporter'
     _CRASH_SENDER_PATH = '/sbin/crash_sender'
     _CRASH_SENDER_RATE_DIR = '/var/lib/crash_sender'
     _CRASH_SENDER_RUN_PATH = '/var/run/crash_sender.pid'
     _MOCK_CRASH_SENDING = '/tmp/mock-crash-sending'
     _PAUSE_FILE = '/var/lib/crash_sender_paused'
     _SYSTEM_CRASH_DIR = '/var/spool/crash'
     _USER_CRASH_DIR = '/home/chronos/user/crash'
 
     def _set_system_sending(self, is_enabled):
@@ skipping 72 matching lines @@
 
     def _get_crash_dir(self, username):
         if username == 'chronos':
             return self._USER_CRASH_DIR
         else:
             return self._SYSTEM_CRASH_DIR
 
 
     def _initialize_crash_reporter(self):
         utils.system('%s --init --nounclean_check' % self._CRASH_REPORTER_PATH)
+        # Completely disable crash_reporter from generating crash dumps
+        # while any tests are running, otherwise a crashy system can make
+        # these tests flaky.
+        self.enable_crash_filtering('none')
 
 
     def write_crash_dir_entry(self, name, contents):
         entry = os.path.join(self._SYSTEM_CRASH_DIR, name)
         if not os.path.exists(self._SYSTEM_CRASH_DIR):
             os.makedirs(self._SYSTEM_CRASH_DIR)
         utils.open_write_close(entry, contents)
         return entry
 
 
@@ skipping 156 matching lines @@
 
         # Show the result for debugging but remove 'output' key
         # since it's large and earlier in debug output.
         debug_result = dict(result)
         del debug_result['output']
         logging.debug('Result of send (besides output): %s' % debug_result)
 
         return result
 
 
+    def _replace_crash_reporter_filter_in(self, new_parameter):
+        core_pattern = utils.read_file(self._CORE_PATTERN)[:-1]
+        core_pattern = re.sub('--filter_in=\S*\s*', '',
+                              core_pattern).rstrip()
+        if new_parameter:
+            core_pattern += ' ' + new_parameter
+        utils.system('echo "%s" > %s' % (core_pattern, self._CORE_PATTERN))
+
+
+    def enable_crash_filtering(self, name):
+        self._replace_crash_reporter_filter_in('--filter_in=' + name)
+
+
+    def disable_crash_filtering(self):
+        self._replace_crash_reporter_filter_in('')
+
+
     def initialize(self):
         test.test.initialize(self)
         self._log_reader = site_log_reader.LogReader()
         self._leave_crash_sending = True
         self._automatic_consent_saving = True
+        self.enable_crash_filtering('none')
 
 
     def cleanup(self):
         self._reset_rate_limiting()
         self._clear_spooled_crashes()
         self._set_system_sending(self._leave_crash_sending)
         self._set_sending_mock(mock_enabled=False)
         if self._automatic_consent_saving:
             self._pop_consent()
+        self.disable_crash_filtering()
         test.test.cleanup(self)
 
 
     def run_crash_tests(self,
                         test_names,
                         initialize_crash_reporter=False,
                         clear_spool_first=True,
                         must_run_all=True):
         """Run crash tests defined in this class.
 
@@ skipping 20 matching lines @@
                 self._initialize_crash_reporter()
             # Disable crash_sender from running, kill off any running ones, but
             # set environment so crash_sender may run as a child process.
             self._set_system_sending(False)
             self._set_child_sending(True)
             self._kill_running_sender()
             self._reset_rate_limiting()
             if clear_spool_first:
                 self._clear_spooled_crashes()
             getattr(self, '_test_' + test_name)()