Fixup timing

Author: hmstepanek

newrelic/common/streaming_utils.py

@@ -63,12 +63,9 @@ def put(self, item):
             # being measured.
             if len(self._queue) >= self._queue.maxlen:
                 self._dropped += 1
-                print(f"dropped: {self._dropped}")
 
             self._queue.append(item)
             self._notify.notify_all()
-            print(f"seen: {self._seen}")
-            print(f"bytes: {self._bytes}")
 
     def stats(self):
         with self._notify:

newrelic/core/application.py

@@ -1384,9 +1384,13 @@ def harvest(self, shutdown=False, flexible=False):
                                 spans_seen = spans.num_seen
                                 spans_sampled = spans.num_samples
                                 internal_count_metric("Supportability/SpanEvent/TotalEventsSeen", spans_seen)
+                                print(f"Supportability/SpanEvent/TotalEventsSeen: {spans_seen}")
                                 internal_count_metric("Supportability/SpanEvent/TotalEventsSent", spans_sampled)
+                                print(f"Supportability/SpanEvent/TotalEventsSent: {spans_sampled}")
                                 internal_count_metric("Supportability/DistributedTracing/Bytes/Seen", spans.bytes)
+                                print(f"Supportability/DistributedTracing/Bytes/Seen: {spans.bytes}")
                                 internal_count_metric("Supportability/SpanEvent/TotalCoreTracingTime", spans.ct_processing_time*1000)  # Time in ms.
+                                print(f"Supportability/SpanEvent/TotalCoreTracingTime: {spans.ct_processing_time*1000}")  # Time in ms.
 
                                 stats.reset_span_events()
 

newrelic/core/node_mixin.py

@@ -62,9 +62,10 @@ def span_event(self, settings, base_attrs=None, parent_guid=None, attr_class=dic
         i_attrs["timestamp"] = int(self.start_time * 1000)
         i_attrs["duration"] = self.duration
         i_attrs["category"] = "generic"
+        # TODO: limit intrinsic attributes but this likely requires changes in the pipeline.
         #if settings.core_tracing.enabled:
-            #i_ct_attrs = {"type", "name", "guid", "parentId", "transaction.name", "traceId", "nr.entryPoint", "transactionId"}
-            #i_attrs = {key: value for key, value in i_attrs.items() if key in i_ct_attrs}
+        #    i_ct_attrs = {"type", "name", "guid", "parentId", "transaction.name", "traceId", "timestamp", "duration", "nr.entryPoint", "transactionId"}
+        #    i_attrs = {key: value for key, value in i_attrs.items() if key in i_ct_attrs}
 
         if parent_guid:
             i_attrs["parentId"] = parent_guid
@@ -81,32 +82,40 @@ def span_event(self, settings, base_attrs=None, parent_guid=None, attr_class=dic
             u_attrs, settings.attribute_filter, DST_SPAN_EVENTS, attr_class=attr_class
         )
 
+        start_time = time.time()
         if settings.core_tracing.drop_inprocess_spans or settings.core_tracing.enabled:
             exit_span_attrs_present = attribute.SPAN_ENTITY_RELATIONSHIP_ATTRIBUTES & set(a_attrs)
             if self.__class__.__name__ == "RootNode":
-                return [i_attrs, u_attrs, a_attrs]
+                if settings.core_tracing.enabled:
+                    ct_processing_time[0] += (time.time() - start_time)
+                    return [i_attrs, {}, {}]
+                else:
+                    ct_processing_time[0] += (time.time() - start_time)
+                    return [i_attrs, u_attrs, a_attrs]
             if not exit_span_attrs_present:
+                ct_processing_time[0] += (time.time() - start_time)
                 return None
-        if settings.core_tracing.enabled:
-            start_time = time.time()
-            # TODO: use attr value name rather than str casting.
-            span_attrs = "".join([str(a_attrs[key]) for key in exit_span_attrs_present])
-            new_exit_span = span_attrs not in ct_exit_spans
-            if new_exit_span:
-                ct_exit_spans[span_attrs] = self.ids
-            else:
-                # For now add ids to user attributes list
-                ct_exit_spans[span_attrs].append(self.guid)
-            ct_processing_time += (time.time() - start_time)
-            if new_exit_span:
-                return [i_attrs, u_attrs, a_attrs]
-            return None
+            if settings.core_tracing.enabled:
+                a_attrs = attr_class({key: a_attrs[key] for key in exit_span_attrs_present})
+                # TODO: use attr value name rather than str casting.
+                span_attrs = "".join([str(a_attrs[key]) for key in exit_span_attrs_present])
+                new_exit_span = span_attrs not in ct_exit_spans
+                if new_exit_span:
+                    ct_exit_spans[span_attrs] = self.ids
+                else:
+                    # For now add ids to user attributes list
+                    ct_exit_spans[span_attrs].append(self.guid)
+                if new_exit_span:
+                    ct_processing_time[0] += (time.time() - start_time)
+                    return [i_attrs, u_attrs, a_attrs]
+                ct_processing_time[0] += (time.time() - start_time)
+                return None
+        ct_processing_time[0] += (time.time() - start_time)
         return [i_attrs, u_attrs, a_attrs]
 
-    def span_events(self, settings, base_attrs=None, parent_guid=None, attr_class=dict, ct_exit_spans=None, ct_processing_time=0):
+    def span_events(self, settings, base_attrs=None, parent_guid=None, attr_class=dict, ct_exit_spans=None, ct_processing_time=None):
         if settings.core_tracing.drop_inprocess_spans or settings.core_tracing.enabled:
             span = self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class, ct_exit_spans=ct_exit_spans, ct_processing_time=ct_processing_time)
