clear all loader caches with new dir for plugin (#53413)

* clear all loader caches with new dir for plugin

  fixes #17078

* added toggle to revert behaviour

changelogs/fragments/fix_plugin_loader_cache.yml

@@ -0,0 +1,3 @@
+bugfixes:
+    - clear all caches in plugin loader for a plugin type when adding new paths,
+       otherwise new versions of already loaded plugin won't be discovered

lib/ansible/config/base.yml

@@ -1608,6 +1608,14 @@ INJECT_FACTS_AS_VARS:
   - {key: inject_facts_as_vars, section: defaults}
   type: boolean
   version_added: "2.5"
+OLD_PLUGIN_CACHE_CLEARING:
+  description: Previouslly Ansible would only clear some of the plugin loading caches when loading new roles, this led to some behaviours in which a plugin loaded in prevoius plays would be unexpectedly 'sticky'. This setting allows to return to that behaviour.
+  env: [{name: ANSIBLE_OLD_PLUGIN_CACHE_CLEAR}]
+  ini:
+  - {key: old_plugin_cache_clear, section: defaults}
+  type: boolean
+  default: False
+  version_added: "2.8"
 PARAMIKO_HOST_KEY_AUTO_ADD:
   # TODO: move to plugin
   default: False

lib/ansible/plugins/loader.py

@@ -113,13 +113,32 @@ class PluginLoader:
         if class_name not in PLUGIN_PATH_CACHE:
             PLUGIN_PATH_CACHE[class_name] = defaultdict(dict)
 
+        # hold dirs added at runtime outside of config
+        self._extra_dirs = []
+
+        # caches
         self._module_cache = MODULE_CACHE[class_name]
         self._paths = PATH_CACHE[class_name]
         self._plugin_path_cache = PLUGIN_PATH_CACHE[class_name]
 
-        self._extra_dirs = []
         self._searched_paths = set()
 
+    def _clear_caches(self):
+
+        if C.OLD_PLUGIN_CACHE_CLEARING:
+            self._paths = None
+        else:
+            # reset global caches
+            MODULE_CACHE[self.class_name] = {}
+            PATH_CACHE[self.class_name] = None
+            PLUGIN_PATH_CACHE[self.class_name] = defaultdict(dict)
+
+            # reset internal caches
+            self._module_cache = MODULE_CACHE[self.class_name]
+            self._paths = PATH_CACHE[self.class_name]
+            self._plugin_path_cache = PLUGIN_PATH_CACHE[self.class_name]
+            self._searched_paths = set()
+
     def __setstate__(self, data):
         '''
         Deserializer.
@@ -276,7 +295,7 @@ class PluginLoader:
             if directory not in self._extra_dirs:
                 # append the directory and invalidate the path cache
                 self._extra_dirs.append(directory)
-                self._paths = None
+                self._clear_caches()
                 display.debug('Added %s to loader search path' % (directory))
 
     def _find_plugin(self, name, mod_type='', ignore_deprecated=False, check_aliases=False):