Make allow_subannotations more robust.

scripts/common/annotator.py

 #!/usr/bin/env python
 # Copyright (c) 2013 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.
 
 """Contains generating and parsing systems of the Chromium Buildbot Annotator.
 
 When executed as a script, this reads step name / command pairs from a file and
 executes those lines while annotating the output. The input is json:
 
@@ skipping 446 matching lines @@
 
     step.step_trigger(json.dumps({
         'builderNames': [builder_name],
         'bucket': trig.get('bucket'),
         'changes': changes,
         'properties': trig.get('properties'),
     }, sort_keys=True))
 
 
 def run_step(stream, name, cmd,
+             step_annotation,
              cwd=None, env=None,
-             allow_subannotations=False,
+             subannotator=None,
              trigger_specs=None,
              **kwargs):
   """Runs a single step.
 
   Context:
     stream: StructuredAnnotationStream to use to emit step
+    step_annotation: optional StructuredAnnotationStep to use instead
+        of creating one
 
   Step parameters:
     name: name of the step, will appear in buildbots waterfall
     cmd: command to run, list of one or more strings
     cwd: absolute path to working directory for the command
     env: dict with overrides for environment variables
-    allow_subannotations: if True, lets the step emit its own annotations
+    subannotator: a callback_implementor class used to parse subannotations
+        that the command emits; if None, subannotations will be suppressed.
     trigger_specs: a list of trigger specifications, which are dict with keys:
         properties: a dict of properties.
             Buildbot requires buildername property.
 
   Known kwargs:
     stdout: Path to a file to put step stdout into. If used, stdout won't appear
             in annotator's stdout (and |allow_subannotations| is ignored).
     stderr: Path to a file to put step stderr into. If used, stderr won't appear
             in annotator's stderr.
     stdin: Path to a file to read step stdin from.
 
   Returns the returncode of the step.
   """
   if isinstance(cmd, basestring):
     cmd = (cmd,)
   cmd = map(str, cmd)
 
   # For error reporting.
   step_dict = kwargs.copy()
   step_dict.update({
       'name': name,
       'cmd': cmd,
       'cwd': cwd,
       'env': env,
-      'allow_subannotations': allow_subannotations,
+      'allow_subannotations': subannotator is not None,
       })
   step_env = _merge_envs(os.environ, env)
 
-  step_annotation = stream.step(name)
-  step_annotation.step_started()
-
   print_step(step_dict, step_env, stream)
   returncode = 0
   if cmd:
     try:
       # Open file handles for IO redirection based on file names in step_dict.
       fhandles = {
         'stdout': subprocess.PIPE,
         'stderr': subprocess.PIPE,
         'stdin': None,
       }
@@ skipping 27 matching lines @@
             universal_newlines=True,
             creationflags=creationflags,
             **fhandles)
 
       # Safe to close file handles now that subprocess has inherited them.
       for handle in fhandles.itervalues():
         if isinstance(handle, file):
           handle.close()
 
       outlock = threading.Lock()
-      def filter_lines(lock, allow_subannotations, inhandle, outhandle):
+      def filter_lines(inhandle, outhandle):
         while True:
           line = inhandle.readline()
           if not line:
             break
-          lock.acquire()
-          try:
-            if not allow_subannotations and line.startswith('@@@'):
-              outhandle.write('!')
-            outhandle.write(line)
+          with outlock:
+            if line.startswith('@@@'):
+              if subannotator:
+                # The subannotator might write to the handle, thus the lock.
+                MatchAnnotation(line.strip(), subannotator)
+              else:
+                outhandle.write('!')
+                outhandle.write(line)
+            else:
+              outhandle.write(line)
             outhandle.flush()
-          finally:
-            lock.release()
 
       # Pump piped stdio through filter_lines. IO going to files on disk is
       # not filtered.
       threads = []
       for key in ('stdout', 'stderr'):
         if fhandles[key] == subprocess.PIPE:
           inhandle = getattr(proc, key)
           outhandle = getattr(sys, key)
           threads.append(threading.Thread(
               target=filter_lines,
-              args=(outlock, allow_subannotations, inhandle, outhandle)))
+              args=(inhandle, outhandle)))
 
       for th in threads:
         th.start()
       proc.wait()
       for th in threads:
         th.join()
       returncode = proc.returncode
     except OSError:
       # File wasn't found, error will be reported to stream when the exception
       # crosses the context manager.
       step_annotation.step_exception_occured(*sys.exc_info())
       raise
 
   # TODO(martiniss) move logic into own module?
   if trigger_specs:
     triggerBuilds(step_annotation, trigger_specs)
 
-  return step_annotation, returncode
+  return returncode
 
 def update_build_failure(failure, retcode, **_kwargs):
   """Potentially moves failure from False to True, depending on returncode of
   the run step and the step's configuration.
 
   can_fail_build: A boolean indicating that a bad retcode for this step should
                   be intepreted as a build failure.
 
   Returns new value for failure.
 
@@ skipping 10 matching lines @@
       sys.exit(1)
 
   stream = StructuredAnnotationStream()
   ret_codes = []
   build_failure = False
   prev_annotation = None
   for step in steps:
     if build_failure and not step.get('always_run', False):
       ret = None
     else:
-      prev_annotation, ret = run_step(stream, **step)
+      prev_annotation = stream.step(step['name'])
+      prev_annotation.step_started()
+      ret = run_step(stream, step_annotation=prev_annotation, **step)
       stream = prev_annotation.annotation_stream
       if ret > 0:
         stream.step_cursor(stream.current_step)
         stream.emit('step returned non-zero exit code: %d' % ret)
         prev_annotation.step_failure()
 
       prev_annotation.step_ended()
     build_failure = update_build_failure(build_failure, ret)
     ret_codes.append(ret)
   if prev_annotation:
@@ skipping 40 matching lines @@
     steps.extend(json.load(sys.stdin, object_hook=force_dict_strs))
   else:
     with open(args[0], 'rb') as f:
       steps.extend(json.load(f, object_hook=force_dict_strs))
 
   return 1 if run_steps(steps, False)[0] else 0
 
 
 if __name__ == '__main__':
   sys.exit(main())