[WIP] switch usages of log to tokio_trace

src/python/pants/bin/local_pants_runner.py

@@ -40,7 +40,7 @@ def parse_options(args, env, setup_logging=False, options_bootstrapper=None):
     return options, build_config, options_bootstrapper
 
   @staticmethod
-  def _maybe_init_graph_session(graph_session, options_bootstrapper,build_config, global_options):
+  def _maybe_init_graph_session(graph_session, options_bootstrapper,build_config, options):
     if graph_session:
       return graph_session
 
@@ -52,7 +52,9 @@ def _maybe_init_graph_session(graph_session, options_bootstrapper,build_config,
       build_config
     )
 
-    return graph_scheduler_helper.new_session(global_options.v2_ui)
+    v2_ui = options.for_global_scope().v2_ui
+    zipkin_trace_v2 = options.for_scope('reporting').zipkin_trace_v2
+    return graph_scheduler_helper.new_session(zipkin_trace_v2, v2_ui)
 
   @staticmethod
   def _maybe_init_target_roots(target_roots, graph_session, options, build_root):
@@ -106,7 +108,7 @@ def create(cls, exiter, args, env, target_roots=None, daemon_graph_session=None,
       daemon_graph_session,
       options_bootstrapper,
       build_config,
-      global_options
+      options
     )
 
     target_roots = cls._maybe_init_target_roots(

src/python/pants/engine/native.py

@@ -683,8 +683,8 @@ def new_execution_request(self):
       self.lib.execution_request_create(),
       self.lib.execution_request_destroy)
 
-  def new_session(self, scheduler, should_render_ui, ui_worker_count):
-    return self.gc(self.lib.session_create(scheduler, should_render_ui, ui_worker_count), self.lib.session_destroy)
+  def new_session(self, scheduler, should_record_zipkin_spans, should_render_ui, ui_worker_count):
+    return self.gc(self.lib.session_create(scheduler, should_record_zipkin_spans, should_render_ui, ui_worker_count), self.lib.session_destroy)
 
   def new_scheduler(self,
                     tasks,

src/python/pants/engine/scheduler.py

@@ -354,9 +354,11 @@ def lease_files_in_graph(self):
   def garbage_collect_store(self):
     self._native.lib.garbage_collect_store(self._scheduler)
 
-  def new_session(self, v2_ui=False):
+  def new_session(self, zipkin_trace_v2, v2_ui=False):
     """Creates a new SchedulerSession for this Scheduler."""
-    return SchedulerSession(self, self._native.new_session(self._scheduler, v2_ui, multiprocessing.cpu_count()))
+    return SchedulerSession(self, self._native.new_session(
+      self._scheduler, zipkin_trace_v2, v2_ui, multiprocessing.cpu_count())
+    )
 
 
 _PathGlobsAndRootCollection = Collection.of(PathGlobsAndRoot)
@@ -444,6 +446,12 @@ def metrics(self):
     """Returns metrics for this SchedulerSession as a dict of metric name to metric value."""
     return self._scheduler._metrics(self._session)
 
+  def metrics_have_engine_workunits(self):
+    return "engine_workunits" in self.metrics()
+
+  def engine_workunits(self):
+    return self.metrics().get("engine_workunits")
+
   def with_fork_context(self, func):
     return self._scheduler.with_fork_context(func)
 

src/python/pants/goal/context.py

@@ -149,7 +149,11 @@ def executing(self):
     """A contextmanager that sets metrics in the context of a (v1) engine execution."""
     self._set_target_root_count_in_runtracker()
     yield
-    self.run_tracker.pantsd_stats.set_scheduler_metrics(self._scheduler.metrics())
+    metrics = self._scheduler.metrics()
+    self.run_tracker.pantsd_stats.set_scheduler_metrics(metrics)
+    engine_workunits = self._scheduler.engine_workunits()
+    if engine_workunits:
+      self.run_tracker.report.bulk_record_workunits(engine_workunits)
     self._set_affected_target_count_in_runtracker()
 
   def _set_target_root_count_in_runtracker(self):

src/python/pants/init/engine_initializer.py

@@ -142,8 +142,8 @@ def _tuplify(v):
 class LegacyGraphScheduler(datatype(['scheduler', 'symbol_table', 'goal_map'])):
   """A thin wrapper around a Scheduler configured with @rules for a symbol table."""
 
-  def new_session(self, v2_ui=False):
-    session = self.scheduler.new_session(v2_ui)
+  def new_session(self, zipkin_trace_v2, v2_ui=False):
+    session = self.scheduler.new_session(zipkin_trace_v2, v2_ui)
     return LegacyGraphSession(session, self.symbol_table, self.goal_map)
 
 

src/python/pants/pantsd/service/scheduler_service.py

@@ -174,8 +174,9 @@ def prefork(self, options, options_bootstrapper):
     if graph_len > 0:
       self._logger.debug('graph len was {}, waiting for initial watchman event'.format(graph_len))
       self._watchman_is_running.wait()
-
-    session = self._graph_helper.new_session(options.for_global_scope().v2_ui)
+    v2_ui = options.for_global_scope().v2_ui
+    zipkin_trace_v2 = options.for_scope('reporting').zipkin_trace_v2
+    session = self._graph_helper.new_session(zipkin_trace_v2, v2_ui)
     if options.for_global_scope().loop:
       return session, self._prefork_loop(session, options, options_bootstrapper)
     else:

src/python/pants/reporting/report.py

@@ -140,3 +140,8 @@ def _notify(self):
         if len(s) > 0:
           for reporter in self._reporters.values():
             reporter.handle_output(workunit, label, s)
+
+  def bulk_record_workunits(self, engine_workunits):
+    with self._lock:
+      for reporter in self._reporters.values():
+        reporter.bulk_record_workunits(engine_workunits)

src/python/pants/reporting/reporter.py

@@ -47,6 +47,10 @@ def end_workunit(self, workunit):
     """A workunit has finished."""
     pass
 
+  def bulk_record_workunits(self, engine_workunits):
+    """A collection of workunits from Rust (engine) part"""
+    pass
+
   def handle_log(self, workunit, level, *msg_elements):
     """Handle a message logged by pants code.
 

src/python/pants/reporting/reporting.py

@@ -62,6 +62,8 @@ def register_options(cls, register):
                    'or not set when running a Pants command.')
     register('--zipkin-sample-rate', advanced=True, default=100.0,
               help='Rate at which to sample Zipkin traces. Value 0.0 - 100.0.')
+    register('--zipkin-trace-v2', advanced=True, type=bool, default=False,
+              help='If enabled, the zipkin spans are tracked for v2 engine execution progress.')
 
   def initialize(self, run_tracker, all_options, start_time=None):
     """Initialize with the given RunTracker.

src/python/pants/reporting/zipkin_reporter.py

@@ -7,7 +7,7 @@
 import logging
 
 import requests
-from py_zipkin import Encoding
+from py_zipkin import Encoding, storage
 from py_zipkin.transport import BaseTransportHandler
 from py_zipkin.util import generate_random_64bit_string
 from py_zipkin.zipkin import ZipkinAttrs, create_attrs_for_span, zipkin_span
@@ -64,6 +64,7 @@ def __init__(self, run_tracker, settings, endpoint, trace_id, parent_id, sample_
     self.parent_id = parent_id
     self.sample_rate = float(sample_rate)
     self.endpoint = endpoint
+    self.span_storage = storage.default_span_storage()
 
   def start_workunit(self, workunit):
     """Implementation of Reporter callback."""
@@ -96,19 +97,24 @@ def start_workunit(self, workunit):
           sample_rate=self.sample_rate, # Value between 0.0 and 100.0
         )
         self.trace_id = zipkin_attrs.trace_id
-
+        # TODO delete this line when parent_id will be passed in v2 engine:
+        #  - with ExecutionRequest when Nodes from v2 engine are called by a workunit;
+        #  - when a v2 engine Node is called by another v2 engine Node.
+        self.parent_id = zipkin_attrs.span_id
 
       span = zipkin_span(
         service_name=service_name,
         span_name=workunit.name,
         transport_handler=self.handler,
         encoding=Encoding.V1_THRIFT,
-        zipkin_attrs=zipkin_attrs
+        zipkin_attrs=zipkin_attrs,
+        span_storage=self.span_storage,
       )
     else:
       span = zipkin_span(
         service_name=service_name,
         span_name=workunit.name,
+        span_storage=self.span_storage,
       )
     self._workunits_to_spans[workunit] = span
     span.start()
@@ -129,3 +135,28 @@ def close(self):
     endpoint = self.endpoint.replace("/api/v1/spans", "")
 
     logger.debug("Zipkin trace may be located at this URL {}/traces/{}".format(endpoint, self.trace_id))
+
+  def bulk_record_workunits(self, engine_workunits):
+    """A collection of workunits from v2 engine part"""
+    for workunit in engine_workunits:
+      duration = workunit['end_timestamp'] - workunit['start_timestamp']
+
+      span = zipkin_span(
+        service_name="pants",
+        span_name=workunit['name'],
+        duration=duration,
+        span_storage=self.span_storage,
+      )
+      span.zipkin_attrs = ZipkinAttrs(
+        trace_id=self.trace_id,
+        span_id=workunit['span_id'],
+        # TODO change it when we properly pass parent_id to the v2 engine Nodes
+        # TODO Pass parent_id with ExecutionRequest when v2 engine is called by a workunit
+        # TODO pass parent_id when v2 engine Node is called by another v2 engine Node
+        parent_span_id=workunit.get("parent_id", self.parent_id),
+        flags='0', # flags: stores flags header. Currently unused
+        is_sampled=True,
+      )
+      span.start()
+      span.start_timestamp = workunit['start_timestamp']
+      span.stop()

src/rust/engine/Cargo.toml

@@ -95,7 +95,14 @@ reqwest = { version = "0.9.10", default_features = false, features = ["rustls-tl
 resettable = { path = "resettable" }
 smallvec = "0.6"
 tokio = "0.1"
+tokio-trace = { git = "https://github.com/tokio-rs/tokio", rev = "92d51202ef5e6b99004d77300de1aba0f8835513" }
+tokio-trace-log = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
+tokio-trace-futures = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
+tokio-trace-macros = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
 tempfile = "3"
 ui = { path = "ui" }
 url = "1.7.1"
 tar_api = { path = "tar_api" }
+
+[patch.crates-io]
+tokio-trace = { git = "https://github.com/tokio-rs/tokio" }

src/rust/engine/fs/Cargo.toml

@@ -23,14 +23,17 @@ indexmap = "1.0.2"
 itertools = "0.7.2"
 lazy_static = "1"
 lmdb = { git = "https://github.com/pantsbuild/lmdb-rs.git", rev = "06bdfbfc6348f6804127176e561843f214fc17f8" }
-log = "0.4"
 parking_lot = "0.6"
 protobuf = { version = "2.0.4", features = ["with-bytes"] }
 serverset = { path = "../serverset" }
 sha2 = "0.8"
 serde = "1.0"
 serde_derive = "1.0"
 tempfile = "3"
+tokio-trace = { git = "https://github.com/tokio-rs/tokio", rev = "92d51202ef5e6b99004d77300de1aba0f8835513" }
+tokio-trace-log = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
+tokio-trace-futures = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
+tokio-trace-macros = { git = "https://github.com/tokio-rs/tokio-trace-nursery", rev = "ee38e0030e3b249b09f0432f4d9140d15048a926" }
 uuid = { version = "0.7.1", features = ["v4"] }
 
 [dev-dependencies]