fix the duplicated service graphs issue

samcli/lib/observability/xray_traces/xray_event_mappers.py

@@ -70,7 +70,10 @@ class XRayServiceGraphConsoleMapper(ObservabilityEventMapper[XRayServiceGraphEve
 
     def map(self, event: XRayServiceGraphEvent) -> XRayServiceGraphEvent:
         formatted_services = self.format_services(event.services)
-        mapped_message = f"\nNew XRay Service Graph" f"{formatted_services}"
+        mapped_message = "\nNew XRay Service Graph"
+        mapped_message += f"\n  Start time: {event.start_time}"
+        mapped_message += f"\n  End time: {event.end_time}"
+        mapped_message += formatted_services
         event.message = mapped_message
 
         return event

samcli/lib/observability/xray_traces/xray_events.py

@@ -1,6 +1,7 @@
 """
 Keeps XRay event definitions
 """
+import hashlib
 import json
 from typing import List
 
@@ -89,13 +90,22 @@ def __init__(self, event: dict):
         self.services: List[XRayGraphServiceInfo] = []
         self.message = str(event)
         self._construct_service(event)
+        self.start_time = event.get("StartTime", None)
+        self.end_time = event.get("EndTime", None)
         super().__init__(event, 0)
 
     def _construct_service(self, event_dict):
         services = event_dict.get("Services", [])
         for service in services:
             self.services.append(XRayGraphServiceInfo(service))
 
+    def get_hash(self):
+        """
+        get the hash of the containing services
+        """
+        services = self.event.get("Services", [])
+        return hashlib.sha1(str(services).encode('utf-8')).hexdigest()
+
 
 class XRayGraphServiceInfo:
     """

samcli/lib/observability/xray_traces/xray_service_graph_event_puller.py

@@ -45,7 +45,7 @@ def __init__(
         self._max_retries = max_retries
         self._poll_interval = poll_interval
         self._had_data = False
-        self._previous_trace_ids: Set[str] = set()
+        self._previous_xray_service_graphs: Set[str] = set()
 
     def tail(self, start_time: Optional[datetime] = None, filter_pattern: Optional[str] = None):
         if start_time:
@@ -101,14 +101,16 @@ def load_time_period(
                 # update latest fetched event
                 event_end_time = result.get("EndTime", None)
                 if event_end_time:
-                    # end_time is in local time zone, need to convert to utc first
                     utc_end_time = to_utc(event_end_time)
                     latest_event_time = utc_to_timestamp(utc_end_time)
                     if latest_event_time > self.latest_event_time:
                         self.latest_event_time = latest_event_time + 1
+
                 self._had_data = True
                 xray_service_graph_event = XRayServiceGraphEvent(result)
-                self.consumer.consume(xray_service_graph_event)
+                if xray_service_graph_event.get_hash() not in self._previous_xray_service_graphs:
+                    self.consumer.consume(xray_service_graph_event)
+                self._previous_xray_service_graphs.add(xray_service_graph_event.get_hash())
 
     def load_events(self, event_ids: List[str]):
         LOG.debug("Loading specific service graph events are not supported via XRay Service Graph")