Remove kafka metric instrumentation. (#626)

Author: umaannamalai

newrelic/hooks/messagebroker_kafkapython.py

@@ -11,24 +11,14 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import logging
-import math
 import sys
-import threading
 
-from kafka.metrics.metrics_reporter import AbstractMetricsReporter
-
-import newrelic.core.agent
 from newrelic.api.application import application_instance
 from newrelic.api.message_trace import MessageTrace
 from newrelic.api.message_transaction import MessageTransaction
 from newrelic.api.time_trace import notice_error
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import wrap_function_wrapper
-from newrelic.packages import six
-from newrelic.samplers.decorators import data_source_factory
-
-_logger = logging.getLogger(__name__)
 
 HEARTBEAT_POLL = "MessageBroker/Kafka/Heartbeat/Poll"
 HEARTBEAT_SENT = "MessageBroker/Kafka/Heartbeat/Sent"
@@ -187,141 +177,11 @@ def wrap_kafkaconsumer_next(wrapped, instance, args, kwargs):
     return record
 
 
-class KafkaMetricsDataSource(object):
-    _instance = None
-
-    def __init__(self):
-        self.reporters = []
-
-    @classmethod
-    @data_source_factory(name="Kafka Metrics Reporter")
-    def factory(cls, settings=None, environ=None):
-        return cls.singleton()
-
-    @classmethod
-    def singleton(cls, register=True):
-        # If already initialized, exit early
-        if cls._instance:
-            return cls._instance
-
-        # Init and register instance on class
-        instance = cls()
-        cls._instance = instance
-
-        # register_data_source takes a callable so let it rerun singleton to retrieve the instance
-        if register:
-            try:
-                _logger.debug("Registering kafka metrics data source.")
-                newrelic.core.agent.agent_instance().register_data_source(cls.factory)
-            except Exception:
-                _logger.exception(
-                    "Attempt to register kafka metrics data source has failed. Data source will be skipped."
-                )
-
-        return instance
-
-    def register(self, reporter):
-        self.reporters.append(reporter)
-
-    def unregister(self, reporter):
-        if reporter in self.reporters:
-            self.reporters.remove(reporter)
-
-    def start(self):
-        return
-
-    def stop(self):
-        # Clear references to reporters to prevent them from participating in a reference cycle.
-        self.reporters = []
-
-    def __call__(self):
-        for reporter in self.reporters:
-            for name, metric in six.iteritems(reporter.snapshot()):
-                yield name, metric
-
-
-class NewRelicMetricsReporter(AbstractMetricsReporter):
-    def __init__(self, *args, **kwargs):
-        super(NewRelicMetricsReporter, self).__init__(*args, **kwargs)
-
-        # Register with data source for harvesting
-        self.data_source = KafkaMetricsDataSource.singleton()
-        self.data_source.register(self)
-
-        self._metrics = {}
-        self._lock = threading.Lock()
-
-    def close(self, *args, **kwargs):
-        self.data_source.unregister(self)
-        with self._lock:
-            self._metrics = {}
-
-    def init(self, metrics):
-        for metric in metrics:
-            self.metric_change(metric)
-
-    @staticmethod
-    def invalid_metric_value(metric):
-        name, value = metric
-        return not any((math.isinf(value), math.isnan(value), value == 0))
-
-    def snapshot(self):
-        with self._lock:
-            # metric.value can only be called once, so care must be taken when filtering
-            metrics = ((name, metric.value()) for name, metric in six.iteritems(self._metrics))
-            return {
-                "MessageBroker/Kafka/Internal/%s" % name: {"count": value}
-                for name, value in filter(self.invalid_metric_value, metrics)
-            }
-
-    def get_metric_name(self, metric):
-        metric_name = metric.metric_name  # Get MetricName object to work with
-
-        name = metric_name.name
-        group = metric_name.group
-
-        if "topic" in metric_name.tags:
-            topic = metric_name.tags["topic"]
-            return "/".join((group, topic, name))
-        else:
-            return "/".join((group, name))
-
-    def metric_change(self, metric):
-        name = self.get_metric_name(metric)
-        with self._lock:
-            self._metrics[name] = metric
-
-    def metric_removal(self, metric):
-        name = self.get_metric_name(metric)
-        with self._lock:
-            if name in self._metrics:
-                self._metrics.pop(name)
-
-    def configure(self, configs):
-        return
-
-
-def wrap_KafkaProducerConsumer_init(wrapped, instance, args, kwargs):
-    try:
-        if "metric_reporters" in kwargs:
-            metric_reporters = list(kwargs.get("metric_reporters", []))
-            metric_reporters.append(NewRelicMetricsReporter)
-            kwargs["metric_reporters"] = [metric_reporters]
-        else:
-            kwargs["metric_reporters"] = [NewRelicMetricsReporter]
-    except Exception:
-        pass
-
-    return wrapped(*args, **kwargs)
-
-
 def instrument_kafka_producer(module):
     if hasattr(module, "KafkaProducer"):
-        wrap_function_wrapper(module, "KafkaProducer.__init__", wrap_KafkaProducerConsumer_init)
         wrap_function_wrapper(module, "KafkaProducer.send", wrap_KafkaProducer_send)
 
 
 def instrument_kafka_consumer_group(module):
     if hasattr(module, "KafkaConsumer"):
-        wrap_function_wrapper(module, "KafkaConsumer.__init__", wrap_KafkaProducerConsumer_init)
         wrap_function_wrapper(module.KafkaConsumer, "__next__", wrap_kafkaconsumer_next)

tests/messagebroker_kafkapython/conftest.py

@@ -26,7 +26,6 @@
 
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import transient_function_wrapper
-from newrelic.hooks.messagebroker_kafkapython import KafkaMetricsDataSource
 
 DB_SETTINGS = kafka_settings()[0]
 
@@ -55,7 +54,7 @@
 
 
 @pytest.fixture(scope="function")
-def producer(topic, data_source):
+def producer(topic):
     producer = kafka.KafkaProducer(
         bootstrap_servers=BROKER, api_version=(2, 0, 2), value_serializer=lambda v: json.dumps(v).encode("utf-8")
     )
@@ -64,7 +63,7 @@ def producer(topic, data_source):
 
 
 @pytest.fixture(scope="function")
-def consumer(topic, data_source, producer):
+def consumer(topic, producer):
     consumer = kafka.KafkaConsumer(
         topic,
         bootstrap_servers=BROKER,
@@ -99,17 +98,6 @@ def topic():
     # admin.delete_topics([topic])
 
 
-@pytest.fixture(scope="session")
-def data_source():
-    """
-    Must be required by consumer and producer fixtures, or the first one of them to be
-    instantiated will create and register the singleton. We rely on the singleton to
-    not be registered to properly test the output of it without interference from the
-    harvest thread.
-    """
-    return KafkaMetricsDataSource.singleton(register=False)
-
-
 @transient_function_wrapper(kafka.producer.kafka, "KafkaProducer.send.__wrapped__")
 # Place transient wrapper underneath instrumentation
 def cache_kafka_producer_headers(wrapped, instance, args, kwargs):