Retrieves the actual tracing details from Cassandra and populates the attributes of this instance. Because tracing details are stored asynchronously by Cassandra, this may need to retry the session detail fetch. If the trace is still not available after `max_wait`
(self, max_wait=2.0, wait_for_complete=True, query_cl=None)
| 932 | self._session = session |
| 933 | |
| 934 | def populate(self, max_wait=2.0, wait_for_complete=True, query_cl=None): |
| 935 | """ |
| 936 | Retrieves the actual tracing details from Cassandra and populates the |
| 937 | attributes of this instance. Because tracing details are stored |
| 938 | asynchronously by Cassandra, this may need to retry the session |
| 939 | detail fetch. If the trace is still not available after `max_wait` |
| 940 | seconds, :exc:`.TraceUnavailable` will be raised; if `max_wait` is |
| 941 | :const:`None`, this will retry forever. |
| 942 | |
| 943 | `wait_for_complete=False` bypasses the wait for duration to be populated. |
| 944 | This can be used to query events from partial sessions. |
| 945 | |
| 946 | `query_cl` specifies a consistency level to use for polling the trace tables, |
| 947 | if different from the session default. |
| 948 | """ |
| 949 | attempt = 0 |
| 950 | start = time.time() |
| 951 | while True: |
| 952 | time_spent = time.time() - start |
| 953 | if max_wait is not None and time_spent >= max_wait: |
| 954 | raise TraceUnavailable( |
| 955 | "Trace information was not available within %f seconds. Consider raising Session.max_trace_wait." % (max_wait,)) |
| 956 | |
| 957 | log.debug("Attempting to fetch trace info for trace ID: %s", self.trace_id) |
| 958 | session_results = self._execute( |
| 959 | SimpleStatement(self._SELECT_SESSIONS_FORMAT, consistency_level=query_cl), (self.trace_id,), time_spent, max_wait) |
| 960 | |
| 961 | # PYTHON-730: There is race condition that the duration mutation is written before started_at the for fast queries |
| 962 | session_row = session_results.one() if session_results else None |
| 963 | is_complete = session_row is not None and session_row.duration is not None and session_row.started_at is not None |
| 964 | if not session_results or (wait_for_complete and not is_complete): |
| 965 | time.sleep(self._BASE_RETRY_SLEEP * (2 ** attempt)) |
| 966 | attempt += 1 |
| 967 | continue |
| 968 | if is_complete: |
| 969 | log.debug("Fetched trace info for trace ID: %s", self.trace_id) |
| 970 | else: |
| 971 | log.debug("Fetching partial trace info for trace ID: %s", self.trace_id) |
| 972 | |
| 973 | self.request_type = session_row.request |
| 974 | self.duration = timedelta(microseconds=session_row.duration) if is_complete else None |
| 975 | self.started_at = session_row.started_at |
| 976 | self.coordinator = session_row.coordinator |
| 977 | self.parameters = session_row.parameters |
| 978 | # since C* 2.2 |
| 979 | self.client = getattr(session_row, 'client', None) |
| 980 | |
| 981 | log.debug("Attempting to fetch trace events for trace ID: %s", self.trace_id) |
| 982 | time_spent = time.time() - start |
| 983 | event_results = self._execute( |
| 984 | SimpleStatement(self._SELECT_EVENTS_FORMAT, consistency_level=query_cl), (self.trace_id,), time_spent, max_wait) |
| 985 | log.debug("Fetched trace events for trace ID: %s", self.trace_id) |
| 986 | self.events = tuple(TraceEvent(r.activity, r.event_id, r.source, r.source_elapsed, r.thread) |
| 987 | for r in event_results) |
| 988 | break |
| 989 | |
| 990 | def _execute(self, query, parameters, time_spent, max_wait): |
| 991 | timeout = (max_wait - time_spent) if max_wait is not None else None |