Repository: scoutapp/scout_apm_python
Pull request #305 (2019-10-28 14:02:11)

Remove unnecessary 'python -m' from Travis invocation


Patch:
diff --git a/.travis.yml b/.travis.yml
index 53c0232..6a203c8 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -48,7 +48,7 @@ jobs:
         - twine check dist/* wheelhouse/*
         - ls -al dist wheelhouse
         - if [ "${TRAVIS_TAG:-}" != "" ]; then
-            python -m twine upload --skip-existing dist/* wheelhouse/*;
+            twine upload --skip-existing dist/* wheelhouse/*;
           fi
 
 notifications:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a39dfee..35f195d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,8 @@
 - Changed `scout_apm.api.install()` signature to not take `*args, **kwargs` but
   just `config` (adds dependency on `kwargs-only`)
   ([PR #304](https://github.com/scoutapp/scout_apm_python/pull/304)).
+- Rewrite background samplers thread to avoid some rare race conditions
+  ([PR #307](https://github.com/scoutapp/scout_apm_python/pull/307)).
 
 ## [2.6.1] 2019-10-23
 
diff --git a/src/scout_apm/core/__init__.py b/src/scout_apm/core/__init__.py
index 1366a9e..123a33c 100644
--- a/src/scout_apm/core/__init__.py
+++ b/src/scout_apm/core/__init__.py
@@ -2,8 +2,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import logging
-import sys
-from os import getpid
+import os
 
 from kwargs_only import kwargs_only
 
@@ -24,14 +23,16 @@ def install(config=None):
         scout_config.set(**config)
     context = AgentContext.build(config=scout_config)
 
-    if sys.platform == "win32":
-        logger.info("APM Not Launching on PID: %s - Windows is not supported", getpid())
+    if os.name == "nt":
+        logger.info(
+            "APM Not Launching on PID: %s - Windows is not supported", os.getpid()
+        )
         return False
 
     if not context.config.value("monitor"):
         logger.info(
             "APM Not Launching on PID: %s - Configuration 'monitor' is not true",
-            getpid(),
+            os.getpid(),
         )
         return False
 
@@ -39,7 +40,7 @@ def install(config=None):
 
     objtrace.enable()
 
-    logger.debug("APM Launching on PID: %s", getpid())
+    logger.debug("APM Launching on PID: %s", os.getpid())
     launched = CoreAgentManager().launch()
 
     AppMetadata.report()
diff --git a/src/scout_apm/core/samplers/__init__.py b/src/scout_apm/core/samplers/__init__.py
index 2f33932..c5b3997 100644
--- a/src/scout_apm/core/samplers/__init__.py
+++ b/src/scout_apm/core/samplers/__init__.py
@@ -1,54 +1,2 @@
 # coding=utf-8
 from __future__ import absolute_import, division, print_function, unicode_literals
-
-import datetime as dt
-import logging
-import threading
-from os import getpid
-from time import sleep
-
-from scout_apm.core.commands import ApplicationEvent
-from scout_apm.core.context import AgentContext
-from scout_apm.core.samplers.cpu import Cpu
-from scout_apm.core.samplers.memory import Memory
-
-logger = logging.getLogger(__name__)
-
-
-class Samplers(object):
-    _thread_lock = threading.Semaphore()
-
-    @classmethod
-    def ensure_running(cls):
-        if cls._thread_lock.acquire(False):
-            th = threading.Thread(target=Samplers.run_samplers)
-            th.daemon = True
-            th.start()
-            cls._thread_lock.release()
-
-    @classmethod
-    def run_samplers(cls):
-        logger.debug("Starting Samplers. Acquiring samplers lock.")
-        try:
-            if cls._thread_lock.acquire(True):
-                logger.debug("Acquired samplers lock.")
-                instances = [Cpu(), Memory()]
-
-                while True:
-                    for instance in instances:
-                        event_value = instance.run()
-                        if event_value is not None:
-                            event_type = (
-                                instance.metric_type() + "/" + instance.metric_name()
-                            )
-                            event = ApplicationEvent(
-                                event_value=event_value,
-                                event_type=event_type,
-                                timestamp=dt.datetime.utcnow(),
-                                source="Pid: " + str(getpid()),
-                            )
-                            AgentContext.socket().send(event)
-                    sleep(60)
-        finally:
-            logger.debug("Shutting down samplers thread.")
-            cls._thread_lock.release()
diff --git a/src/scout_apm/core/samplers/cpu.py b/src/scout_apm/core/samplers/cpu.py
index 12ac9c8..fb20b37 100644
--- a/src/scout_apm/core/samplers/cpu.py
+++ b/src/scout_apm/core/samplers/cpu.py
@@ -10,23 +10,18 @@ logger = logging.getLogger(__name__)
 
 
 class Cpu(object):
+    metric_type = "CPU"
+    metric_name = "Utilization"
+    human_name = "Process CPU"
+
     def __init__(self):
         self.last_run = dt.datetime.utcnow()
         self.last_cpu_times = psutil.Process().cpu_times()
         self.num_processors = psutil.cpu_count()
         if self.num_processors is None:
-            logger.debug("Could not determine CPU count - assume there is one.")
+            logger.debug("Could not determine CPU count - assuming there is one.")
             self.num_processors = 1
 
-    def metric_type(self):
-        return "CPU"
-
-    def metric_name(self):
-        return "Utilization"
-
-    def human_name(self):
-        return "Process CPU"
-
     def run(self):
         now = dt.datetime.utcnow()
         process = psutil.Process()  # get a handle on the current process
@@ -37,7 +32,7 @@ class Cpu(object):
             self.save_times(now, cpu_times)
             logger.debug(
                 "%s: Negative time elapsed. now: %s, last_run: %s.",
-                self.human_name(),
+                self.human_name,
                 now,
                 self.last_run,
             )
@@ -55,7 +50,7 @@ class Cpu(object):
                 "%s: Negative process time elapsed. "
                 "utime: %s, stime: %s, total time: %s. "
                 "This is normal to see when starting a forking web server.",
-                self.human_name(),
+                self.human_name,
                 utime_elapsed,
                 stime_elapsed,
                 process_elapsed,
@@ -73,7 +68,7 @@ class Cpu(object):
 
         self.save_times(now, cpu_times)
 
-        logger.debug("%s: %s [%s CPU(s)]", self.human_name(), res, self.num_processors)
+        logger.debug("%s: %s [%s CPU(s)]", self.human_name, res, self.num_processors)
 
         return res
 
diff --git a/src/scout_apm/core/samplers/memory.py b/src/scout_apm/core/samplers/memory.py
index ace3c8d..7728d82 100644
--- a/src/scout_apm/core/samplers/memory.py
+++ b/src/scout_apm/core/samplers/memory.py
@@ -8,44 +8,17 @@ import psutil
 logger = logging.getLogger(__name__)
 
 
+def get_rss_in_mb():
+    rss_in_bytes = psutil.Process().memory_info().rss
+    return rss_in_bytes / (1024 * 1024)
+
+
 class Memory(object):
-    @staticmethod
-    def rss_to_mb(rss_in_bytes):
-        """
-        Convert a number of bytes to a number of megabytes.
-        """
-        bytes_per_kb = 1024
-        kb_per_megabyte = 1024
-        return float(rss_in_bytes) / bytes_per_kb / kb_per_megabyte
-
-    @staticmethod
-    def rss():
-        """
-        Returns the memory usage (RSS) of this process, in bytes
-        """
-        return psutil.Process().memory_info().rss
-
-    @staticmethod
-    def rss_in_mb():
-        return Memory.rss_to_mb(Memory.rss())
-
-    @staticmethod
-    def get_delta(prior_rss_in_mb):
-        mem = Memory.rss_in_mb()
-        if mem > prior_rss_in_mb:
-            return mem - prior_rss_in_mb
-        return 0.0
-
-    def metric_type(self):
-        return "Memory"
-
-    def metric_name(self):
-        return "Physical"
-
-    def human_name(self):
-        return "Process Memory"
+    metric_type = "Memory"
+    metric_name = "Physical"
+    human_name = "Process Memory"
 
     def run(self):
-        res = self.__class__.rss_in_mb()
-        logger.debug("%s: #%s", self.human_name(), res)
-        return res
+        value = get_rss_in_mb()
+        logger.debug("%s: #%s", self.human_name, value)
+        return value
diff --git a/src/scout_apm/core/samplers/thread.py b/src/scout_apm/core/samplers/thread.py
new file mode 100644
index 0000000..5f90feb
--- /dev/null
+++ b/src/scout_apm/core/samplers/thread.py
@@ -0,0 +1,71 @@
+# coding=utf-8
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+import datetime as dt
+import logging
+import os
+import threading
+
+from scout_apm.core.commands import ApplicationEvent
+from scout_apm.core.context import AgentContext
+from scout_apm.core.samplers.cpu import Cpu
+from scout_apm.core.samplers.memory import Memory
+
+logger = logging.getLogger(__name__)
+
+
+class SamplersThread(threading.Thread):
+    _instance = None
+    _instance_lock = threading.Lock()
+    _stop_event = threading.Event()
+
+    @classmethod
+    def ensure_started(cls):
+        with cls._instance_lock:
+            if cls._instance is None:
+                cls._instance = cls()
+                cls._instance.start()
+
+    @classmethod
+    def ensure_stopped(cls):
+        with cls._instance_lock:
+            if cls._instance is None:
+                # Nothing to stop
+                return
+            elif not cls._instance.is_alive():
+                # Thread died
+                cls._instance = None
+                return
+
+            # Signal stopping
+            cls._stop_event.set()
+            cls._instance.join()
+
+            cls._instance = None
+            cls._stop_event.clear()
+
+    def __init__(self, *args, **kwargs):
+        super(SamplersThread, self).__init__(*args, **kwargs)
+        self.daemon = True
+
+    def run(self):
+        logger.debug("Starting Samplers.")
+        instances = [Cpu(), Memory()]
+
+        while True:
+            for instance in instances:
+                event_value = instance.run()
+                if event_value is not None:
+                    event_type = instance.metric_type + "/" + instance.metric_name
+                    event = ApplicationEvent(
+                        event_value=event_value,
+                        event_type=event_type,
+                        timestamp=dt.datetime.utcnow(),
+                        source="Pid: " + str(os.getpid()),
+                    )
+                    AgentContext.socket().send(event)
+
+            should_stop = self._stop_event.wait(timeout=60)
+            if should_stop:
+                logger.debug("Stopping Samplers.")
+                break
diff --git a/src/scout_apm/core/tracked_request.py b/src/scout_apm/core/tracked_request.py
index 723ae49..f5e8256 100644
--- a/src/scout_apm/core/tracked_request.py
+++ b/src/scout_apm/core/tracked_request.py
@@ -9,7 +9,8 @@ from scout_apm.core import backtrace, objtrace
 from scout_apm.core.commands import BatchCommand
 from scout_apm.core.context import AgentContext
 from scout_apm.core.n_plus_one_call_set import NPlusOneCallSet
-from scout_apm.core.samplers import Memory, Samplers
+from scout_apm.core.samplers.memory import get_rss_in_mb
+from scout_apm.core.samplers.thread import SamplersThread
 from scout_apm.core.thread_local import ThreadLocalSingleton
 
 logger = logging.getLogger(__name__)
@@ -30,7 +31,7 @@ class TrackedRequest(ThreadLocalSingleton):
         "complete_spans",
         "tags",
         "_is_real_request",
-        "memory_start",
+        "_memory_start",
         "callset",
     )
 
@@ -42,7 +43,7 @@ class TrackedRequest(ThreadLocalSingleton):
         self.complete_spans = []
         self.tags = {}
         self._is_real_request = False
-        self.memory_start = Memory.rss_in_mb()
+        self._memory_start = get_rss_in_mb()
         self.callset = NPlusOneCallSet()
         logger.debug("Starting request: %s", self.request_id)
 
@@ -111,11 +112,11 @@ class TrackedRequest(ThreadLocalSingleton):
         if self.end_time is None:
             self.end_time = dt.datetime.utcnow()
         if self.is_real_request():
-            self.tag("mem_delta", Memory.get_delta(self.memory_start))
+            self.tag("mem_delta", self._get_mem_delta())
             if not self.is_ignored():
                 batch_command = BatchCommand.from_tracked_request(self)
                 AgentContext.socket().send(batch_command)
-            Samplers.ensure_running()
+            SamplersThread.ensure_started()
 
         # This can fail if the Tracked Request was created directly,
         # not through instance()
@@ -124,6 +125,12 @@ class TrackedRequest(ThreadLocalSingleton):
         except Exception:
             pass
 
+    def _get_mem_delta(self):
+        current_mem = get_rss_in_mb()
+        if current_mem > self._memory_start:
+            return current_mem - self._memory_start
+        return 0.0
+
     # A request is ignored if the tag "ignore_transaction" is set to True
     def is_ignored(self):
         return self.tags.get("ignore_transaction", False)


Test patch:
diff --git a/tests/unit/core/samplers/test_cpu.py b/tests/unit/core/samplers/test_cpu.py
index f529d8b..7a4de7d 100644
--- a/tests/unit/core/samplers/test_cpu.py
+++ b/tests/unit/core/samplers/test_cpu.py
@@ -10,18 +10,6 @@ from scout_apm.core.samplers.cpu import Cpu
 from tests.compat import mock
 
 
-def test_metric_type():
-    assert Cpu().metric_type() == "CPU"
-
-
-def test_metric_name():
-    assert Cpu().metric_name() == "Utilization"
-
-
-def test_human_name():
-    assert Cpu().human_name() == "Process CPU"
-
-
 def test_run_negative_time_elapsed(caplog):
     cpu = Cpu()
     cpu.last_run = dt.datetime.utcnow() + dt.timedelta(days=100)
@@ -103,7 +91,7 @@ def test_run_undetermined_cpus(caplog):
     assert record_tuples[0] == (
         "scout_apm.core.samplers.cpu",
         logging.DEBUG,
-        "Could not determine CPU count - assume there is one.",
+        "Could not determine CPU count - assuming there is one.",
     )
     _, level, message = record_tuples[1]
     assert level == logging.DEBUG
diff --git a/tests/unit/core/samplers/test_memory.py b/tests/unit/core/samplers/test_memory.py
index 0702fbc..45d0414 100644
--- a/tests/unit/core/samplers/test_memory.py
+++ b/tests/unit/core/samplers/test_memory.py
@@ -3,57 +3,9 @@ from __future__ import absolute_import, division, print_function, unicode_litera
 
 import logging
 
-import pytest
-import six
-
 from scout_apm.core.samplers.memory import Memory
 
 
-@pytest.mark.parametrize(
-    "given, expected",
-    [
-        (0, 0.0),
-        (1024 * 1024, 1.0),
-        (2 * 1024 * 1024, 2.0),
-        (2 * 1024 * 1024 + 1, 2.0000009536743164),
-    ],
-)
-def test_rss_to_mb(given, expected):
-    assert Memory.rss_to_mb(given) == expected
-
-
-def test_rss():
-    result = Memory.rss()
-    assert isinstance(result, six.integer_types) and result > 0
-
-
-def test_rss_in_mb():
-    result = Memory.rss_in_mb()
-    assert isinstance(result, float) and result > 0.0
-
-
-def test_get_delta():
-    result = Memory.get_delta(1.0)
-    assert isinstance(result, float) and result > 0.0
-
-
-def test_get_delta_big():
-    result = Memory.get_delta(1e12)
-    assert result == 0.0
-
-
-def test_metric_type():
-    assert Memory().metric_type() == "Memory"
-
-
-def test_metric_name():
-    assert Memory().metric_name() == "Physical"
-
-
-def test_human_name():
-    assert Memory().human_name() == "Process Memory"
-
-
 def test_run(caplog):
     result = Memory().run()
     assert isinstance(result, float) and result > 0.0
diff --git a/tests/unit/core/samplers/test_thread.py b/tests/unit/core/samplers/test_thread.py
new file mode 100644
index 0000000..6cff924
--- /dev/null
+++ b/tests/unit/core/samplers/test_thread.py
@@ -0,0 +1,43 @@
+# coding=utf-8
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+import time
+
+import pytest
+
+from scout_apm.core.samplers.thread import SamplersThread
+
+
+@pytest.fixture(autouse=True)
+def ensure_stopped_before_and_after():
+    SamplersThread.ensure_stopped()
+    try:
+        yield
+    finally:
+        assert not SamplersThread._stop_event.is_set()
+        SamplersThread.ensure_stopped()
+
+
+def test_ensure_started(caplog):
+    SamplersThread.ensure_started()
+    time.sleep(0.001)
+
+    assert SamplersThread._instance is not None
+    assert caplog.record_tuples[0] == (
+        "scout_apm.core.samplers.thread",
+        10,
+        "Starting Samplers.",
+    )
+
+
+def test_ensure_stopped(caplog):
+    SamplersThread.ensure_started()
+    time.sleep(0.001)
+    SamplersThread.ensure_stopped()
+
+    assert SamplersThread._instance is None
+    assert caplog.record_tuples[-1] == (
+        "scout_apm.core.samplers.thread",
+        10,
+        "Stopping Samplers.",
+    )
diff --git a/tests/unit/test_core.py b/tests/unit/test_core.py
index bcfe54b..3cdda80 100644
--- a/tests/unit/test_core.py
+++ b/tests/unit/test_core.py
@@ -3,7 +3,6 @@ from __future__ import absolute_import, division, print_function, unicode_litera
 
 import logging
 import os
-import sys
 
 from scout_apm.api import Config
 from scout_apm.core import install
@@ -12,7 +11,7 @@ from tests.compat import mock
 
 
 def test_install_fail_windows(caplog):
-    with mock.patch.object(sys, "platform", new="win32"):
+    with mock.patch.object(os, "name", new="nt"):
         installed = install()
 
     assert installed is False