add the tests for the current problem and fixes for it

Author: jettero

hubblestack/extmods/modules/pulsar.py

@@ -32,7 +32,8 @@
 try:
     import pyinotify
     HAS_PYINOTIFY = True
-    DEFAULT_MASK = pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MODIFY
+    DEFAULT_MASK = pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_DELETE_SELF | pyinotify.IN_MODIFY
+    RM_WATCH_MASK = pyinotify.IN_DELETE | pyinotify.IN_DELETE_SELF | pyinotify.IN_IGNORED
     MASKS = {}
     for var in dir(pyinotify):
         if var.startswith('IN_'):
@@ -73,6 +74,18 @@ def _enqueue(revent):
     """
     __context__['pulsar.queue'].append(revent)
 
+def _maskname_filter(name):
+    """ deleting a directly watched file produces IN_DELETE_SELF (not
+        IN_DELETE) and also kicks up a an IN_IGNORED (whether you mask for it
+        or not) to indicate the file is nolonger watched.
+
+        We avoid returning IN_IGNORED if we can... but IN_DELETE_SELF is
+        corrected to IN_DELETE
+    """
+    if name == 'IN_DELETE_SELF':
+        return 'IN_DELETE'
+    return name
+
 class ConfigManager(object):
     _config = {}
     _last_update = 0
@@ -360,9 +373,9 @@ def watch(self, path, mask=None, **kw):
             kw['rec'] = kw.get('rec')
             if kw['rec'] is None:
                 kw['rec'] = pconf['recurse']
-            self.add_watch(path,mask,**kw)
-            log.debug('add-watch wd={0} path={1} watch_files={2} recurse={3}'.format(
-                self.watch_db.get(path), path, pconf['watch_files'], kw['rec']))
+            self.add_watch(path, mask, **kw)
+            log.debug('add-watch wd={0} path={1} watch_files={2} recurse={3} mask={4}'.format(
+                self.watch_db.get(path), path, pconf['watch_files'], kw['rec'], mask))
 
         if new_file: # process() says this is a new file
             self._add_recursed_file_watch(path)
@@ -772,10 +785,11 @@ def process(configfile='salt://hubblestack_pulsar/hubblestack_pulsar_config.yaml
 
             excludes = _preprocess_excludes( config[cpath].get('exclude') )
             _append = not excludes(pathname)
+
             if _append:
                 config_path = config['paths'][0]
                 pulsar_config = config_path[config_path.rfind('/') + 1:len(config_path)]
-                sub = { 'change': event.maskname,
+                sub = { 'change': _maskname_filter(event.maskname),
                         'path': abspath,  # goes to object_path in splunk
                         'tag':  dirname,  # goes to file_path in splunk
                         'name': basename, # goes to file_name in splunk
@@ -816,20 +830,19 @@ def process(configfile='salt://hubblestack_pulsar/hubblestack_pulsar_config.yaml
                     if os.path.isfile(pathname):
                         sub['size'] = os.path.getsize(pathname)
 
-
-
-                ret.append(sub)
+                if event.mask != pyinotify.IN_IGNORED:
+                    ret.append(sub)
 
                 if not event.mask & pyinotify.IN_ISDIR:
                     if event.mask & pyinotify.IN_CREATE:
                         watch_this = config[cpath].get('watch_new_files', False) \
                             or config[cpath].get('watch_files', False)
                         if watch_this:
                             if not excludes(pathname):
-                                log.debug("add file-watch path={0}".format(pathname))
+                                log.debug("add file-watch path={0} mask={1}".format(pathname,
+                                    pyinotify.IN_MODIFY))
                                 wm.watch(pathname, pyinotify.IN_MODIFY, new_file=True)
-
-                    elif event.mask & pyinotify.IN_DELETE:
+                    elif event.mask & RM_WATCH_MASK:
                         wm.rm_watch(pathname)
             else:
                 log.debug('Excluding {0} from event for {1}'.format(pathname, cpath))
@@ -847,22 +860,18 @@ def process(configfile='salt://hubblestack_pulsar/hubblestack_pulsar_config.yaml
                 continue
             if isinstance(config[path], dict):
                 mask = config[path].get('mask', DEFAULT_MASK)
-             ## commented out on 2019-08-05
-             #  watch_files = config[path].get('watch_files', DEFAULT_MASK)
-             ## note: it seems like the below was commented out because the above seems to be bugged
-             ## clearly watch_files should default to False, not DEFAULT_MASK
-             ## (aka true); that probably seemd like spurious watches from ##
-             ## duplications or whatever caused the below to be commented out.
-             ## Regardless, the above is probably wrong *and* seems unused.
-             ## commented out previous to 2019-08-05
-             #  if watch_files:
-             #      # we're going to get dup modify events if watch_files is set
-             #      # and we still monitor modify for the dir
-             #      a = mask & pyinotify.IN_MODIFY
-             #      if a:
-             #          log.debug("mask={0} -= mask & pyinotify.IN_MODIFY={1} ==> {2}".format(
-             #              mask, a, mask-a))
-             #          mask -= mask & pyinotify.IN_MODIFY
+                watch_files = config[path].get('watch_files', False)
+                if watch_files:
+                    # we're going to get dup modify events if watch_files is set
+                    # and we still monitor modify for the dir
+                    mask_and_modify = mask & pyinotify.IN_MODIFY
+                    if mask_and_modify:
+                        log.debug("mask={0} -= mask & pyinotify.IN_MODIFY={1}" \
+                            " ==> {2}".format(
+                                mask,
+                                mask_and_modify,
+                                mask-mask_and_modify))
+                        mask -= mask_and_modify
                 excludes = _preprocess_excludes( config[path].get('exclude') )
                 if isinstance(mask, list):
                     r_mask = 0
@@ -880,7 +889,32 @@ def process(configfile='salt://hubblestack_pulsar/hubblestack_pulsar_config.yaml
                 rec = False
                 auto_add = False
 
+            if os.path.isfile(path) and not wm.get_wd(path):
+                # We were not previously watching this file generate a fake
+                # IN_CREATE to announce this fact.  We'd like to only generate
+                # CREATE events when files are created, but we can't actually
+                # watch files that don't exist yet (not with inotify anyway).
+                # The kernel would rather be watching directories anyway.
+                #
+                # You might worry we'll get lots of spurious IN_CREATEs when
+                # the database is cleared or at startup or whatever.  We
+                # actually watch everyhthing from config at startup anyway; so
+                # we avoid these fake IN_CREATE events at startup. They only
+                # happen when we add a watch during update, which means the
+                # file really is new since the last time we thought about it
+                # (aka the last time we ran the process() function).
+                _, abspath, dirname, basename = cm.format_path(path)
+                config_path = config['paths'][0]
+                pulsar_config = config_path[config_path.rfind('/') + 1:len(config_path)]
+                fake_sub = { 'change': 'IN_CREATE',
+                        'path': abspath,  # goes to object_path in splunk
+                        'tag':  dirname,  # goes to file_path in splunk
+                        'name': basename, # goes to file_name in splunk
+                        'pulsar_config': pulsar_config}
+                ret.append(fake_sub)
+
             wm.watch(path, mask, rec=rec, auto_add=auto_add, exclude_filter=excludes)
+
         dt.fin()
         dt.mark('prune_watches')
         wm.prune()

tests/unittests/test_pulsar.py

@@ -183,6 +183,11 @@ def config_get(_, default):
     def process(self):
         self.events.extend([ "{change}({path})".format(**x) for x in pulsar.process() ])
 
+    def get_clear_events(self):
+        ret = self.events
+        self.events = list()
+        return ret
+
     def nuke_tdir(self):
         if os.path.isdir(self.tdir):
             shutil.rmtree(self.tdir)
@@ -402,55 +407,92 @@ def test_pruning_watch_files_then_nothing(self):
         assert set1 == set([self.atdir, fname1, fname2, self.atfile])
         assert set2 == set()
 
-    def test_fim_single_file(self):
-        config = {self.atfile: {}}
+    def test_watch_files_events(self):
+        config = {self.atdir: { 'watch_files': True }}
         self.reset(**config)
         self.mk_tdir_and_write_tfile()
-        self.watch_manager.watch(self.tfile)
 
         set0 = set(self.watch_manager.watch_db)
-        levents0 = len(self.events)
-        # we should be watching 1 file, no events
 
-        self.process()
+        pulsar.process()
         set1 = set(self.watch_manager.watch_db)
         levents1 = len(self.events)
-        # we should be watching 1 file, no events
+        assert set0 == set()
+        assert set1 == set([self.atdir, self.atfile])
+        assert levents1 == 0
 
         with open(self.atfile, 'a') as fh:
             fh.write('supz\n')
-
         self.process()
-        set2 = set(self.watch_manager.watch_db)
-        levents2 = len(self.events)
-        # we should still be watching 1 file, 1 event
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set1
+        assert events_ == ['IN_MODIFY({})'.format(self.atfile)]
 
         os.unlink(self.atfile)
+        self.process()
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set([self.atdir])
+        assert events_ == ['IN_DELETE({})'.format(self.atfile)]
+
+        with open(self.atfile, 'a') as fh:
+            fh.write('supz\n')
+        self.process()
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set1
+        assert events_ == ['IN_CREATE({})'.format(self.atfile)]
 
+        with open(self.atfile, 'a') as fh:
+            fh.write('supz\n')
         self.process()
-        set3 = set(self.watch_manager.watch_db)
-        levents3 = len(self.events)
-        # we should now be watching 0 files, 2 events
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set1
+        assert events_ == ['IN_MODIFY({})'.format(self.atfile)]
 
-        assert levents0 == 0
-        assert levents1 == 0
-        assert levents2 == 1
-        assert levents3 == 2
-        assert set0 == set([self.atfile])
+    def test_single_file_events(self):
+        config = {self.atfile: dict()}
+        self.reset(**config)
+        self.mk_tdir_and_write_tfile()
+
+        set0 = set(self.watch_manager.watch_db)
+        assert set0 == set()
+
+        pulsar.process()
+        set1 = set(self.watch_manager.watch_db)
+        levents1 = len(self.events)
         assert set1 == set([self.atfile])
-        assert set2 == set([self.atfile])
-        assert set3 == set()
-        ### BASELINE ###
+        assert levents1 == 0
 
-        # the relevant connundrum: if we put the file back, this config should
-        # somehow know to re-watch the atfile
-        # at the time of this writing, this test fails
         with open(self.atfile, 'a') as fh:
             fh.write('supz\n')
+        self.process()
+        set2 = set(self.watch_manager.watch_db)
+        events2 = self.get_clear_events()
+        assert set2 == set1
+        assert events2 == ['IN_MODIFY({})'.format(self.atfile)]
 
+        os.unlink(self.atfile)
         self.process()
-        set4 = set(self.watch_manager.watch_db)
-        levents4 = len(self.events)
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set() # this is DELETE_SELF now (technically)
+        assert events_ == ['IN_DELETE({})'.format(self.atfile)]
 
-        assert set4 == set([self.atfile])
-        assert levents4 == 3 # XXX: 4? CREATE? CREATE_MODIFY? CREATE+MODIFY??
+        with open(self.atfile, 'a') as fh:
+            fh.write('supz\n')
+        self.process()
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set1
+        assert events_ == ['IN_CREATE({})'.format(self.atfile)]
+
+        with open(self.atfile, 'a') as fh:
+            fh.write('supz\n')
+        self.process()
+        set_ = set(self.watch_manager.watch_db)
+        events_ = self.get_clear_events()
+        assert set_ == set1
+        assert events_ == ['IN_MODIFY({})'.format(self.atfile)]