MCPcopy Create free account
hub / github.com/apache/cassandra-python-driver / MonitorReporter

Class MonitorReporter

cassandra/datastax/insights/reporter.py:38–223  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

36
37
38class MonitorReporter(Thread):
39
40 def __init__(self, interval_sec, session):
41 """
42 takes an int indicating interval between requests, a function returning
43 the connection to be used, and the timeout per request
44 """
45 # Thread is an old-style class so we can't super()
46 Thread.__init__(self, name='monitor_reporter')
47
48 initialize_registry(insights_registry)
49
50 self._interval, self._session = interval_sec, session
51
52 self._shutdown_event = Event()
53 self.daemon = True
54 self.start()
55
56 def run(self):
57 self._send_via_rpc(self._get_startup_data())
58
59 # introduce some jitter -- send up to 1/10 of _interval early
60 self._shutdown_event.wait(self._interval * random.uniform(.9, 1))
61
62 while not self._shutdown_event.is_set():
63 start_time = time.time()
64
65 self._send_via_rpc(self._get_status_data())
66
67 elapsed = time.time() - start_time
68 self._shutdown_event.wait(max(self._interval - elapsed, 0.01))
69
70 # TODO: redundant with ConnectionHeartbeat.ShutdownException
71 class ShutDownException(Exception):
72 pass
73
74 def _send_via_rpc(self, data):
75 try:
76 self._session.execute(
77 "CALL InsightsRpc.reportInsight(%s)", (json.dumps(data),)
78 )
79 log.debug('Insights RPC data: {}'.format(data))
80 except Exception as e:
81 log.debug('Insights RPC send failed with {}'.format(e))
82 log.debug('Insights RPC data: {}'.format(data))
83
84 def _get_status_data(self):
85 cc = self._session.cluster.control_connection
86
87 connected_nodes = {
88 host.address: {
89 'connections': state['open_count'],
90 'inFlightQueries': state['in_flights']
91 }
92 for (host, state) in self._session.get_pool_state().items()
93 }
94
95 return {

Callers 1

__init__Method · 0.90

Calls

no outgoing calls

Tested by

no test coverage detected