PillarCache: reimplement using salt.cache

took the liberty of making it a proper subclass in the process. this now
uses the salt.cache infrastructure such that it can be driven by the
base cache driver or a different one if so desired. functionality should
be equivalent, including using the base bank=pillar key=minion_id for
merged pillar, such that minion_data_cache can take advantage of the
same cache. because we are updating the cache at the source, we no
longer need to update the cache in master/masterapi.

Author: mattp-

changelog/68030.changed.md

@@ -0,0 +1,3 @@
+PillarCache: reimplement using salt.cache
+fix minion data cache organization/move pillar and grains to dedicated cache banks
+salt.cache: allow cache.store() to set expires per key

changelog/68030.fixed.md

@@ -0,0 +1 @@
+salt.key: check_minion_cache performance optimization

salt/cache/__init__.py

@@ -12,6 +12,7 @@
 import salt.config
 import salt.loader
 import salt.syspaths
+from salt.utils.decorators import cached_property
 
 log = logging.getLogger(__name__)
 
@@ -59,30 +60,28 @@ class Cache:
 
     def __init__(self, opts, cachedir=None, **kwargs):
         self.opts = opts
-        if cachedir is None:
-            self.cachedir = opts.get("cachedir", salt.syspaths.CACHE_DIR)
+
+        if kwargs.get("driver"):
+            self.driver = kwargs["driver"]
         else:
-            self.cachedir = cachedir
-        self.driver = kwargs.get(
-            "driver", opts.get("cache", salt.config.DEFAULT_MASTER_OPTS["cache"])
+            self.driver = opts.get("cache", salt.config.DEFAULT_MASTER_OPTS["cache"])
+
+        self.cachedir = kwargs["cachedir"] = cachedir or opts.get(
+            "cachedir", salt.syspaths.CACHE_DIR
         )
         self._modules = None
         self._kwargs = kwargs
-        self._kwargs["cachedir"] = self.cachedir
-
-    def __lazy_init(self):
-        self._modules = salt.loader.cache(self.opts)
-        fun = f"{self.driver}.init_kwargs"
-        if fun in self.modules:
-            self._kwargs = self.modules[fun](self._kwargs)
-        else:
-            self._kwargs = {}
 
-    @property
+    @cached_property
     def modules(self):
-        if self._modules is None:
-            self.__lazy_init()
-        return self._modules
+        return salt.loader.cache(self.opts)
+
+    @cached_property
+    def kwargs(self):
+        try:
+            return self.modules[f"{self.driver}.init_kwargs"](self._kwargs)
+        except KeyError:
+            return {}
 
     def cache(self, bank, key, fun, loop_fun=None, **kwargs):
         """
@@ -148,7 +147,7 @@ def store(self, bank, key, data, expires=None):
         """
         fun = f"{self.driver}.store"
         try:
-            return self.modules[fun](bank, key, data, expires=expires, **self._kwargs)
+            return self.modules[fun](bank, key, data, expires=expires, **self.kwargs)
         except TypeError:
             # if the backing store doesnt natively support expiry, we handle it as a fallback
             if expires:
@@ -157,10 +156,10 @@ def store(self, bank, key, data, expires=None):
                 )
                 expires_at = int(expires_at.timestamp())
                 return self.modules[fun](
-                    bank, key, {"data": data, "_expires": expires_at}, **self._kwargs
+                    bank, key, {"data": data, "_expires": expires_at}, **self.kwargs
                 )
             else:
-                return self.modules[fun](bank, key, data, **self._kwargs)
+                return self.modules[fun](bank, key, data, **self.kwargs)
 
     def fetch(self, bank, key):
         """
@@ -184,7 +183,7 @@ def fetch(self, bank, key):
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.fetch"
-        ret = self.modules[fun](bank, key, **self._kwargs)
+        ret = self.modules[fun](bank, key, **self.kwargs)
 
         # handle fallback if necessary
         if isinstance(ret, dict) and set(ret.keys()) == {"data", "_expires"}:
@@ -218,7 +217,7 @@ def updated(self, bank, key):
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.updated"
-        return self.modules[fun](bank, key, **self._kwargs)
+        return self.modules[fun](bank, key, **self.kwargs)
 
     def flush(self, bank, key=None):
         """
@@ -239,7 +238,7 @@ def flush(self, bank, key=None):
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.flush"
-        return self.modules[fun](bank, key=key, **self._kwargs)
+        return self.modules[fun](bank, key=key, **self.kwargs)
 
     def list(self, bank):
         """
@@ -258,7 +257,7 @@ def list(self, bank):
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.list"
-        return self.modules[fun](bank, **self._kwargs)
+        return self.modules[fun](bank, **self.kwargs)
 
     def contains(self, bank, key=None):
         """
