Recognize crashed tests as failures in sharding_supervisor

tools/sharding_supervisor/sharding_supervisor.py

Index: tools/sharding_supervisor/sharding_supervisor.py
diff --git a/tools/sharding_supervisor/sharding_supervisor.py b/tools/sharding_supervisor/sharding_supervisor.py
index 2c4bd606dfff4ff93af21a2dc522aa393c11a026..75c0cfb68860f2f65c8c8bf682ce7a4c4cc78b78 100755
--- a/tools/sharding_supervisor/sharding_supervisor.py
+++ b/tools/sharding_supervisor/sharding_supervisor.py
@@ -81,21 +81,45 @@ class ShardRunner(threading.Thread):
     counter: Called to get the next shard index to run.
   """
 
-  def __init__(self, supervisor, counter, test_fail, test_timeout):
+  def __init__(self, supervisor, counter, test_start, test_ok, test_fail,
+               test_timeout):
     """Inits ShardRunner with a supervisor and counter."""
     threading.Thread.__init__(self)
     self.supervisor = supervisor
     self.counter = counter
+    self.test_start = test_start
+    self.test_ok = test_ok
     self.test_fail = test_fail
     self.test_timeout = test_timeout
+    self.current_test = ""
 
-  def SearchForFailure(self, regex, prefix, line, description):
-    results = regex.search(line)
+  def ReportFailure(self, description, prefix, test_name):
+    log_line = "%s: %s%s\n" % (description, prefix, test_name)
+    self.supervisor.LogLineFailure(log_line)
+
+  def ProcessLine(self, prefix, line):
+    results = self.test_start.search(line)
+    if results:
+      if self.current_test:
+        self.ReportFailure("CRASHED", prefix, results.group(1))
+      self.current_test = results.group(1)
+      return
+
+    results = self.test_ok.search(line)
+    if results:
+      self.current_test = ""
+      return
+
+    results = self.test_fail.search(line)
+    if results:
+      self.ReportFailure("FAILED", prefix, results.group(1))
+      self.current_test = ""
+      return
+
+    results = self.test_timeout.search(line)

[Paweł Hajdan Jr., 2011/08/15 17:33:17]

Don't rely on that line to be present. It's fragile.

[clee, 2011/08/15 18:07:09]

It's what gtest_command uses to detect timeouts. There isn't a '[  FAILED  ]'
line for timeouts, so this seems to be the only way to detect one. Or can
timeouts and crashes be grouped together (as 'incomplete' perhaps?), in which
case the current check for 'crashed' would suffice? Based on past experience,
would there be an advantage in having timeouts and crashes be two separate
categories? wdyt? 

[Paweł Hajdan Jr., 2011/08/16 17:36:07]

And guess what? It's broken.
It's not really useful to trying to distinguish between crashes and hangs IMO.

I think that a more reliable solution would be to detect that you see one test
starting, but you never see it finishing: you either encounter EOF or another
test starting. That is a pretty good sign something bad has happened.

     if results:
-      log_line = "%s: %s%s\n" % (description, prefix, results.group(1))
-      self.supervisor.LogLineFailure(log_line)
-      return True
-    return False
+      self.ReportFailure("TIMEOUT", prefix, results.group(1))
+      self.current_test = ""
 
   def run(self):
     """Runs shards and outputs the results.
@@ -124,9 +148,7 @@ class ShardRunner(threading.Thread):
           line = chars.getvalue()
           if not line and not shard_running:
             break
-          if not self.SearchForFailure(
-              self.test_fail, prefix, line, "FAILED"):
-            self.SearchForFailure(self.test_timeout, prefix, line, "TIMEOUT")
+          self.ProcessLine(prefix, line)
           line = prefix + line
           self.supervisor.LogOutputLine(index, line)
           chars.close()
@@ -185,11 +207,14 @@ class ShardingSupervisor(object):
     test_name_regex = r"((\w+/)?\w+\.\w+(/\d+)?)"
 
     # Regex for filtering out ANSI escape codes when using color.
