1from ddtrace.internal.runtime.constants import GC_COUNT_GEN0 2from ddtrace.internal.runtime.constants import GC_RUNTIME_METRICS 3from ddtrace.internal.runtime.constants import PSUTIL_RUNTIME_METRICS 4from ddtrace.internal.runtime.metric_collectors import GCRuntimeMetricCollector 5from ddtrace.internal.runtime.metric_collectors import PSUtilRuntimeMetricCollector 6from ddtrace.internal.runtime.metric_collectors import RuntimeMetricCollector 7from tests.utils import BaseTestCase 8 9 10class TestRuntimeMetricCollector(BaseTestCase): 11 def test_failed_module_load_collect(self): 12 """Attempts to collect from a collector when it has failed to load its 13 module should return no metrics gracefully. 14 """ 15 16 class A(RuntimeMetricCollector): 17 required_modules = ["moduleshouldnotexist"] 18 19 def collect_fn(self, keys): 20 return {"k": "v"} 21 22 self.assertIsNotNone(A().collect(), "collect should return valid metrics") 23 24 25class TestPSUtilRuntimeMetricCollector(BaseTestCase): 26 def test_metrics(self): 27 collector = PSUtilRuntimeMetricCollector() 28 for (key, value) in collector.collect(PSUTIL_RUNTIME_METRICS): 29 self.assertIsNotNone(value) 30 31 32class TestGCRuntimeMetricCollector(BaseTestCase): 33 def test_metrics(self): 34 collector = GCRuntimeMetricCollector() 35 for (key, value) in collector.collect(GC_RUNTIME_METRICS): 36 self.assertIsNotNone(value) 37 38 def test_gen1_changes(self): 39 # disable gc 40 import gc 41 42 gc.disable() 43 44 # start collector and get current gc counts 45 collector = GCRuntimeMetricCollector() 46 gc.collect() 47 start = gc.get_count() 48 49 # create reference 50 a = [] 51 collected = collector.collect([GC_COUNT_GEN0]) 52 self.assertGreater(collected[0][1], start[0]) 53 54 # delete reference and collect 55 del a 56 gc.collect() 57 collected_after = collector.collect([GC_COUNT_GEN0]) 58 assert len(collected_after) == 1 59 assert collected_after[0][0] == "runtime.python.gc.count.gen0" 60 assert isinstance(collected_after[0][1], int) 61