Adding a separate dogstreams.d directory for dynamic dogstream config…

…uration. Also fixes #753. Supercedes #1550.

CHANGELOG.md

@@ -1,6 +1,16 @@
 Changes
 =======
 
+# 5.4.0 / Unreleased
+### Details
+https://github.com/DataDog/dd-agent/compare/5.3.0...5.4.0
+
+### Changes
+* [FEATURE] dogstreams config directive now supports wildcards in paths. [#753][] & [#1550][].
+* [FEATURE] Add a dogstreams.d directory which may contain YAML files. Makes
+  dynamically configuring dogstreams easier.
+
+
 # 5.3.2 / 04-29-2015
 **Debian only**
 

checks/datadog.py

@@ -12,7 +12,10 @@
 import modules
 from checks import LaconicFilter
 from checks.utils import TailFile
-from util import windows_friendly_colon_split
+from util import windows_friendly_colon_split, yLoader
+
+# 3rd party
+import yaml
 
 if hasattr('some string', 'partition'):
     def partition(s, sep):
@@ -60,7 +63,8 @@ def _instantiate_dogstreams(cls, logger, config, dogstreams_config):
         or
            <log path>:<module>:<parser function>
         """
-        dogstreams = []
+        # Load Dogstream objects from dogstreams.d.
+        dogstreams = cls._load_dogstreams_from_dir(logger, config)
         # Create a Dogstream object for each <dogstream value>
         for config_item in dogstreams_config.split(','):
             try:
@@ -84,6 +88,44 @@ def _instantiate_dogstreams(cls, logger, config, dogstreams_config):
                 logger.exception("Cannot build dogstream")
         return dogstreams
 
+    @classmethod
+    def _load_dogstreams_from_dir(cls, logger, config):
+        dogstreamsd_path = config.get('additional_dogstreamsd', None)
+        if not dogstreamsd_path:
+            return []
+
+        dogstreams = []
+        dogstream_yamls = glob.glob(os.path.join(dogstreamsd_path, "*.yaml"))
+        for dogstream_yaml in dogstream_yamls:
+            parsed = yaml.load(open(dogstream_yaml).read(), Loader=yLoader)
+            dogstreams += cls._dogstream_yaml_to_instance(logger, config, parsed)
+        return dogstreams
+
+    @classmethod
+    def _dogstream_yaml_to_instance(cls, logger, config, parsed):
+        if 'dogstreams' not in parsed:
+            return []
+
+        dogstreams = []
+        for stream in parsed['dogstreams']:
+            # There is only one key in a stream, and it is always its name.
+            stream_name = stream.keys()[0]
+            conf = stream[stream_name]['conf']
+            if 'path' not in conf:
+                logger.error('No path section for dogstream: %s' % stream_name)
+                continue
+            stream_path = conf['path']
+            parser_path = conf.get('parser', None)
+            for log_path in cls._get_dogstream_log_paths(stream_path):
+                dogstream = Dogstream.init(
+                    logger,
+                    log_path=log_path,
+                    parser_spec=parser_path,
+                    parser_args=None,
+                    config=config)
+                dogstreams.append(dogstream)
+        return dogstreams
+
     @classmethod
     def _get_dogstream_log_paths(cls, path):
         """

config.py

@@ -309,6 +309,7 @@ def get_config(parse_args=True, cfg_path=None, options=None):
         'version': get_version(),
         'watchdog': True,
         'additional_checksd': '/etc/dd-agent/checks.d/',
+        'additional_dogstreamsd': '/etc/dd-agent/dogstreams.d/',
         'bind_host': get_default_bind_host(),
         'statsd_metric_namespace': None,
         'utf8_decoding': False

datadog.conf.example

@@ -172,6 +172,9 @@ use_mount: no
 #     metric timestamp value key0=val0 key1=val1 ...
 #
 
+# Additional directory to look for Dogstreams
+# additional_dogstreamsd: /etc/dd-agent/dogstreams.d/
+
 # ========================================================================== #
 # Custom Emitters                                                            #
 # ========================================================================== #

dogstreams.d/nginx.yaml.example

@@ -0,0 +1,5 @@
+dogstreams:
+  - nginx:
+      conf:
+        path: /var/log/nginx/*
+        parser: /path/to/my/parsers_module.py:custom_parser

packaging/datadog-agent/win32/build.ps1

@@ -29,6 +29,10 @@ cp ..\..\..\checks.d\* install_files\checks.d
 mkdir install_files\conf.d
 cp ..\..\..\conf.d\* install_files\conf.d
 
+# Copy the dogstreams.d files into the install_files
+mkdir install_files\dogstreams.d
+cp ..\..\..\dogstreams.d\* install_files\dogstreams.d
+
 # Copy JMX Fetch into the install_files
 cp -R ..\..\..\dist\jmxfetch install_files\files\jmxfetch
 
@@ -50,10 +54,11 @@ cp ..\..\..\win32\status.html install_files\files
     heat dir install_files\files -gg -dr INSTALLDIR -t wix\files.xslt -var var.InstallFilesBins -cg files -o wix\files.wxs
     heat dir install_files\checks.d -gg -dr INSTALLDIR -var var.InstallFilesChecksD -cg checks.d -o wix\checksd.wxs
     heat dir install_files\conf.d -gg -dr APPLIDATIONDATADIRECTORY -t wix\confd.xslt -var var.InstallFilesConfD -cg conf.d -o wix\confd.wxs
+    heat dir install_files\conf.d -gg -dr APPLIDATIONDATADIRECTORY -t wix\confd.xslt -var var.InstallFilesConfD -cg dogstreams.d -o wix\dogstreamsd.wxs
 
-    # Create .wixobj files from agent.wxs, confd.wxs, checksd.wxs
-    $opts = '-dInstallFiles=install_files', '-dWixRoot=wix', '-dInstallFilesChecksD=install_files\checks.d', '-dInstallFilesConfD=install_files\conf.d', '-dInstallFilesBins=install_files\files', "-dAgentVersion=$version"
-    candle $opts wix\agent.wxs wix\checksd.wxs wix\confd.wxs wix\files.wxs -ext WixUIExtension -ext WixUtilExtension
+    # Create .wixobj files from agent.wxs, confd.wxs, checksd.wxs, dogstreamsd.wxs
+    $opts = '-dInstallFiles=install_files', '-dWixRoot=wix', '-dInstallFilesChecksD=install_files\checks.d', '-dInstallFilesConfD=install_files\conf.d', '-dInstallFilesDogstreamsD=install_files\dogstreams.d', '-dInstallFilesBins=install_files\files', "-dAgentVersion=$version"
+    candle $opts wix\agent.wxs wix\checksd.wxs wix\confd.wxs wix\dogstreamsd.wxs wix\files.wxs -ext WixUIExtension -ext WixUtilExtension
 
     # Light to create the msi
     light agent.wixobj checksd.wixobj confd.wixobj files.wixobj -o ..\..\..\build\ddagent.msi -ext WixUIExtension -ext WixUtilExtension
@@ -66,6 +71,7 @@ rm -r install_files\files\gohai
 rm install_files\files\*.*
 rm -r install_files\conf.d
 rm -r install_files\checks.d
+rm -r install_files\dogstreams.d
 rm -r install_files\Microsoft.VC90.CRT
 
 

tests/core/test_datadog.py

@@ -1,11 +1,14 @@
 # stdlib
 import logging
 import unittest
-from tempfile import NamedTemporaryFile, gettempdir
+from tempfile import NamedTemporaryFile, gettempdir, mkdtemp
 import re
+import os
 
-# project
 from checks.datadog import Dogstreams, EventDefaults
+from util import yLoader
+
+import yaml
 
 log = logging.getLogger('datadog.test')
 
@@ -200,25 +203,6 @@ def test_dogstream_ancient_function_plugin(self):
         plugdog = Dogstreams.init(self.logger, {'dogstreams': '{0}:{1}:parse_ancient_function_plugin'.format(self.log_file.name, __name__)})
         actual_output = plugdog.check(self.config, move_end=False)
 
-    def test_dogstream_log_path_globbing(self):
-        """Make sure that globbed dogstream logfile matching works."""
-        # Create a tmpfile to serve as a prefix for the other temporary 
-        # files we'll be globbing.
-        first_tmpfile = NamedTemporaryFile()
-        tmp_fprefix = os.path.basename(first_tmpfile.name)
-        all_tmp_filenames = set([first_tmpfile.name])
-        # We stick the file objects in here to avoid garbage collection (and
-        # tmpfile deletion). Not sure why this was happening, but it's working
-        # with this hack in.
-        avoid_gc = []
-        for i in range(3):
-            new_tmpfile = NamedTemporaryFile(prefix=tmp_fprefix)
-            all_tmp_filenames.add(new_tmpfile.name)
-            avoid_gc.append(new_tmpfile)
-        dogstream_glob = os.path.join(gettempdir(), tmp_fprefix + '*')
-        paths = Dogstreams._get_dogstream_log_paths(dogstream_glob)
-        self.assertEqual(set(paths), all_tmp_filenames)
-
     def test_dogstream_function_plugin(self):
         """Ensure that non-class-based stateful plugins work"""
         log_data = [
@@ -464,3 +448,88 @@ def test_supervisord_parser(self):
         dogstream = Dogstreams.init(self.logger, {'dogstreams': '%s:dogstream.supervisord_log:parse_supervisord' % self.log_file.name})
         actual_output = dogstream.check(self.config, move_end=False)
         self.assertEquals(expected_output, actual_output)
+
+
+class TestDogstreamLoading(TailTestCase):
+
+    def setUp(self):
+        TailTestCase.setUp(self)
+
+        self.dogstreamd_dir = mkdtemp()
+        self.config = {
+            'dogstreams': self.log_file.name,
+            'additional_dogstreamsd': self.dogstreamd_dir,
+        }
+
+        self.fictional_log = NamedTemporaryFile(
+            dir=self.dogstreamd_dir, suffix='.log', delete=False)
+        example_parser = NamedTemporaryFile(
+            dir=self.dogstreamd_dir, suffix='.py', delete=False)
+        example_parser.write(self._get_sample_dogstreamd_parser())
+        example_parser.close()
+        self.example_parser = example_parser
+
+        example_stream = NamedTemporaryFile(
+            dir=self.dogstreamd_dir, suffix='.yaml', delete=False)
+        example_stream.write(
+            self._get_sample_dogstreamd_yaml_string(example_parser.name))
+        example_stream.close()
+        self.example_stream = example_stream
+
+    def _get_sample_dogstreamd_parser(self):
+        return "def example_parser(logger, line): return 'Yay'"
+
+    def _get_sample_dogstreamd_yaml_string(self, parser_module_path):
+        return (
+            """
+            dogstreams:
+              - nginx:
+                  conf:
+                    path: {fake_log_path}
+                    parser: {fake_parser_module}:example_parser
+            """.format(
+                fake_log_path=os.path.join(self.dogstreamd_dir, "*.log"),
+                fake_parser_module=parser_module_path,
+            ))
+
+    def test_dogstream_log_path_globbing(self):
+        """Make sure that globbed dogstream logfile matching works."""
+        # Create a tmpfile to serve as a prefix for the other temporary
+        # files we'll be globbing.
+        first_tmpfile = NamedTemporaryFile()
+        tmp_fprefix = os.path.basename(first_tmpfile.name)
+        all_tmp_filenames = set([first_tmpfile.name])
+        # We stick the file objects in here to avoid garbage collection (and
+        # tmpfile deletion). Not sure why this was happening, but it's working
+        # with this hack in.
+        avoid_gc = []
+        for i in range(3):
+            new_tmpfile = NamedTemporaryFile(prefix=tmp_fprefix)
+            all_tmp_filenames.add(new_tmpfile.name)
+            avoid_gc.append(new_tmpfile)
+        dogstream_glob = os.path.join(gettempdir(), tmp_fprefix + '*')
+        paths = Dogstreams._get_dogstream_log_paths(dogstream_glob)
+        self.assertEqual(set(paths), all_tmp_filenames)
+
+    def test_dogstream_dir_loading(self):
+        """Tests loading a directory full of YAML dogstreams."""
+        dogstreams = Dogstreams._load_dogstreams_from_dir(self.logger, self.config)
+        self.assertEqual(len(dogstreams), 1)
+
+    def test_dogstream_yaml_to_instance(self):
+        """Tests the parsing of a dogstream YAML to a Dogstream instance."""
+        # Generate a test dogstream YAML config.
+        example_yaml = self._get_sample_dogstreamd_yaml_string(self.example_parser.name)
+        parsed = yaml.load(example_yaml, Loader=yLoader)
+        dogstreams = Dogstreams._dogstream_yaml_to_instance(
+            self.logger, self.config, parsed)
+        # Our example config was nginx.
+        nginx_s = dogstreams[0]
+        self.assertEqual(len(dogstreams), 1)
+        self.assertEqual(nginx_s.log_path, self.fictional_log.name)
+        self.assertEqual(nginx_s.parse_func.__name__, 'example_parser')
+
+
+if __name__ == '__main__':
+    logging.basicConfig(format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d %(message)s")
+    unittest.main()