Add recording of metrics for Core Tracing options

Author: hmstepanek

newrelic/common/streaming_utils.py

@@ -15,6 +15,7 @@
 import collections
 import logging
 import threading
+from newrelic.packages import objsize
 
 try:
     from newrelic.core.infinite_tracing_pb2 import AttributeValue, SpanBatch
@@ -32,6 +33,11 @@ def __init__(self, maxlen, batching=False):
         self._shutdown = False
         self._seen = 0
         self._dropped = 0
+        self._ft_seen = 0
+        self._ft_dropped = 0
+        self._bytes = 0
+        self._ft_bytes = 0
+        self._ct_bytes = 0
         self._settings = None
 
         self.batching = batching
@@ -45,12 +51,30 @@ def shutdown(self):
             self._shutdown = True
             self._notify.notify_all()
 
+    def is_ft(self, sample, entity_relationship_attrs):
+        # It's a FT span if it's an exit or entry span.
+        return entity_relationship_attrs or not sample.intrinsics.get("parentId")
+
     def put(self, item):
         with self._notify:
             if self._shutdown:
                 return
 
+            entity_relationship_attrs = item[1]
+            item = item[0]
             self._seen += 1
+            # The last index contains the set of entity synthesis attrs in the span.
+            self._bytes += objsize.get_deep_size([item.intrinsics, item.user_attributes, item.agent_attributes])
+            if self.is_ft(item, entity_relationship_attrs):
+                self._ft_seen += 1
+                # The last index contains the set of entity synthesis attrs in the span.
+                self._ft_bytes += objsize.get_deep_size([item.intrinsics, item.user_attributes, item.agent_attributes])
+
+                i_ct_attrs = {"type", "name", "guid", "parentId", "transaction.name", "traceId", "nr.entryPoint", "transactionId"}
+                i_attrs = {attr: value for attr, value in item.intrinsics.items() if attr in i_ct_attrs}
+                u_attrs = {}
+                a_attrs = {attr: value for attr, value in item.agent_attributes.items() if attr in entity_relationship_attrs}
+                self._ct_bytes += objsize.get_deep_size([i_attrs, u_attrs, a_attrs])
 
             # NOTE: dropped can be over-counted as the queue approaches
             # capacity while data is still being transmitted.
@@ -59,16 +83,23 @@ def put(self, item):
             # being measured.
             if len(self._queue) >= self._queue.maxlen:
                 self._dropped += 1
+                if self.is_ft(item, entity_relationship_attrs):
+                    self._ft_dropped += 1
 
+            # Drop last index that contains the entity relationship attrs present on the span.
             self._queue.append(item)
             self._notify.notify_all()
 
     def stats(self):
         with self._notify:
             seen, dropped = self._seen, self._dropped
             self._seen, self._dropped = 0, 0
+            ft_seen, ft_dropped = self._ft_seen, self._ft_dropped
+            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
 
-        return seen, dropped
+        return seen, dropped, ft_seen, ft_dropped, _bytes, ft_bytes, ct_bytes
 
     def __bool__(self):
         return bool(self._queue)

newrelic/core/application.py

@@ -1353,6 +1353,7 @@ def harvest(self, shutdown=False, flexible=False):
 
                     # Send span events
 
+                    print(f"\n\r\n\r\n\r{configuration.span_events.enabled} {configuration.collect_span_events} {configuration.distributed_tracing.enabled} {configuration.infinite_tracing.enabled} \n\r\n\r\n\r")
                     if (
                         configuration.span_events.enabled
                         and configuration.collect_span_events
@@ -1362,11 +1363,18 @@ def harvest(self, shutdown=False, flexible=False):
                             span_stream = stats.span_stream
                             # Only merge stats as part of default harvest
                             if span_stream is not None and not flexible:
-                                spans_seen, spans_dropped = span_stream.stats()
+                                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_sent = spans_seen - spans_dropped
+                                spans_ft_sent = spans_ft_seen - spans_ft_dropped
 
                                 internal_count_metric("Supportability/InfiniteTracing/Span/Seen", spans_seen)
                                 internal_count_metric("Supportability/InfiniteTracing/Span/Sent", spans_sent)
+                                internal_count_metric("Supportability/FullTracing/Span/Seen", spans_ft_seen)
+                                internal_count_metric("Supportability/FullTracing/Span/Sent", spans_ft_sent)
+                                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)
                         else:
                             spans = stats.span_events
                             if spans:
@@ -1383,6 +1391,11 @@ def harvest(self, shutdown=False, flexible=False):
                                 spans_sampled = spans.num_samples
                                 internal_count_metric("Supportability/SpanEvent/TotalEventsSeen", spans_seen)
                                 internal_count_metric("Supportability/SpanEvent/TotalEventsSent", spans_sampled)
+                                internal_count_metric("Supportability/SpanEvent/FullTracing/TotalEventsSeen", spans.ft_seen)
+                                internal_count_metric("Supportability/SpanEvent/FullTracing/TotalEventsSent", spans.ft_sent)
+                                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)
 
                                 stats.reset_span_events()
 

newrelic/core/node_mixin.py

@@ -78,7 +78,11 @@ def span_event(self, settings, base_attrs=None, parent_guid=None, attr_class=dic
             exit_span_attrs_present = set_inprocess_attrs & set_a_attrs
             if not exit_span_attrs_present:
                 return None
-        return [i_attrs, u_attrs, a_attrs]
+
+        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):
         if settings.core_tracing.drop_inprocess_spans:
