[auto-discovery] re-pipe configurations after JMXFetch restart

agent.py

@@ -48,7 +48,7 @@
 )
 from daemon import AgentSupervisor, Daemon
 from emitter import http_emitter
-from jmxfetch import get_jmx_checks
+from jmxfetch import get_jmx_checks, JMXFetch
 
 # utils
 from utils.cloud_metadata import EC2
@@ -103,6 +103,7 @@ def __init__(self, pidfile, autorestart, start_event=True, in_developer_mode=Fal
         self.sd_backend = None
         self.supervisor_proxy = None
         self.sd_pipe = None
+        self.last_jmx_piped = None
 
     def _handle_sigterm(self, signum, frame):
         """Handles SIGTERM and SIGINT, which gracefully stops the agent."""
@@ -218,6 +219,11 @@ def info(cls, verbose=None):
         logging.getLogger().setLevel(logging.ERROR)
         return CollectorStatus.print_latest_status(verbose=verbose)
 
+    def sd_pipe_jmx_configs(self, hostname):
+        jmx_sd_configs = generate_jmx_configs(self._agentConfig, hostname)
+        if jmx_sd_configs:
+            self._submit_jmx_service_discovery(jmx_sd_configs)
+
     def run(self, config=None):
         """Main loop of the collector"""
 
@@ -258,7 +264,7 @@ def run(self, config=None):
         if self._agentConfig.get('service_discovery'):
             self.sd_backend = get_sd_backend(self._agentConfig)
 
-        if _is_affirmative(self._agentConfig.get('sd_jmx_enable', False)):
+        if self.sd_backend and _is_affirmative(self._agentConfig.get('sd_jmx_enable', False)):
             pipe_path = get_jmx_pipe_path()
             if Platform.is_windows():
                 pipe_name = pipe_path.format(pipename=SD_PIPE_NAME)
@@ -280,9 +286,7 @@ def run(self, config=None):
 
         # Load JMX configs if available
         if self._jmx_service_discovery_enabled:
-            jmx_sd_configs = generate_jmx_configs(self._agentConfig, hostname)
-            if jmx_sd_configs:
-                self._submit_jmx_service_discovery(jmx_sd_configs)
+            self.sd_pipe_jmx_configs(hostname)
 
         # Initialize the Collector
         self.collector = Collector(self._agentConfig, emitters, systemStats, hostname)
@@ -326,6 +330,16 @@ def run(self, config=None):
                 else:
                     self.reload_configs()
 
+            # JMXFetch restarts should prompt re-piping *all* JMX configs
+            if self._jmx_service_discovery_enabled and \
+                    (not self.reload_configs_flag or isinstance(self.reload_configs_flag, set)):
+                try:
+                    jmx_launch = JMXFetch._get_jmx_launchtime()
+                    if self.last_jmx_piped and self.last_jmx_piped < jmx_launch:
+                        self.sd_pipe_jmx_configs(hostname)
+                except Exception as e:
+                    log.debug("could not stat JMX lunch file: %s", e)
+
             # Do the work. Pass `configs_reloaded` to let the collector know if it needs to
             # look for the AgentMetrics check and pop it out.
             self.collector.run(checksd=self._checksd,
@@ -454,6 +468,7 @@ def _submit_jmx_service_discovery(self, jmx_sd_configs):
                 # JMX will unblock when it reads on the other end.
                 os.write(self.sd_pipe, buffer)
                 os.write(self.sd_pipe, SD_CONFIG_TERM)
+                self.last_jmx_piped = time.time()
         except Exception as e:
             log.exception("unable to submit YAML via pipe: %s", e)
         else:

jmxfetch.py

@@ -70,6 +70,8 @@
     'list_limited_attributes': "List attributes that do match one of your instances configuration but that are not being collected because it would exceed the number of metrics that can be collected",
     JMX_COLLECT_COMMAND: "Start the collection of metrics based on your current configuration and display them in the console"}
 
+JMX_LAUNCH_FILE = 'jmx.launch'
+
 LINK_TO_DOC = "See http://docs.datadoghq.com/integrations/java/ for more information"
 
 
@@ -336,6 +338,16 @@ def _start(self, path_to_java, java_run_opts, jmx_checks, command, reporter, too
             log.exception("Couldn't launch JMXFetch")
             raise
 
+    @staticmethod
+    def _get_jmx_launchtime():
+        fpath = os.path.join(get_jmx_pipe_path(), JMX_LAUNCH_FILE)
+        try:
+            _stat = os.stat(fpath)
+        except OSError as e:
+            raise e
+
+        return _stat.st_mtime
+
     @staticmethod
     def _is_jmx_check(check_config, check_name, checks_list):
         init_config = check_config.get('init_config', {}) or {}