Add LogDog / annotation protobuf support.

recipes.py

 #!/usr/bin/env python
 # Copyright 2015 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Tool to interact with recipe repositories.
 
 This tool operates on the nearest ancestor directory containing an
 infra/config/recipes.cfg.
 """
@@ skipping 74 matching lines @@
     return recipe_result.result['status_code']
   else:
     return 0
 
 
 def run(package_deps, args, op_args):
   from recipe_engine import run as recipe_run
   from recipe_engine import loader
   from recipe_engine import step_runner
   from recipe_engine import stream
+  from recipe_engine import stream_logdog
 
   def get_properties_from_args(args):
     properties = dict(x.split('=', 1) for x in args)
     for key, val in properties.iteritems():
       try:
         properties[key] = json.loads(val)
       except (ValueError, SyntaxError):
         pass  # If a value couldn't be evaluated, keep the string version
     return properties
 
@@ skipping 40 matching lines @@
           package_deps, config_file), package_deps.root_package)
 
   workdir = (args.workdir or
       os.path.join(os.path.dirname(os.path.realpath(__file__)), 'workdir'))
   logging.info('Using %s as work directory' % workdir)
   if not os.path.exists(workdir):
     os.makedirs(workdir)
 
   old_cwd = os.getcwd()
   os.chdir(workdir)
-  stream_engine = stream.ProductStreamEngine(
-      stream.StreamEngineInvariants(),
-      stream.AnnotatorStreamEngine(
+
+  # Construct our stream engines. We may want to share stream events with more
+  # than one StreamEngine implementation, so we will accumulate them in a
+  # "stream_engines" list and compose them into a MultiStreamEngine.
+  def build_annotation_stream_engine():
+    return stream.AnnotatorStreamEngine(
         sys.stdout,
         emit_timestamps=(args.timestamps or
-                         op_args.annotation_flags.emit_timestamp)))
-  with stream_engine:
+                         op_args.annotation_flags.emit_timestamp),
+    )
+
+  stream_engines = []
+  if op_args.logdog.streamserver_uri:
+    logging.debug('Using LogDog with parameters: [%s]', op_args.logdog)
+    stream_engines.append(stream_logdog.StreamEngine(
+        streamserver_uri=op_args.logdog.streamserver_uri,
+        name_base=(op_args.logdog.name_base or None),
+    ))
+
+    # If we're teeing, also fold in a standard annotation stream engine.
+    if op_args.logdog.tee:
+      stream_engines.append(build_annotation_stream_engine())
+  else:
+    # Not using LogDog; use a standard annotation stream engine.
+    stream_engines.append(build_annotation_stream_engine())
+  multi_stream_engine = stream.MultiStreamEngine.create(*stream_engines)
+
+  # Have a top-level set of invariants to enforce StreamEngine expectations.
+  with stream.StreamEngineInvariants.wrap(multi_stream_engine) as stream_engine:
     # Emit initial properties if configured to do so.
     if op_args.annotation_flags.emit_initial_properties:
       with stream_engine.new_step_stream('Initial Properties') as s:
         for key in sorted(properties.iterkeys()):
           s.set_build_property(key, json.dumps(properties[key], sort_keys=True))
 
     try:
       ret = recipe_run.run_steps(
           properties, stream_engine,
           step_runner.SubprocessStepRunner(stream_engine),
@@ skipping 403 matching lines @@
 
   if not isinstance(ret, int):
     if ret is None:
       ret = 0
     else:
       print >> sys.stderr, ret
       ret = 1
   sys.stdout.flush()
   sys.stderr.flush()
   os._exit(ret)