Better fork detection & handling (#156)

* Better os.fork() handling

* improved load conditions

* Add harmless exception handling

* Fix comment of explanation

* Better; smarter; stronger Fork handling

Author: pglombardo

instana/__init__.py

@@ -95,9 +95,11 @@ def boot_agent():
 do_not_load_list = ["pip", "pip2", "pip3", "pipenv", "docker-compose", "easy_install", "easy_install-2.7",
                     "smtpd.py", "ufw", "unattended-upgrade"]
 
-if os.path.basename(sys.argv[0]) in do_not_load_list:
+# There are cases when sys.argv may not be defined at load time.  Seems to happen in embedded Python,
+# and some Pipenv installs.  If this is the case, it's best effort.
+if hasattr(sys, 'argv') and len(sys.argv) > 0 and (os.path.basename(sys.argv[0]) in do_not_load_list):
     if "INSTANA_DEBUG" in os.environ:
-        print("Instana: No use in monitoring this process type (%s).  Will go sit in a corner quietly.", sys.argv[0])
+        print("Instana: No use in monitoring this process type (%s).  Will go sit in a corner quietly." % os.path.basename(sys.argv[0]))
 else:
     if "INSTANA_MAGIC" in os.environ:
         # If we're being loaded into an already running process, then delay agent initialization

instana/agent.py

@@ -3,7 +3,7 @@
 import json
 import os
 from datetime import datetime
-
+from threading import Timer
 import requests
 
 import instana.singletons
@@ -25,6 +25,16 @@ def __init__(self, **kwds):
 
 
 class Agent(object):
+    """
+    The Agent class is the central controlling entity for the Instana Python language sensor.  The key
+    parts it handles are the announce state and the collection and reporting of metrics and spans to the
+    Instana Host agent.
+
+    To do this, there are 3 major components to this class:
+      1. TheMachine - finite state machine related to announce state
+      2. Sensor -> Meter - metric collection and reporting
+      3. Tracer -> Recorder - span queueing and reporting
+    """
     sensor = None
     host = AGENT_DEFAULT_HOST
     port = AGENT_DEFAULT_PORT
@@ -45,9 +55,27 @@ def __init__(self):
 
     def start(self, e):
         """ Starts the agent and required threads """
-        logger.debug("Spawning metric & trace reporting threads")
-        self.sensor.meter.run()
-        instana.singletons.tracer.recorder.run()
+        logger.debug("Spawning metric & span reporting threads")
+        self.sensor.start()
+        instana.singletons.tracer.recorder.start()
+
+    def handle_fork(self):
+        """
+        Forks happen.  Here we handle them.  Affected components are the singletons: Agent, Sensor & Tracers
+        """
+        # Reset the Agent
+        self.reset()
+
+        # Ask the sensor to handle the fork
+        self.sensor.handle_fork()
+
+        # Ask the tracer to handle the fork
+        instana.singletons.tracer.handle_fork()
+
+    def reset(self):
+        self.last_seen = None
+        self.from_ = From()
+        self.machine.reset()
 
     def to_json(self, o):
         def extractor(o):
@@ -66,10 +94,11 @@ def is_timed_out(self):
         return False
 
     def can_send(self):
-        # Watch for pid change in the case of ; if so, re-announce
+        # Watch for pid change (fork)
         current_pid = os.getpid()
         if self._boot_pid != current_pid:
             self._boot_pid = current_pid
+            logger.debug("Fork detected; Handling like a pro...")
             self.handle_fork()
             return False
 
@@ -96,19 +125,6 @@ def set_from(self, json_string):
 
         self.from_ = From(pid=res_data['pid'], agentUuid=res_data['agentUuid'])
 
-    def reset(self):
-        self.last_seen = None
-        self.from_ = From()
-        self.machine.reset()
-
-    def handle_fork(self):
-        """
-        Forks happen.  Here we handle them.
-        """
-        self.reset()
-        self.sensor.handle_fork()
-        instana.singletons.tracer.handle_fork()
-
     def is_agent_listening(self, host, port):
         """
         Check if the Instana Agent is listening on <host> and <port>.

instana/fsm.py

@@ -62,18 +62,20 @@ def __init__(self, agent):
                 ("pending",  "announced",    "wait4init"),
                 ("ready",    "wait4init",    "good2go")],
             "callbacks": {
+                # Can add the following to debug
+                # "onchangestate":  self.print_state_change,
                 "onlookup":       self.lookup_agent_host,
                 "onannounce":     self.announce_sensor,
                 "onpending":      self.agent.start,
-                "onready":        self.on_ready,
-                "onchangestate":  self.printstatechange}})
+                "onready":        self.on_ready}})
 
         self.timer = t.Timer(5, self.fsm.lookup)
         self.timer.daemon = True
-        self.timer.name = "Startup"
+        self.timer.name = "Instana Machine"
         self.timer.start()
 
-    def printstatechange(self, e):
+    @staticmethod
+    def print_state_change(e):
         logger.debug('========= (%i#%s) FSM event: %s, src: %s, dst: %s ==========' %
                      (os.getpid(), t.current_thread().name, e.event, e.src, e.dst))
 

instana/meter.py

@@ -7,6 +7,7 @@
 import sys
 import threading
 from types import ModuleType
+from fysom import FysomError
 
 from pkg_resources import DistributionNotFound, get_distribution
 
@@ -105,16 +106,19 @@ def to_dict(self):
 
 class Meter(object):
     SNAPSHOT_PERIOD = 600
-    snapshot_countdown = 0
+    THREAD_NAME = "Instana Metric Collection"
 
     # The agent that this instance belongs to
     agent = None
 
+    # We send Snapshot data every 10 minutes.  This is the countdown variable.
+    snapshot_countdown = 0
+
     last_usage = None
     last_collect = None
     last_metrics = None
     djmw = None
-    thr = None
+    thread = None
 
     # A True value signals the metric reporting thread to shutdown
     _shutdown = False
@@ -123,33 +127,57 @@ def __init__(self, agent):
         self.agent = agent
         pass
 
-    def run(self):
-        """ Spawns the metric reporting thread """
-        self.thr = threading.Thread(target=self.collect_and_report)
-        self.thr.daemon = True
-        self.thr.name = "Instana Metric Collection"
-        self.thr.start()
+    def start(self):
+        """
+        This function can be called at first boot or after a fork.  In either case, it will
+        assure that the Meter is in a proper state (via reset()) and spawn a new background
+        thread to periodically report queued spans
+
+        Note that this will abandon any previous thread object that (in the case of an `os.fork()`)
+        should no longer exist in the forked process.
+
+        (Forked processes carry forward only the thread that called `os.fork()`
+        into the new process space.  All other background threads need to be recreated.)
+
+        Calling this directly more than once without an actual fork will cause errors.
+        """
+        self.reset()
+
+        if self.thread.isAlive() is False:
+            self.thread.start()
 
     def reset(self):
         """" Reset the state as new """
         self.last_usage = None
         self.last_collect = None
         self.last_metrics = None
         self.snapshot_countdown = 0
+        self.thread = None
+
+        # Prepare the thread for metric collection/reporting
+        for thread in threading.enumerate():
+            if thread.getName() == self.THREAD_NAME:
+                # Metric thread already exists; Make sure we re-use this one.
+                self.thread = thread
+
+        if self.thread is None:
+            self.thread = threading.Thread(target=self.collect_and_report)
+            self.thread.daemon = True
+            self.thread.name = self.THREAD_NAME
 
     def handle_fork(self):
-        self.reset()
-        self.run()
+        self.start()
 
     def collect_and_report(self):
         """
         Target function for the metric reporting thread.  This is a simple loop to
         collect and report entity data every 1 second.
         """
-        logger.debug("Metric reporting thread is now alive")
+        logger.debug(" -> Metric reporting thread is now alive")
 
         def metric_work():
             self.process()
+
             if self.agent.is_timed_out():
                 logger.warn("Host agent offline for >1 min.  Going to sit in a corner...")
                 self.agent.reset()
@@ -160,12 +188,17 @@ def metric_work():
 
     def process(self):
         """ Collects, processes & reports metrics """
-        if self.agent.machine.fsm.current is "wait4init":
-            # Test the host agent if we're ready to send data
-            if self.agent.is_agent_ready():
-                self.agent.machine.fsm.ready()
-            else:
-                return
+        try:
+            if self.agent.machine.fsm.current is "wait4init":
+                # Test the host agent if we're ready to send data
+                if self.agent.is_agent_ready():
+                    if self.agent.machine.fsm.current is not "good2go":
+                        self.agent.machine.fsm.ready()
+                else:
+                    return
+        except FysomError:
+            logger.debug('Harmless state machine thread disagreement.  Will self-correct on next timer cycle.')
+            return
 
         if self.agent.can_send():
             self.snapshot_countdown = self.snapshot_countdown - 1

instana/recorder.py

@@ -2,7 +2,7 @@
 
 import os
 import sys
-import threading as t
+import threading
 
 import opentracing.ext.tags as ext
 from basictracer import Sampler, SpanRecorder
@@ -15,13 +15,14 @@
 from .log import logger
 from .util import every
 
-if sys.version_info.major is 2:
+if sys.version_info.major == 2:
     import Queue as queue
 else:
     import queue
 
 
 class InstanaRecorder(SpanRecorder):
+    THREAD_NAME = "Instana Span Reporting"
     registered_spans = ("aiohttp-client", "aiohttp-server", "django", "log", "memcache", "mysql",
                         "rabbitmq", "redis", "rpc-client", "rpc-server", "sqlalchemy", "soap",
                         "tornado-client", "tornado-server", "urllib3", "wsgi")
@@ -35,23 +36,51 @@ class InstanaRecorder(SpanRecorder):
     entry_kind = ["entry", "server", "consumer"]
     exit_kind = ["exit", "client", "producer"]
 
-    queue = queue.Queue()
-
-    timer = None
+    # Recorder thread for collection/reporting of spans
+    thread = None
 
     def __init__(self):
         super(InstanaRecorder, self).__init__()
+        self.queue = queue.Queue()
+
+    def start(self):
+        """
+        This function can be called at first boot or after a fork.  In either case, it will
+        assure that the Recorder is in a proper state (via reset()) and spawn a new background
+        thread to periodically report queued spans
+
+        Note that this will abandon any previous thread object that (in the case of an `os.fork()`)
+        should no longer exist in the forked process.
+
+        (Forked processes carry forward only the thread that called `os.fork()`
+        into the new process space.  All other background threads need to be recreated.)
+
+        Calling this directly more than once without an actual fork will cause errors.
+        """
+        self.reset()
+
+        if self.thread.isAlive() is False:
+            self.thread.start()
+
+    def reset(self):
+        # Prepare the thread for metric collection/reporting
+        for thread in threading.enumerate():
+            if thread.getName() == self.THREAD_NAME:
+                # Span reporting thread already exists; Make sure we re-use this one.
+                self.thread = thread
+
+        # Prepare the thread for span collection/reporting
+        if self.thread is None:
+            self.thread = threading.Thread(target=self.report_spans)
+            self.thread.daemon = True
+            self.thread.name = self.THREAD_NAME
 
-    def run(self):
-        """ Span a background thread to periodically report queued spans """
-        self.timer = t.Thread(target=self.report_spans)
-        self.timer.daemon = True
-        self.timer.name = "Instana Span Reporting"
-        self.timer.start()
+    def handle_fork(self):
+        self.start()
 
     def report_spans(self):
         """ Periodically report the queued spans """
-        logger.debug("Span reporting thread is now alive")
+        logger.debug(" -> Span reporting thread is now alive")
 
         def span_work():
             queue_size = self.queue.qsize()
@@ -69,14 +98,15 @@ def queue_size(self):
 
     def queued_spans(self):
         """ Get all of the spans in the queue """
+        span = None
         spans = []
         while True:
             try:
-                s = self.queue.get(False)
+                span = self.queue.get(False)
             except queue.Empty:
                 break
             else:
-                spans.append(s)
+                spans.append(span)
         return spans
 
     def clear_spans(self):

instana/sensor.py

@@ -23,6 +23,10 @@ def set_options(self, options):
         if not self.options:
             self.options = Options()
 
+    def start(self):
+        # Nothing to do for the Sensor;  Pass onto Meter
+        self.meter.start()
+
     def handle_fork(self):
         # Nothing to do for the Sensor;  Pass onto Meter
         self.meter.handle_fork()

instana/singletons.py

@@ -4,8 +4,6 @@
 from .agent import Agent
 from .tracer import InstanaTracer, InstanaRecorder
 
-from distutils.version import LooseVersion
-
 
 # The Instana Agent which carries along with it a Sensor that collects metrics.
 agent = Agent()

instana/tracer.py

@@ -29,6 +29,10 @@ def __init__(self, options=Options(), scope_manager=None, recorder=None):
         self._propagators[ot.Format.HTTP_HEADERS] = HTTPPropagator()
         self._propagators[ot.Format.TEXT_MAP] = TextPropagator()
 
+    def handle_fork(self):
+        # Nothing to do for the Tracer;  Pass onto Recorder
+        self.recorder.handle_fork()
+
     def start_active_span(self,
                           operation_name,
                           child_of=None,
@@ -118,9 +122,6 @@ def extract(self, format, carrier):
         else:
             raise ot.UnsupportedFormatException()
 
-    def handle_fork(self):
-        self.recorder = InstanaRecorder()
-
     def __add_stack(self, span, limit=None):
         """ Adds a backtrace to this span """
         span.stack = []