[BACKPORT] Send TCP metrics for bytes sent and received to server

hazelcast/reactor.py

@@ -344,6 +344,8 @@ def __init__(self):
  loop.shutdown()
  loop = _BasicLoop(self.map)
  self._loop = loop
+ self.bytes_sent = 0
+ self.bytes_received = 0
 
  def start(self):
  self._loop.start()
@@ -421,9 +423,11 @@ def handle_read(self):
  try:
  while True:
  data = self.recv(receive_buffer_size)
+ bytes_received = len(data)
+ self._reactor.bytes_received += bytes_received
  reader.read(data)
  self.last_read_time = time.time()
- if len(data) < receive_buffer_size:
+ if bytes_received < receive_buffer_size:
  break
  except socket.error as err:
  if err.args[0] not in _RETRYABLE_ERROR_CODES:
@@ -461,7 +465,7 @@ def handle_write(self):
  buf.truncate(0)
 
  try:
- sent = self.send(bytes_)
+ bytes_sent = self.send(bytes_)
  except socket.error as err:
  if err.args[0] in _RETRYABLE_ERROR_CODES:
  # Couldn't write the bytes but we should
@@ -474,9 +478,9 @@ def handle_write(self):
  # No exception is thrown during the send
  self.last_write_time = time.time()
  self.sent_protocol_bytes = True
-
- if sent < len(bytes_):
- write_queue.appendleft(bytes_[sent:])
+ self._reactor.bytes_sent += bytes_sent
+ if bytes_sent < len(bytes_):
+ write_queue.appendleft(bytes_[bytes_sent:])
 
  def handle_close(self):
  _logger.warning("Connection closed by server")

hazelcast/statistics.py

@@ -25,6 +25,8 @@
 _NEAR_CACHE_DESCRIPTOR_PREFIX = "nearcache"
 _NEAR_CACHE_DESCRIPTOR_DISCRIMINATOR = "name"
 
+_TCP_METRICS_PREFIX = "tcp"
+
 
 class Statistics(object):
  def __init__(
@@ -130,7 +132,7 @@ def _register_system_gauge(self, gauge_name, gauge_fn, value_type=ValueType.LONG
  self._registered_system_gauges[gauge_name] = (gauge_fn, value_type)
  except Exception as e:
  _logger.debug(
- "Unable to register the system related gauge %s. Error: %s" % (gauge_name, e)
+ "Unable to register the system related gauge %s. Error: %s", gauge_name, e
  )
 
  def _register_process_gauge(self, gauge_name, gauge_fn, value_type=ValueType.LONG):
@@ -141,7 +143,7 @@ def _register_process_gauge(self, gauge_name, gauge_fn, value_type=ValueType.LON
  self._registered_process_gauges[gauge_name] = (gauge_fn, value_type)
  except Exception as e:
  _logger.debug(
- "Unable to register the process related gauge %s. Error: %s" % (gauge_name, e)
+ "Unable to register the process related gauge %s. Error: %s", gauge_name, e
  )
 
  def _collect_and_send_stats(self):
@@ -157,6 +159,7 @@ def _collect_and_send_stats(self):
  self._add_client_attributes(attributes, connection)
  self._add_near_cache_metrics(attributes, compressor)
  self._add_system_and_process_metrics(attributes, compressor)
+ self._add_tcp_metrics(compressor)
  self._send_stats(
  collection_timestamp, "".join(attributes), compressor.generate_blob(), connection
  )
@@ -180,7 +183,7 @@ def _add_system_and_process_metrics(self, attributes, compressor):
  attributes, compressor, gauge_name, value, value_type
  )
  except:
- _logger.exception("Error while collecting '%s'." % gauge_name)
+ _logger.exception("Error while collecting '%s'.", gauge_name)
 
  if not self._registered_process_gauges:
  # Do not create the process object if no process-related
@@ -195,7 +198,7 @@ def _add_system_and_process_metrics(self, attributes, compressor):
  attributes, compressor, gauge_name, value, value_type
  )
  except:
- _logger.exception("Error while collecting '%s'." % gauge_name)
+ _logger.exception("Error while collecting '%s'.", gauge_name)
 
  def _add_system_or_process_metric(self, attributes, compressor, gauge_name, value, value_type):
  # We don't have any metrics that do not have prefix.
@@ -342,9 +345,26 @@ def _add_near_cache_metric(
  self._add_attribute(attributes, metric, value, nc_name_with_prefix)
  except:
  _logger.exception(
- "Error while collecting %s metric for near cache '%s'" % (metric, nc_name)
+ "Error while collecting %s metric for near cache '%s'.", metric, nc_name
  )
 
+ def _add_tcp_metrics(self, compressor):
+ self._add_tcp_metric(compressor, "bytesSend", self._reactor.bytes_sent)
+ self._add_tcp_metric(compressor, "bytesReceived", self._reactor.bytes_received)
+
+ def _add_tcp_metric(
+ self, compressor, metric, value, value_type=ValueType.LONG, unit=ProbeUnit.BYTES
+ ):
+ descriptor = MetricDescriptor(
+ metric=metric,
+ prefix=_TCP_METRICS_PREFIX,
+ unit=unit,
+ )
+ try:
+ self._add_metric(compressor, descriptor, value, value_type)
+ except:
+ _logger.exception("Error while collecting '%s.%s'.", _TCP_METRICS_PREFIX, metric)
+
  def _add_metric(self, compressor, descriptor, value, value_type):
  if value_type == ValueType.LONG:
  compressor.add_long(descriptor, value)

tests/integration/backward_compatible/client_test.py

@@ -1,10 +1,11 @@
 import time
+import unittest
 
-from tests.base import HazelcastTestCase
+from tests.base import HazelcastTestCase, SingleMemberTestCase
 from hazelcast.client import HazelcastClient
 from hazelcast.lifecycle import LifecycleState
 from tests.hzrc.ttypes import Lang
-from tests.util import get_current_timestamp
+from tests.util import get_current_timestamp, random_string
 
 
 class ClientTest(HazelcastTestCase):
@@ -110,3 +111,32 @@ def get_labels_from_member(self, client_uuid):
  client_uuid
  )
  return self.rc.executeOnController(self.cluster.id, script, Lang.JAVASCRIPT).result
+
+
+class ClientTcpMetricsTest(SingleMemberTestCase):
+ @classmethod
+ def configure_client(cls, config):
+ config["cluster_name"] = cls.cluster.id
+ return config
+
+ def test_bytes_received(self):
+ reactor = self.client._reactor
+
+ bytes_received = reactor.bytes_received
+ self.assertGreater(bytes_received, 0)
+
+ m = self.client.get_map(random_string()).blocking()
+ m.get(random_string())
+
+ self.assertGreater(reactor.bytes_received, bytes_received)
+
+ def test_bytes_sent(self):
+ reactor = self.client._reactor
+
+ bytes_sent = reactor.bytes_sent
+ self.assertGreater(bytes_sent, 0)
+
+ m = self.client.get_map(random_string()).blocking()
+ m.set(random_string(), random_string())
+
+ self.assertGreater(reactor.bytes_sent, bytes_sent)