@@ -170,6 +174,11 @@ def span_event(self, *args, **kwargs):
             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
+            attrs[3] = exit_span_attrs_present
             if not exit_span_attrs_present:
                 return None
+        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
+        attrs[3] = exit_span_attrs_present
         return attrs

newrelic/core/stats_engine.py

@@ -31,6 +31,7 @@
 import zlib
 from heapq import heapify, heapreplace
 
+from newrelic.packages import objsize
 from newrelic.api.settings import STRIP_EXCEPTION_MESSAGE
 from newrelic.api.time_trace import get_linking_metadata
 from newrelic.common.encoding_utils import json_encode
@@ -446,13 +447,17 @@ def merge(self, other_data_set, priority=None):
         self.num_seen += other_data_set.num_seen - other_data_set.num_samples
 
 
-class SpanSampledDataSet:
+class SpanSampledDataSet(SampledDataSet):
     def __init__(self, capacity=100):
         self.pq = []
         self.heap = False
         self.capacity = capacity
         self.num_seen = 0
-        self.ft_num_seen = 0
+        self.ft_seen = 0
+        self.ft_sent = 0
+        self.bytes = 0
+        self.ft_bytes = 0
+        self.ct_bytes = 0
 
         if capacity <= 0:
 
@@ -465,29 +470,46 @@ def add(*args, **kwargs):
     def ft_samples(self):
         return (x[-1] for x in self.pq if is_ft(x[-1]))
 
-    def is_ft(sample):
+    def is_ft(self, sample):
         # It's a FT span if it's an exit or entry span.
         return (len(sample) > 3 and sample[3]) or not sample[0].get("parentId")
 
     def add(self, sample, priority=None):
         self.num_seen += 1
-        if is_ft(sample):
-            self.ft_num_seen += 1
+        self.bytes += objsize.get_deep_size(sample[:-1])
+        is_ft = self.is_ft(sample)
+        entity_relationship_attrs = sample[3] if len(sample) > 3 else {}
+        if is_ft:
+            self.ft_seen += 1
+            # The last index contains the set of entity synthesis attrs in the span.
+            self.ft_bytes += objsize.get_deep_size([sample[0], sample[1], sample[2]])
+
+            i_ct_attrs = {"type", "name", "guid", "parentId", "transaction.name", "traceId", "nr.entryPoint", "transactionId"}
+            i_attrs = {attr: value for attr, value in sample[0].items() if attr in sample[0]}
+            u_attrs = {}
+            a_attrs = {attr: value for attr, value in sample[2].items() if attr in entity_relationship_attrs}
+            self.ct_bytes += objsize.get_deep_size([i_attrs, u_attrs, a_attrs])
 
         if priority is None:
             priority = random.random()  # noqa: S311
 
-        entry = (priority, self.num_seen, sample)
+        entry = (priority, self.num_seen, sample[:-1])
         if self.num_seen == self.capacity:
             self.pq.append(entry)
             self.heap = self.heap or heapify(self.pq) or True
+            if is_ft:
+                self.ft_sent += 1
         elif not self.heap:
             self.pq.append(entry)
+            if is_ft:
+                self.ft_sent += 1
         else:
             sampled = self.should_sample(priority)
             if not sampled:
                 return
             heapreplace(self.pq, entry)
+            if is_ft:
+                self.ft_sent -= 1
 
 
 class LimitedDataSet(list):
@@ -573,7 +595,7 @@ def __init__(self):
         self._error_events = SampledDataSet()
         self._custom_events = SampledDataSet()
         self._ml_events = SampledDataSet()
-        self._span_events = SampledDataSet()
+        self._span_events = SpanSampledDataSet()
         self._log_events = SampledDataSet()
         self._span_stream = None
         self.__sql_stats_table = {}
@@ -1785,9 +1807,9 @@ def reset_ml_events(self):
 
     def reset_span_events(self):
         if self.__settings is not None:
-            self._span_events = SampledDataSet(self.__settings.event_harvest_config.harvest_limits.span_event_data)
+            self._span_events = SpanSampledDataSet(self.__settings.event_harvest_config.harvest_limits.span_event_data)
         else:
-            self._span_events = SampledDataSet()
+            self._span_events = SpanSampledDataSet()
 
     def reset_log_events(self):
         if self.__settings is not None:

newrelic/core/transaction_node.py

@@ -621,8 +621,11 @@ def _add_call_count(source, target):
         return intrinsics
 
     def span_protos(self, settings):
-        for i_attrs, u_attrs, a_attrs in self.span_events(settings, attr_class=SpanProtoAttrs):
-            yield Span(trace_id=self.trace_id, intrinsics=i_attrs, user_attributes=u_attrs, agent_attributes=a_attrs)
+        for span in self.span_events(settings, attr_class=SpanProtoAttrs):
+            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):
         base_attrs = attr_class(

newrelic/packages/objsize/__init__.py

@@ -5,7 +5,7 @@
 import warnings
 from typing import Any, Iterable, Iterator, Optional
 
-from objsize.traverse import (
+from newrelic.packages.objsize.traverse import (
     FilterFunc,
     GetReferentsFunc,
     MarkedSet,