Better Fork detection & handling (#96)

* Detect pid change and reset if so.

* Fix log message formatting.

* Remove unnecessary parens

* Check queue size before more expensive can_send

* Cleanup & simplify boot architecture

* Simpler on import process
* Clearer relationships and inter-access between Agent, Sensor, Meter & Tracer
* Better fork detection and handling of forks

* Restore boot message; remove unused import

* Centralized singletons; safeties

* Remove remnant debug lines

* Linter; Better exceptions; Safety

Author: pglombardo

instana/__init__.py

@@ -4,13 +4,22 @@
 from pkg_resources import get_distribution
 
 """
-The Instana package has two core components: the sensor and the tracer.
+The Instana package has two core components: the agent and the tracer.
 
-The sensor is individual to each python process and handles process metric
+The agent is individual to each python process and handles process metric
 collection and reporting.
 
 The tracer upholds the OpenTracing API and is responsible for reporting
 span data to Instana.
+
+The following outlines the hierarchy of classes for these two components.
+
+Agent
+  Sensor
+    Meter
+
+Tracer
+  Recorder
 """
 
 __author__ = 'Instana Inc.'
@@ -33,23 +42,13 @@ def load(module):
         print("==========================================================")
 
 
-# Optional application wide service name.
-# Can be configured via environment variable or via code:
-#
-# export INSTANA_SERVICE_NAME=myservice
-#   or
-# instana.service_name = "myservice"
-service_name = None
-
 # User configurable EUM API key for instana.helpers.eum_snippet()
 eum_api_key = ''
 
-if "INSTANA_SERVICE_NAME" in os.environ:
-    service_name = os.environ["INSTANA_SERVICE_NAME"]
+import instana.singletons #noqa
 
 if "INSTANA_DISABLE_AUTO_INSTR" not in os.environ:
     # Import & initialize instrumentation
-    # noqa: ignore=W0611
     from .instrumentation import urllib3  # noqa
     from .instrumentation import sudsjurko  # noqa
     from .instrumentation import mysqlpython  # noqa

instana/agent.py

@@ -1,10 +1,15 @@
+from __future__ import absolute_import
+
 import json
+import os
 import threading
 from datetime import datetime
 
-import instana.agent_const as a
-import instana.fsm as f
-from instana import log
+from .log import logger
+from .agent_const import AGENT_DEFAULT_HOST, AGENT_DEFAULT_PORT
+from .fsm import Fsm
+from .sensor import Sensor
+import instana.singletons
 
 try:
     import urllib.request as urllib2
@@ -34,24 +39,25 @@ def get_method(self):
 
 class Agent(object):
     sensor = None
-    host = a.AGENT_DEFAULT_HOST
-    port = a.AGENT_DEFAULT_PORT
+    host = AGENT_DEFAULT_HOST
+    port = AGENT_DEFAULT_PORT
     fsm = None
     from_ = From()
     last_seen = None
+    last_fork_check = None
+    _boot_pid = os.getpid()
 
-    def __init__(self, sensor):
-        log.debug("initializing agent")
-
-        self.sensor = sensor
-        self.fsm = f.Fsm(self)
+    def __init__(self):
+        logger.debug("initializing agent")
+        self.sensor = Sensor(self)
+        self.fsm = Fsm(self)
 
     def to_json(self, o):
         try:
             return json.dumps(o, default=lambda o: {k.lower(): v for k, v in o.__dict__.items()},
                               sort_keys=False, separators=(',', ':')).encode()
         except Exception as e:
-            log.info("to_json: ", e, o)
+            logger.info("to_json: ", e, o)
 
     def is_timed_out(self):
         if self.last_seen and self.can_send:
@@ -61,7 +67,17 @@ def is_timed_out(self):
         return False
 
     def can_send(self):
-        return self.fsm.fsm.current == "good2go"
+        # Watch for pid change in the case of ; if so, re-announce
+        current_pid = os.getpid()
+        if self._boot_pid != current_pid:
+            self._boot_pid = current_pid
+            self.handle_fork()
+            return False
+
+        if (self.fsm.fsm.current == "good2go"):
+            return True
+
+        return False
 
     def head(self, url):
         return self.request(url, "HEAD", None)
@@ -96,7 +112,7 @@ def full_request_response(self, url, method, o, body, header):
                 self.reset()
             else:
                 if response.getcode() < 200 or response.getcode() >= 300:
-                    log.error("Request returned erroneous code", response.getcode())
+                    logger.error("Request returned erroneous code", response.getcode())
                     if self.can_send():
                         self.reset()
                 else:
@@ -113,8 +129,8 @@ def full_request_response(self, url, method, o, body, header):
             # No need to show the initial 404s or timeouts.  The agent
             # should handle those correctly.
             if not (type(e) is urllib2.HTTPError and e.code == 404):
-                log.debug("%s: full_request_response: %s" %
-                          (threading.current_thread().name, str(e)))
+                logger.debug("%s: full_request_response: %s" %
+                             (threading.current_thread().name, str(e)))
 
         return (b, h)
 
@@ -124,7 +140,7 @@ def make_url(self, prefix):
     def make_host_url(self, host, prefix):
         port = self.sensor.options.agent_port
         if port == 0:
-            port = a.AGENT_DEFAULT_PORT
+            port = AGENT_DEFAULT_PORT
 
         return self.make_full_url(host, port, prefix)
 
@@ -135,21 +151,20 @@ def make_full_url(self, host, port, prefix):
 
         return s
 
-    def reset(self):
-        self.last_seen = None
-        self.from_ = From()
-        self.fsm.reset()
-
-    def set_host(self, host):
-        self.host = host
-
-    def set_port(self, port):
-        self.port = port
-
     def set_from(self, json_string):
         if type(json_string) is bytes:
             raw_json = json_string.decode("UTF-8")
         else:
             raw_json = json_string
 
         self.from_ = From(**json.loads(raw_json))
+
+    def reset(self):
+        self.last_seen = None
+        self.from_ = From()
+        self.fsm.reset()
+
+    def handle_fork(self):
+        self.reset()
+        self.sensor.handle_fork()
+        instana.singletons.tracer.handle_fork()

instana/fsm.py

@@ -10,8 +10,8 @@
 import fysom as f
 from pkg_resources import get_distribution
 
-from . import agent_const as a
-from . import log
+from .agent_const import AGENT_HEADER, AGENT_DISCOVERY_URL, AGENT_DATA_URL, AGENT_DEFAULT_HOST, AGENT_DEFAULT_PORT
+from .log import logger
 
 
 class Discovery(object):
@@ -44,8 +44,9 @@ class Fsm(object):
     warnedPeriodic = False
 
     def __init__(self, agent):
-        log.info("Stan is on the scene.  Starting Instana instrumentation version", get_distribution('instana').version)
-        log.debug("initializing fsm")
+        logger.info("Stan is on the scene.  Starting Instana instrumentation version: %s" %
+                    get_distribution('instana').version)
+        logger.debug("initializing fsm")
 
         self.agent = agent
         self.fsm = f.Fysom({
@@ -65,8 +66,8 @@ def __init__(self, agent):
         timer.start()
 
     def printstatechange(self, e):
-        log.debug('========= (%i#%s) FSM event: %s, src: %s, dst: %s ==========' %
-                  (os.getpid(), t.current_thread().name, e.event, e.src, e.dst))
+        logger.debug('========= (%i#%s) FSM event: %s, src: %s, dst: %s ==========' %
+                     (os.getpid(), t.current_thread().name, e.event, e.src, e.dst))
 
     def reset(self):
         self.fsm.lookup()
@@ -78,30 +79,30 @@ def lookup_agent_host(self, e):
         host, port = self.__get_agent_host_port()
 
         h = self.check_host(host, port)
-        if h == a.AGENT_HEADER:
-            self.agent.set_host(host)
-            self.agent.set_port(port)
+        if h == AGENT_HEADER:
+            self.agent.host = host
+            self.agent.port = port
             self.fsm.announce()
             return True
         elif os.path.exists("/proc/"):
             host = self.get_default_gateway()
             if host:
                 h = self.check_host(host, port)
-                if h == a.AGENT_HEADER:
-                    self.agent.set_host(host)
-                    self.agent.set_port(port)
+                if h == AGENT_HEADER:
+                    self.agent.host = host
+                    self.agent.port = port
                     self.fsm.announce()
                     return True
 
         if (self.warnedPeriodic is False):
-            log.warn("Instana Host Agent couldn't be found. Will retry periodically...")
+            logger.warn("Instana Host Agent couldn't be found. Will retry periodically...")
             self.warnedPeriodic = True
 
         self.schedule_retry(self.lookup_agent_host, e, "agent_lookup")
         return False
 
     def get_default_gateway(self):
-        log.debug("checking default gateway")
+        logger.debug("checking default gateway")
 
         try:
             proc = subprocess.Popen(
@@ -111,20 +112,20 @@ def get_default_gateway(self):
             addr = proc.stdout.read()
             return addr.decode("UTF-8")
         except Exception as e:
-            log.error(e)
+            logger.error(e)
 
             return None
 
     def check_host(self, host, port):
-        log.debug("checking %s:%d" % (host, port))
+        logger.debug("checking %s:%d" % (host, port))
 
         (_, h) = self.agent.request_header(
             self.agent.make_host_url(host, "/"), "GET", "Server")
 
         return h
 
     def announce_sensor(self, e):
-        log.debug("announcing sensor to the agent")
+        logger.debug("announcing sensor to the agent")
         s = None
         pid = os.getpid()
         cmdline = []
@@ -144,9 +145,9 @@ def announce_sensor(self, e):
                 (out, err) = proc.communicate()
                 parts = out.split(b'\n')
                 cmdline = [parts[1].decode("utf-8")]
-        except Exception as err:
+        except Exception:
             cmdline = sys.argv
-            log.debug(err)
+            logger.debug("announce_sensor", exc_info=True)
 
         d = Discovery(pid=self.__get_real_pid(),
                       name=cmdline[0],
@@ -161,29 +162,29 @@ def announce_sensor(self, e):
             d.inode = os.readlink(path)
 
         (b, _) = self.agent.request_response(
-            self.agent.make_url(a.AGENT_DISCOVERY_URL), "PUT", d)
+            self.agent.make_url(AGENT_DISCOVERY_URL), "PUT", d)
         if b:
             self.agent.set_from(b)
             self.fsm.ready()
-            log.info("Host agent available. We're in business. Announced pid: %i (true pid: %i)" %
-                     (pid, self.agent.from_.pid))
+            logger.info("Host agent available. We're in business. Announced pid: %s (true pid: %s)" %
+                        (str(pid), str(self.agent.from_.pid)))
             return True
         else:
-            log.debug("Cannot announce sensor. Scheduling retry.")
+            logger.debug("Cannot announce sensor. Scheduling retry.")
             self.schedule_retry(self.announce_sensor, e, "announce")
         return False
 
     def schedule_retry(self, fun, e, name):
-        log.debug("Scheduling: " + name)
+        logger.debug("Scheduling: " + name)
         self.timer = t.Timer(self.RETRY_PERIOD, fun, [e])
         self.timer.daemon = True
         self.timer.name = name
         self.timer.start()
 
     def test_agent(self, e):
-        log.debug("testing communication with the agent")
+        logger.debug("testing communication with the agent")
 
-        (b, _) = self.agent.head(self.agent.make_url(a.AGENT_DATA_URL))
+        (b, _) = self.agent.head(self.agent.make_url(AGENT_DATA_URL))
 
         if not b:
             self.schedule_retry(self.test_agent, e, "agent test")
@@ -209,7 +210,7 @@ def __get_real_pid(self):
                     if len(g.groups()) == 1:
                         pid = int(g.groups()[0])
                 except Exception:
-                    log.debug("parsing sched file failed", exc_info=True)
+                    logger.debug("parsing sched file failed", exc_info=True)
                     pass
 
         if pid is None:
@@ -222,8 +223,8 @@ def __get_agent_host_port(self):
         Iterates the the various ways the host and port of the Instana host
         agent may be configured: default, env vars, sensor options...
         """
-        host = a.AGENT_DEFAULT_HOST
-        port = a.AGENT_DEFAULT_PORT
+        host = AGENT_DEFAULT_HOST
+        port = AGENT_DEFAULT_PORT
 
         if "INSTANA_AGENT_HOST" in os.environ:
             host = os.environ["INSTANA_AGENT_HOST"]

instana/helpers.py

@@ -2,7 +2,7 @@
 from string import Template
 
 from instana import eum_api_key as global_eum_api_key
-from instana.tracer import internal_tracer
+from .singletons import tracer
 from instana.log import logger
 
 # Usage:
@@ -33,7 +33,7 @@ def eum_snippet(trace_id=None, eum_api_key=None, meta={}):
         ids = {}
         ids['meta_kvs'] = ''
 
-        parent_span = internal_tracer.active_span
+        parent_span = tracer.active_span
 
         if trace_id or parent_span:
             ids['trace_id'] = trace_id or parent_span.trace_id
@@ -78,7 +78,7 @@ def eum_test_snippet(trace_id=None, eum_api_key=None, meta={}):
         ids = {}
         ids['meta_kvs'] = ''
 
-        parent_span = internal_tracer.active_span
+        parent_span = tracer.active_span
         if trace_id or parent_span:
             ids['trace_id'] = trace_id or parent_span.trace_id
         else: