Stability and determinism improvements (#140)

* Stability and logistical improvements:

1. Better timing mechanism via instana.util.every
  - avoids time drift
  - skip late executions (if runs become backed up)

2. Add a new FSM state: agent ready to determine when host agent is
ready to accept data (allows for #3)

3. Force snapshot reporting immediately once agent is ready to accept
data

This fixes:

- Entity queue full when metric reporting may be backed up
- Sending initial spans without Infrastructure snapshot
data.

* Fix return value and log level

* Rename the state machine

Author: pglombardo

instana/agent.py

@@ -11,7 +11,7 @@
 from .agent_const import (AGENT_DATA_PATH, AGENT_DEFAULT_HOST,
                           AGENT_DEFAULT_PORT, AGENT_DISCOVERY_PATH,
                           AGENT_HEADER, AGENT_RESPONSE_PATH, AGENT_TRACES_PATH)
-from .fsm import Fsm
+from .fsm import TheMachine
 from .log import logger
 from .sensor import Sensor
 
@@ -28,7 +28,7 @@ class Agent(object):
     sensor = None
     host = AGENT_DEFAULT_HOST
     port = AGENT_DEFAULT_PORT
-    fsm = None
+    machine = None
     from_ = From()
     last_seen = None
     last_fork_check = None
@@ -41,7 +41,7 @@ class Agent(object):
     def __init__(self):
         logger.debug("initializing agent")
         self.sensor = Sensor(self)
-        self.fsm = Fsm(self)
+        self.machine = TheMachine(self)
 
     def start(self, e):
         """ Starts the agent and required threads """
@@ -73,7 +73,7 @@ def can_send(self):
             self.handle_fork()
             return False
 
-        if self.fsm.fsm.current == "good2go":
+        if self.machine.fsm.current == "good2go":
             return True
 
         return False
@@ -99,7 +99,7 @@ def set_from(self, json_string):
     def reset(self):
         self.last_seen = None
         self.from_ = From()
-        self.fsm.reset()
+        self.machine.reset()
 
     def handle_fork(self):
         """
@@ -151,6 +151,19 @@ def announce(self, discovery):
         finally:
             return response
 
+    def is_agent_ready(self):
+        """
+        Used after making a successful announce to test when the agent is ready to accept data.
+        """
+        try:
+            response = self.client.head(self.__data_url(), timeout=0.8)
+
+            if response.status_code is 200:
+                return True
+            return False
+        except (requests.ConnectTimeout, requests.ConnectionError):
+            logger.debug("is_agent_ready: host agent connection error")
+
     def report_data(self, entity_data):
         """
         Used to report entity data (metrics & snapshot) to the host agent.
@@ -162,6 +175,8 @@ def report_data(self, entity_data):
                                         headers={"Content-Type": "application/json"},
                                         timeout=0.8)
 
+            # logger.warn("report_data: response.status_code is %s" % response.status_code)
+
             if response.status_code is 200:
                 self.last_seen = datetime.now()
         except (requests.ConnectTimeout, requests.ConnectionError):
@@ -179,6 +194,9 @@ def report_traces(self, spans):
                                         data=self.to_json(spans),
                                         headers={"Content-Type": "application/json"},
                                         timeout=0.8)
+
+            # logger.warn("report_traces: response.status_code is %s" % response.status_code)
+
             if response.status_code is 200:
                 self.last_seen = datetime.now()
         except (requests.ConnectTimeout, requests.ConnectionError):

instana/fsm.py

@@ -7,7 +7,7 @@
 import sys
 import threading as t
 
-import fysom as f
+from fysom import Fysom
 import pkg_resources
 
 from .agent_const import AGENT_DEFAULT_HOST, AGENT_DEFAULT_PORT
@@ -35,7 +35,7 @@ def to_dict(self):
         return kvs
 
 
-class Fsm(object):
+class TheMachine(object):
     RETRY_PERIOD = 30
 
     agent = None
@@ -55,15 +55,17 @@ def __init__(self, agent):
         logger.debug("initializing fsm")
 
         self.agent = agent
-        self.fsm = f.Fysom({
+        self.fsm = Fysom({
             "events": [
                 ("lookup",   "*",            "found"),
                 ("announce", "found",        "announced"),
-                ("ready",    "announced",    "good2go")],
+                ("pending",  "announced",    "wait4init"),
+                ("ready",    "wait4init",    "good2go")],
             "callbacks": {
                 "onlookup":       self.lookup_agent_host,
                 "onannounce":     self.announce_sensor,
-                "onready":        self.agent.start,
+                "onpending":      self.agent.start,
+                "onready":        self.on_ready,
                 "onchangestate":  self.printstatechange}})
 
         self.timer = t.Timer(5, self.fsm.lookup)
@@ -79,6 +81,7 @@ def reset(self):
         self.fsm.lookup()
 
     def lookup_agent_host(self, e):
+        logger.debug("lookup_agent_host")
         host, port = self.__get_agent_host_port()
 
         if self.agent.is_agent_listening(host, port):
@@ -95,7 +98,7 @@ def lookup_agent_host(self, e):
                     self.fsm.announce()
                     return True
 
-        if (self.warnedPeriodic is False):
+        if self.warnedPeriodic is False:
             logger.warn("Instana Host Agent couldn't be found. Will retry periodically...")
             self.warnedPeriodic = True
 
@@ -143,9 +146,8 @@ def announce_sensor(self, e):
 
         if response and (response.status_code is 200) and (len(response.content) > 2):
             self.agent.set_from(response.content)
-            self.fsm.ready()
-            logger.info("Host agent available. We're in business. Announced pid: %s (true pid: %s)" %
-                        (str(pid), str(self.agent.from_.pid)))
+            self.fsm.pending()
+            logger.debug("Announced pid: %s (true pid: %s) Waiting for Agent Ready" % (str(pid), str(self.agent.from_.pid)))
             return True
         else:
             logger.debug("Cannot announce sensor. Scheduling retry.")
@@ -159,6 +161,10 @@ def schedule_retry(self, fun, e, name):
         self.timer.name = name
         self.timer.start()
 
+    def on_ready(self, e):
+        logger.info("Host agent available. We're in business. Announced pid: %s (true pid: %s)" %
+                    (str(os.getpid()), str(self.agent.from_.pid)))
+
     def __get_real_pid(self):
         """
         Attempts to determine the true process ID by querying the

instana/meter.py

@@ -6,13 +6,12 @@
 import resource
 import sys
 import threading
-import time
 from types import ModuleType
 
 from pkg_resources import DistributionNotFound, get_distribution
 
 from .log import logger
-from .util import get_py_source, package_version
+from .util import get_py_source, package_version, every
 
 
 class Snapshot(object):
@@ -106,16 +105,16 @@ def to_dict(self):
 
 class Meter(object):
     SNAPSHOT_PERIOD = 600
-    snapshot_countdown = 5
+    snapshot_countdown = 0
 
     # The agent that this instance belongs to
     agent = None
 
     last_usage = None
     last_collect = None
     last_metrics = None
-    last_data_report_status = None
     djmw = None
+    thr = None
 
     # A True value signals the metric reporting thread to shutdown
     _shutdown = False
@@ -136,7 +135,7 @@ def reset(self):
         self.last_usage = None
         self.last_collect = None
         self.last_metrics = None
-        self.snapshot_countdown = 5
+        self.snapshot_countdown = 0
         self.run()
 
     def collect_and_report(self):
@@ -145,22 +144,33 @@ def collect_and_report(self):
         collect and report entity data every 1 second.
         """
         logger.debug("Metric reporting thread is now alive")
-        while 1:
+
+        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()
-                break
-            time.sleep(1)
+                return False
+            return True
+
+        every(1, metric_work, "Metrics Collection")
 
     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
+
         if self.agent.can_send():
             self.snapshot_countdown = self.snapshot_countdown - 1
             ss = None
             cm = self.collect_metrics()
 
-            if self.snapshot_countdown < 1 and self.last_data_report_status is 200:
+            if self.snapshot_countdown < 1:
+                logger.debug("Sending process snapshot data")
                 self.snapshot_countdown = self.SNAPSHOT_PERIOD
                 ss = self.collect_snapshot()
                 md = copy.deepcopy(cm).delta_data(None)
@@ -171,8 +181,6 @@ def process(self):
             response = self.agent.report_data(ed)
 
             if response:
-                self.last_data_report_status = response.status_code
-
                 if response.status_code is 200 and len(response.content) > 2:
                     # The host agent returned something indicating that is has a request for us that we
                     # need to process.

instana/recorder.py

@@ -4,7 +4,6 @@
 import socket
 import sys
 import threading as t
-import time
 
 import opentracing.ext.tags as ext
 from basictracer import Sampler, SpanRecorder
@@ -15,6 +14,7 @@
                         RabbitmqData, RedisData, RPCData, SDKData, SoapData,
                         SQLAlchemyData)
 from .log import logger
+from .util import every
 
 if sys.version_info.major is 2:
     import Queue as queue
@@ -37,6 +37,8 @@ class InstanaRecorder(SpanRecorder):
 
     queue = queue.Queue()
 
+    timer = None
+
     def __init__(self):
         super(InstanaRecorder, self).__init__()
 
@@ -50,13 +52,16 @@ def run(self):
     def report_spans(self):
         """ Periodically report the queued spans """
         logger.debug("Span reporting thread is now alive")
-        while 1:
+
+        def span_work():
             queue_size = self.queue.qsize()
             if queue_size > 0 and instana.singletons.agent.can_send():
                 response = instana.singletons.agent.report_traces(self.queued_spans())
                 if response:
                     logger.debug("reported %d spans" % queue_size)
-            time.sleep(1)
+            return True
+
+        every(2, span_work, "Span Reporting")
 
     def queue_size(self):
         """ Return the size of the queue; how may spans are queued, """

instana/util.py

@@ -231,5 +231,26 @@ def get_py_source(file):
         return response
 
 
+def every(delay, task, name):
+    """
+    Executes a task every `delay` seconds
+    
+    :param delay: the delay in seconds
+    :param task: the method to run.  The method should return False if you want the loop to stop.
+    :return: None
+    """
+    next_time = time.time() + delay
+
+    while True:
+        time.sleep(max(0, next_time - time.time()))
+        try:
+            if task() is False:
+                break
+        except Exception:
+            logger.debug("Problem while executing repetitive task: %s" % name, exc_info=True)
+
+        # skip tasks if we are behind schedule:
+        next_time += (time.time() - next_time) // delay * delay + delay
+
 # Used by get_py_source
 regexp_py = re.compile('\.py$')