Add support for Core Tracing FT and 2B options

Author: hmstepanek

newrelic/api/transaction.py

@@ -658,6 +658,13 @@ def __exit__(self, exc, value, tb):
     def sampled(self):
         return self._sampled
 
+    @property
+    def ct_sampled(self):
+        # If DT doesn't sample it CT will.
+        if not self.sampled and settings.core_tracing.enabled:
+            return True
+        return False
+
     @property
     def priority(self):
         return self._priority
@@ -1082,12 +1089,13 @@ def _create_distributed_trace_data(self):
             return
 
         self._compute_sampled_and_priority()
+        sampled = self.sampled or self.ct_sampled
         data = {
             "ty": "App",
             "ac": account_id,
             "ap": application_id,
             "tr": self.trace_id,
-            "sa": self.sampled,
+            "sa": sampled,
             "pr": self.priority,
             "tx": self.guid,
             "ti": int(time.time() * 1000.0),

newrelic/common/streaming_utils.py

@@ -38,6 +38,7 @@ def __init__(self, maxlen, batching=False):
         self._bytes = 0
         self._ft_bytes = 0
         self._ct_bytes = 0
+        self.ct_processing_time = 0
         self._settings = None
 
         self.batching = batching
@@ -98,8 +99,10 @@ def stats(self):
             self._ft_seen, self._ft_dropped = 0, 0
             _bytes, ft_bytes, ct_bytes = self._bytes, self._ft_bytes, self._ct_bytes
             self._bytes, self._ft_bytes, self._ct_bytes = 0, 0, 0
+            ct_processing_time = self._ct_processing_time
+            self._ct_processing_time = 0
 
-        return seen, dropped, ft_seen, ft_dropped, _bytes, ft_bytes, ct_bytes
+        return seen, dropped, ft_seen, ft_dropped, _bytes, ft_bytes, ct_byte, ct_processing_times
 
     def __bool__(self):
         return bool(self._queue)

newrelic/core/application.py

@@ -1364,7 +1364,7 @@ def harvest(self, shutdown=False, flexible=False):
                             # Only merge stats as part of default harvest
                             if span_stream is not None and not flexible:
                                 print("\n\r\n\r\n\rSending Supportability metrics for inifinte tracing\n\r\n\r\n\r")
-                                spans_seen, spans_dropped, spans_ft_seen, spans_ft_dropped, _bytes, ft_bytes, ct_bytes = span_stream.stats()
+                                spans_seen, spans_dropped, spans_ft_seen, spans_ft_dropped, _bytes, ft_bytes, ct_bytes, ct_processing_time = span_stream.stats()
                                 spans_sent = spans_seen - spans_dropped
                                 spans_ft_sent = spans_ft_seen - spans_ft_dropped
 
@@ -1375,6 +1375,7 @@ def harvest(self, shutdown=False, flexible=False):
                                 internal_count_metric("Supportability/InfiniteTracing/Bytes/Seen", _bytes)
                                 internal_count_metric("Supportability/FullTracing/Bytes/Seen", ft_bytes)
                                 internal_count_metric("Supportability/CoreTracing/Bytes/Seen", ct_bytes)
+                                internal_count_metric("Supportability/CoreTracing/TotalTime", ct_processing_time*1000)  # Time in ms.
                         else:
                             spans = stats.span_events
                             if spans:
@@ -1396,6 +1397,7 @@ def harvest(self, shutdown=False, flexible=False):
                                 internal_count_metric("Supportability/DistributedTracing/Bytes/Seen", spans.bytes)
                                 internal_count_metric("Supportability/FullTracing/Bytes/Seen", spans.ft_bytes)
                                 internal_count_metric("Supportability/CoreTracing/Bytes/Seen", spans.ct_bytes)
+                                internal_count_metric("Supportability/SpanEvent/TotalCoreTracingTime", spans.ct_processing_time*1000)  # Time in ms.
 
                                 stats.reset_span_events()
 

newrelic/core/node_mixin.py

@@ -70,34 +70,69 @@ def span_event(self, settings, base_attrs=None, parent_guid=None, attr_class=dic
         )
 
         # intrinsics, user attrs, agent attrs
-        if settings.core_tracing.drop_inprocess_spans:
+        if settings.core_tracing.drop_inprocess_spans or settings.core_tracing.enabled:
             if not parent_guid:
                 return [i_attrs, u_attrs, a_attrs]
-            set_inprocess_attrs = set(attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES)
+            set_inprocess_attrs = attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES
             set_a_attrs = set(a_attrs)
             exit_span_attrs_present = set_inprocess_attrs & set_a_attrs
             if not exit_span_attrs_present:
                 return None
+        if settings.core_tracing.enabled:
+            a_attrs["ids"] = []  # This is the list of span guids that share this unqiue exit span.
 
         set_inprocess_attrs = set(attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES)
         set_a_attrs = set(a_attrs)
         exit_span_attrs_present = set_inprocess_attrs & set_a_attrs
         return [i_attrs, u_attrs, a_attrs, exit_span_attrs_present]
 
-    def span_events(self, settings, base_attrs=None, parent_guid=None, attr_class=dict):
+    def span_events(self, settings, base_attrs=None, parent_guid=None, attr_class=dict, ct_exit_spans=None, ct_processing_time=0):
         if settings.core_tracing.drop_inprocess_spans:
+            ct_exit_spans = ct_exit_spans or {}
             span = self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class)