-            start_time = time.time()
             parent_id = parent_guid
             if span:  # span will be None if the span is an inprocess span or repeated exit span.
                 yield span
@@ -118,11 +127,12 @@ def span_events(self, settings, base_attrs=None, parent_guid=None, attr_class=di
                     if event:  # event will be None if the span is an inprocess span or repeated exit span.
                         yield event
         else:
-            yield self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class)
+            span = self.span_event(settings, base_attrs=base_attrs, parent_guid=parent_guid, attr_class=attr_class, ct_processing_time=ct_processing_time)
+            yield span
 
             for child in self.children:
                 for event in child.span_events(  # noqa: UP028
-                    settings, base_attrs=base_attrs, parent_guid=self.guid, attr_class=attr_class
+                    settings, base_attrs=base_attrs, parent_guid=self.guid, attr_class=attr_class, ct_processing_time=ct_processing_time
                 ):
                     yield event
 

newrelic/core/stats_engine.py

@@ -55,6 +55,26 @@
 from newrelic.core.metric import TimeMetric
 from newrelic.core.stack_trace import exception_stack
 
+def get_deep_size(obj, seen=None):
+    """Recursively calculates the size of an object including nested lists and dicts."""
+    if seen is None:
+        seen = set()
+
+    # Avoid recursion for already seen objects (handle circular references)
+    obj_id = id(obj)
+    if obj_id in seen:
+        return 0
+    seen.add(obj_id)
+
+    size = sys.getsizeof(obj)
+
+    if isinstance(obj, dict):
+        size += sum(get_deep_size(k, seen) + get_deep_size(v, seen) for k, v in obj.items())
+    elif isinstance(obj, (list, tuple, set, frozenset)):
+        size += sum(get_deep_size(i, seen) for i in obj)
+
+    return size
+
 _logger = logging.getLogger(__name__)
 
 EVENT_HARVEST_METHODS = {
@@ -449,40 +469,21 @@ def merge(self, other_data_set, priority=None):
 
 class SpanSampledDataSet(SampledDataSet):
     def __init__(self, capacity=100):
-        self.pq = []
-        self.heap = False
-        self.capacity = capacity
-        self.num_seen = 0
-        self.bytes = 0
+        super().__init__(capacity=capacity)
         self.ct_processing_time = 0
-
-        if capacity <= 0:
-
-            def add(*args, **kwargs):
-                self.num_seen += 1
-
-            self.add = add
+        self.bytes = 0
 
     def add(self, sample, priority=None):
-        self.num_seen += 1
-        self.bytes += objsize.get_deep_size(sample[:-1])
+        super().add(sample=sample, priority=priority)
+        self.bytes += get_deep_size(sample)
 
-        if priority is None:
-            priority = random.random()  # noqa: S311
+    def reset(self):
+        super().reset()
+        self.ct_processing_time = 0
 
-        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
-        elif not self.heap:
-            self.pq.append(entry)
-        else:
-            sampled = self.should_sample(priority)
-            if not sampled:
-                return
-            heapreplace(self.pq, entry)
-            if is_ft:
-                self.ft_sent -= 1
+    def merge(self, other_data_set, priority=None):
+        super().merge(other_data_set=other_data_set, priority=priority)
+        self.ct_processing_time += other_data_set.ct_processing_time
 
 
 class LimitedDataSet(list):
@@ -1235,17 +1236,15 @@ 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:
-                ct_processing_time = 0
+                ct_processing_time = [0]
                 for event in transaction.span_protos(settings, ct_processing_time=ct_processing_time):
-                    print(event.intrinsics["name"])
                     self._span_stream.put(event)
-                self._span_stream._ct_processing_time += ct_processing_time
+                self._span_stream._ct_processing_time += ct_processing_time[0]
             elif transaction.sampled:
-                ct_processing_time = 0
+                ct_processing_time = [0]
                 for event in transaction.span_events(self.__settings, ct_processing_time=ct_processing_time):
-                    print(event[0]["name"])
                     self._span_events.add(event, priority=transaction.priority)
-                self._span_events.ct_processing_time += ct_processing_time
+                self._span_events.ct_processing_time += ct_processing_time[0]
 
 
         # Merge in log events