-    ansi_code_regex = r"(?:\x1b\[.*?[a-zA-Z])?"
+    ansi_regex = r"(?:\x1b\[.*?[a-zA-Z])?"
 
+    test_start = re.compile(
+        ansi_regex + "\[\s+RUN\s+\] " + ansi_regex + test_name_regex)

[M-A Ruel, 2011/08/15 15:46:08]

Use r"".

[clee, 2011/08/15 18:07:09]

Done.

+    test_ok = re.compile(
+        ansi_regex + "\[\s+OK\s+\] " + ansi_regex + test_name_regex)
     test_fail = re.compile(
-        ansi_code_regex + "\[\s+FAILED\s+\] " + ansi_code_regex +
-        test_name_regex)
+        ansi_regex + "\[\s+FAILED\s+\] " + ansi_regex + test_name_regex)
     test_timeout = re.compile(
         "Test timeout \([0-9]+ ms\) exceeded for " + test_name_regex)
 
@@ -198,12 +223,9 @@ class ShardingSupervisor(object):
     for i in range(self.num_shards):
       counter.put(i)
 
-    # Disable stdout buffering to read shard output one character at a time.
-    # This allows for shard crashes that do not end with a "\n".
-    sys.stdout = os.fdopen(sys.stdout.fileno(), "w", 0)
-
     for i in range(self.num_runs):
-      worker = ShardRunner(self, counter, test_fail, test_timeout)
+      worker = ShardRunner(
+          self, counter, test_start, test_ok, test_fail, test_timeout)
       worker.start()
       workers.append(worker)
     if self.reorder:
@@ -212,10 +234,7 @@ class ShardingSupervisor(object):
       for worker in workers:
         worker.join()
 
-    # Re-enable stdout buffering.
-    sys.stdout = sys.__stdout__
-
-    return self.PrintSummary()
+    return self.PrintSummary(self.failure_log)
 
   def LogLineFailure(self, line):
     """Saves a line in the failure log to be printed at the end.
@@ -251,7 +270,7 @@ class ShardingSupervisor(object):
   def ShardIndexCompleted(self, index):
     self.shard_output[index].put(self.SHARD_COMPLETED)
 
-  def PrintSummary(self):
+  def PrintSummary(self, failed_tests):
     """Prints a summary of the test results.
 
     If any shards had failing tests, the list is sorted and printed. Then all
@@ -264,25 +283,26 @@ class ShardingSupervisor(object):
     num_failed = len(self.failed_shards)
     if num_failed > 0:
       self.failed_shards.sort()
-      if self.color:
-        sys.stderr.write("\x1b[1;5;31m")
-      sys.stderr.write("SHARDS THAT FAILED: %s\n" % str(self.failed_shards))
+      self.WriteText(sys.stderr,
+                     "FAILED SHARDS: %s\n" % str(self.failed_shards),
+                     "\x1b[1;5;31m")
     else:
-      if self.color:
-        sys.stderr.write("\x1b[1;5;32m")
-      sys.stderr.write("ALL SHARDS PASSED!\n")
-    if self.failure_log:
-      if self.color:
-        sys.stderr.write("\x1b[1;5;31m")
-      sys.stderr.write("TESTS THAT DID NOT PASS:\n")
-      if self.color:
-        sys.stderr.write("\x1b[m")
-      for line in self.failure_log:
+      self.WriteText(sys.stderr, "ALL SHARDS PASSED!\n", "\x1b[1;5;32m")
+    if failed_tests:
+      self.WriteText(sys.stderr, "FAILED TESTS:\n", "\x1b[1;5;31m")
+      for line in failed_tests:
         sys.stderr.write(line)
     if self.color:
       sys.stderr.write("\x1b[m")
     return num_failed
 
+  def WriteText(self, pipe, text, ansi):
+    if self.color:
+      pipe.write(ansi)
+    pipe.write(text)
+    if self.color:
+      pipe.write("\x1b[m")
+
 
 def main():
   parser = optparse.OptionParser(usage=SS_USAGE)