+            start_time = time.time()
             parent_id = parent_guid
             if span:  # span will be None if the span is an inprocess span.
                 yield span
                 parent_id = self.guid
-
             for child in self.children:
                 for event in child.span_events(  # noqa: UP028
-                    settings, base_attrs=base_attrs, parent_guid=parent_id, attr_class=attr_class
+                    settings, base_attrs=base_attrs, parent_guid=parent_id, attr_class=attr_class, ct_exit_spans=ct_exit_spans, ct_processing_time=ct_processing_time
                 ):
                     if event:
                         yield event
+        elif settings.core_tracing.enabled:
+            ct_exit_spans = ct_exit_spans or {}
+            span = self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class)
+            start_time = time.time()
+            parent_id = parent_guid
+            if span:  # span will be None if the span is an inprocess span.
+                span_attrs = "".join([value for key, value in span[2].items() if key in attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES])
+                new_exit_span = span_attrs not in ct_exit_spans
+                if parent_id:  # If this is not the entry span.
+                    if new_exit_span:
+                        ct_exit_spans[span_attrs] = span
+                    else:
+                        ct_exit_spans[span_attrs][2]["ids"] = event[2]["guid"]
+                ct_processing_time += (time.time() - start_time)
+                if not parent_id or new_exit_span:
+                    yield span
+                parent_id = self.guid
+            for child in self.children:
+                for event in child.span_events(  # noqa: UP028
+                    settings, base_attrs=base_attrs, parent_guid=parent_id, attr_class=attr_class, ct_exit_spans=ct_exit_spans, ct_processing_time=ct_processing_time
+                ):
+                    if event:
+                        start_time = time.time()
+                        span_attrs = "".join([value for key, value in event[2].items() if key in attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES])
+                        new_exit_span = span_attrs not in ct_exit_spans
+                        if new_exit_span:
+                            ct_exit_spans[span_attrs] = event
+                        else:
+                            ct_exit_spans[span_attrs][2]["ids"] = event[2]["guid"]
+                        ct_processing_time += (time.time() - start_time)
+                        if new_exit_span:
+                            yield event
         else:
             yield self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class)
 
@@ -168,7 +203,7 @@ def span_event(self, *args, **kwargs):
             pass
 
         # intrinsics, user attrs, agent attrs
-        if settings.core_tracing.drop_inprocess_spans:
+        if settings.core_tracing.drop_inprocess_spans or settings.core_tracing.enabled:
             if not parent_guid:
                 return [i_attrs, u_attrs, a_attrs]
             set_inprocess_attrs = set(attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES)

newrelic/core/stats_engine.py

@@ -458,6 +458,7 @@ def __init__(self, capacity=100):
         self.bytes = 0
         self.ft_bytes = 0
         self.ct_bytes = 0
+        self.ct_processing_time = 0
 
         if capacity <= 0:
 
@@ -1262,11 +1263,16 @@ def record_transaction(self, transaction):
 
         if settings.distributed_tracing.enabled and settings.span_events.enabled and settings.collect_span_events:
             if settings.infinite_tracing.enabled:
-                for event in transaction.span_protos(settings):
+                ct_processing_time = 0
+                for event in transaction.span_protos(settings, ct_processing_time=ct_processing_time):
                     self._span_stream.put(event)
+                self._span_stream._ct_processing_time += ct_processing_time
             elif transaction.sampled:
-                for event in transaction.span_events(self.__settings):
+                ct_processing_time = 0
+                for event in transaction.span_events(self.__settings, ct_processing_time=ct_processing_time):
                     self._span_events.add(event, priority=transaction.priority)
+                self._span_events.ct_processing_time += ct_processing_time
+
 
         # Merge in log events
 

newrelic/core/transaction_node.py

@@ -620,14 +620,14 @@ def _add_call_count(source, target):
 
         return intrinsics
 
-    def span_protos(self, settings):
-        for span in self.span_events(settings, attr_class=SpanProtoAttrs):
+    def span_protos(self, settings, ct_process_time=0):
+        for span in self.span_events(settings, attr_class=SpanProtoAttrs, ct_process_time=ct_process_time):
             if len(span) > 3:
                 yield Span(trace_id=self.trace_id, intrinsics=span[0], user_attributes=span[1], agent_attributes=span[2]), span[3]
             else:
                 yield Span(trace_id=self.trace_id, intrinsics=span[0], user_attributes=span[1], agent_attributes=span[2])
 
-    def span_events(self, settings, attr_class=dict):
+    def span_events(self, settings, attr_class=dict, ct_processing_time=0):
         base_attrs = attr_class(
             (
                 ("transactionId", self.guid),
@@ -637,4 +637,4 @@ def span_events(self, settings, attr_class=dict):
             )
         )
 
-        yield from self.root.span_events(settings, base_attrs, parent_guid=self.parent_span, attr_class=attr_class)
+        yield from self.root.span_events(settings, base_attrs, parent_guid=self.parent_span, attr_class=attr_class, ct_processing_time=ct_processing_time)