@@ -283,7 +282,7 @@ def contains(self, bank, key=None):
             in the cache backend (auth, permissions, etc).
         """
         fun = f"{self.driver}.contains"
-        return self.modules[fun](bank, key, **self._kwargs)
+        return self.modules[fun](bank, key, **self.kwargs)
 
 
 class MemCache(Cache):
@@ -336,10 +335,15 @@ def fetch(self, bank, key):
         if self.debug:
             self.call += 1
         now = time.time()
+        expires = None
         record = self.storage.pop((bank, key), None)
         # Have a cached value for the key
         if record is not None:
-            (created_at, expires, data) = record
+            if len(record) == 2:
+                (created_at, data) = record
+            elif len(record) == 3:
+                (created_at, expires, data) = record
+
             if (created_at + (expires or self.expire)) >= now:
                 if self.debug:
                     self.hit += 1
@@ -352,7 +356,7 @@ def fetch(self, bank, key):
                 # update atime and return
                 record[0] = now
                 self.storage[(bank, key)] = record
-                return record[1]
+                return data
 
         # Have no value for the key or value is expired
         data = super().fetch(bank, key)
@@ -361,12 +365,12 @@ def fetch(self, bank, key):
                 MemCache.__cleanup(self.expire)
             if len(self.storage) >= self.max:
                 self.storage.popitem(last=False)
-        self.storage[(bank, key)] = [now, data]
+        self.storage[(bank, key)] = [now, self.expire, data]
         return data
 
     def store(self, bank, key, data, expires=None):
         self.storage.pop((bank, key), None)
-        super().store(bank, key, data, expires)
+        super().store(bank, key, data, expires=expires)
         if len(self.storage) >= self.max:
             if self.cleanup:
                 MemCache.__cleanup(self.expire)

salt/cache/localfs.py

@@ -76,6 +76,7 @@ def fetch(bank, key, cachedir):
     inkey = False
     key_file = salt.utils.path.join(cachedir, os.path.normpath(bank), f"{key}.p")
     if not os.path.isfile(key_file):
+        log.debug('Cache file "%s" does not exist', key_file)
         # The bank includes the full filename, and the key is inside the file
         key_file = salt.utils.path.join(cachedir, os.path.normpath(bank) + ".p")
         inkey = True

salt/config/__init__.py

@@ -1035,6 +1035,8 @@ def _gather_buffer_space():
         "request_server_aes_session": int,
         # Minimum authentication protocol version to accept from minions
         "minimum_auth_version": int,
+        # optional cache driver for pillar cache
+        "pillar.cache_driver": (type(None), str),
     }
 )
 
@@ -1350,6 +1352,7 @@ def _gather_buffer_space():
         "encryption_algorithm": "OAEP-SHA1",
         "signing_algorithm": "PKCS1v15-SHA1",
         "keys.cache_driver": "localfs_key",
+        "pillar.cache_driver": None,
     }
 )
 
@@ -1715,6 +1718,7 @@ def _gather_buffer_space():
         "request_server_aes_session": 0,
         "request_server_ttl": 0,
         "minimum_auth_version": 3,
+        "pillar.cache_driver": None,
     }
 )
 

salt/daemons/masterapi.py

@@ -789,7 +789,6 @@ def _pillar(self, load):
         )
         data = pillar.compile_pillar()
         if self.opts.get("minion_data_cache", False):
-            self.cache.store("pillar", load["id"], data)
             self.cache.store("grains", load["id"], load["grains"])
             if self.opts.get("minion_data_cache_events") is True:
                 self.event.fire_event(

salt/master.py

@@ -724,16 +724,6 @@ def _pre_flight(self):
         if not self.opts["fileserver_backend"]:
             errors.append("No fileserver backends are configured")
 
-        # Check to see if we need to create a pillar cache dir
-        if self.opts["pillar_cache"] and not os.path.isdir(
-            os.path.join(self.opts["cachedir"], "pillar_cache")
-        ):
-            try:
-                with salt.utils.files.set_umask(0o077):
-                    os.mkdir(os.path.join(self.opts["cachedir"], "pillar_cache"))
-            except OSError:
-                pass
-
         if self.opts.get("git_pillar_verify_config", True):
             try:
                 git_pillars = [
@@ -1766,7 +1756,6 @@ def _pillar(self, load):
         self.fs_.update_opts()
         if self.opts.get("minion_data_cache", False):
             self.masterapi.cache.store("grains", load["id"], load["grains"])
-            self.masterapi.cache.store("pillar", load["id"], data)
 
             if self.opts.get("minion_data_cache_events") is True:
                 self.event.fire_event(