diff --git a/changelogs/fragments/61078-vars-plugin-whitelist-and-execution-settings.yaml b/changelogs/fragments/61078-vars-plugin-whitelist-and-execution-settings.yaml
new file mode 100644
index 00000000000..f4b447b7530
--- /dev/null
+++ b/changelogs/fragments/61078-vars-plugin-whitelist-and-execution-settings.yaml
@@ -0,0 +1,6 @@
+---
+minor_changes:
+  - vars plugins - Support vars plugins in collections by adding the ability to whitelist plugins.
+  - host_group_vars plugin - Require whitelisting and whitelist by default.
+  - Add a global toggle to control when vars plugins are executed (per task by default for backward compatibility or after importing inventory).
+  - Add a per-plugin stage option to override the global toggle to control the execution of individual vars plugins (per task, after inventory, or both).
diff --git a/docs/docsite/rst/dev_guide/developing_plugins.rst b/docs/docsite/rst/dev_guide/developing_plugins.rst
index 3c8c19d3339..2f6212586ce 100644
--- a/docs/docsite/rst/dev_guide/developing_plugins.rst
+++ b/docs/docsite/rst/dev_guide/developing_plugins.rst
@@ -442,9 +442,38 @@ The parameters are:
  * path: this is 'directory data' for every inventory source and the current play's playbook directory, so they can search for data in reference to them. ``get_vars`` will be called at least once per available path.
  * entities: these are host or group names that are pertinent to the variables needed. The plugin will get called once for hosts and again for groups.
 
-This ``get vars`` method just needs to return a dictionary structure with the variables.
+This ``get_vars`` method just needs to return a dictionary structure with the variables.
 
-Since Ansible version 2.4, vars plugins only execute as needed when preparing to execute a task. This avoids the costly 'always execute' behavior that occurred during inventory construction in older versions of Ansible.
+Since Ansible version 2.4, vars plugins only execute as needed when preparing to execute a task. This avoids the costly 'always execute' behavior that occurred during inventory construction in older versions of Ansible. Since Ansible version 2.10, vars plugin execution can be toggled by the user to run when preparing to execute a task or after importing an inventory source.
+
+Since Ansible 2.10, vars plugins can require whitelisting. Vars plugins that don't require whitelisting will run by default. To require whitelisting for your plugin set the class variable ``REQUIRES_WHITELIST``:
+
+.. code-block:: python
+
+    class VarsModule(BaseVarsPlugin):
+        REQUIRES_WHITELIST = True
+
+Include the ``vars_plugin_staging`` documentation fragment to allow users to determine when vars plugins run.
+
+.. code-block:: python
+
+    DOCUMENTATION = '''
+        vars: custom_hostvars
+        version_added: "2.10"
+        short_description: Load custom host vars
+        description: Load custom host vars
+        options:
+          stage:
+            ini:
+              - key: stage
+                section: vars_custom_hostvars
+            env:
+              - name: ANSIBLE_VARS_PLUGIN_STAGE
+        extends_documentation_fragment:
+          - vars_plugin_staging
+    '''
+
+Also since Ansible 2.10, vars plugins can reside in collections. Vars plugins in collections must require whitelisting to be functional.
 
 For example vars plugins, see the source code for the `vars plugins included with Ansible Core
 <https://github.com/ansible/ansible/tree/devel/lib/ansible/plugins/vars>`_.
diff --git a/docs/docsite/rst/plugins/vars.rst b/docs/docsite/rst/plugins/vars.rst
index 8596fcc9c98..dc57aa3837c 100644
--- a/docs/docsite/rst/plugins/vars.rst
+++ b/docs/docsite/rst/plugins/vars.rst
@@ -19,16 +19,34 @@ The :ref:`host_group_vars <host_group_vars_vars>` plugin shipped with Ansible en
 Enabling vars plugins
 ---------------------
 
-You can activate a custom vars plugin by either dropping it into a ``vars_plugins`` directory adjacent to your play,  inside a role, or by putting it in one of the directory sources configured in :ref:`ansible.cfg <ansible_configuration_settings>`.
+You can activate a custom vars plugin by either dropping it into a ``vars_plugins`` directory adjacent to your play, inside a role, or by putting it in one of the directory sources configured in :ref:`ansible.cfg <ansible_configuration_settings>`.
 
+Starting in Ansible 2.10, vars plugins can require whitelisting rather than running by default. To enable a plugin that requires whitelisting set ``vars_plugins_enabled`` in the ``defaults`` section of :ref:`ansible.cfg <ansible_configuration_settings>` or set the ``ANSIBLE_VARS_ENABLED`` environment variable to the list of vars plugins you want to execute. By default, the :ref:`host_group_vars <host_group_vars_vars>` plugin shipped with Ansible is whitelisted.
+
+Starting in Ansible 2.10, you can use vars plugins in collections. All vars plugins in collections require whitelisting and need to use the fully qualified collection name in the format ``namespace.collection_name.vars_plugin_name``.
+
+.. code-block:: yaml
+
+    [defaults]
+    vars_plugins_enabled = host_group_vars,namespace.collection_name.vars_plugin_name
 
 .. _using_vars:
 
 Using vars plugins
 ------------------
 
-Vars plugins are used automatically after they are enabled.
+By default, vars plugins are used on demand automatically after they are enabled.
+
+Starting in Ansible 2.10, vars plugins can be made to run at specific times.
+
+The global setting ``RUN_VARS_PLUGINS`` can be set in ``ansible.cfg`` using ``run_vars_plugins`` in the ``defaults`` section or by the ``ANSIBLE_RUN_VARS_PLUGINS`` environment variable. The default option, ``demand``, runs any enabled vars plugins relative to inventory sources whenever variables are demanded by tasks. You can use the option ``start`` to run any enabled vars plugins relative to inventory sources after importing that inventory source instead.
+
+You can also control vars plugin execution on a per-plugin basis for vars plugins that support the ``stage`` option. To run the :ref:`host_group_vars <host_group_vars_vars>` plugin after importing inventory you can add the following to :ref:`ansible.cfg <ansible_configuration_settings>`:
+
+.. code-block:: ini
 
+    [vars_host_group_vars]
+    stage = inventory
 
 .. _vars_plugin_list:
 
diff --git a/lib/ansible/cli/inventory.py b/lib/ansible/cli/inventory.py
index fae1a4f9dab..18356eb26f9 100644
--- a/lib/ansible/cli/inventory.py
+++ b/lib/ansible/cli/inventory.py
@@ -18,6 +18,7 @@ from ansible.module_utils._text import to_bytes, to_native
 from ansible.plugins.loader import vars_loader
 from ansible.utils.vars import combine_vars
 from ansible.utils.display import Display
+from ansible.vars.plugins import get_vars_from_inventory_sources
 
 display = Display()
 
@@ -184,41 +185,12 @@ class InventoryCLI(CLI):
 
         return results
 
-    # FIXME: refactor to use same for VM
-    def get_plugin_vars(self, path, entity):
-
-        data = {}
-
-        def _get_plugin_vars(plugin, path, entities):
-            data = {}
-            try:
-                data = plugin.get_vars(self.loader, path, entity)
-            except AttributeError:
-                try:
-                    if isinstance(entity, Host):
-                        data = combine_vars(data, plugin.get_host_vars(entity.name))
-                    else:
-                        data = combine_vars(data, plugin.get_group_vars(entity.name))
-                except AttributeError:
-                    if hasattr(plugin, 'run'):
-                        raise AnsibleError("Cannot use v1 type vars plugin %s from %s" % (plugin._load_name, plugin._original_path))
-                    else:
-                        raise AnsibleError("Invalid vars plugin %s from %s" % (plugin._load_name, plugin._original_path))
-            return data
-
-        for plugin in vars_loader.all():
-            data = combine_vars(data, _get_plugin_vars(plugin, path, entity))
-
-        return data
-
     def _get_group_variables(self, group):
 
         # get info from inventory source
         res = group.get_vars()
 
-        # FIXME: add switch to skip vars plugins, add vars plugin info
-        for inventory_dir in self.inventory._sources:
-            res = combine_vars(res, self.get_plugin_vars(inventory_dir, group))
+        res = combine_vars(res, get_vars_from_inventory_sources(self.loader, self.inventory._sources, [group], 'inventory'))
 
         if group.priority != 1:
             res['ansible_group_priority'] = group.priority
@@ -231,12 +203,10 @@ class InventoryCLI(CLI):
             # only get vars defined directly host
             hostvars = host.get_vars()
 
-            # FIXME: add switch to skip vars plugins, add vars plugin info
-            for inventory_dir in self.inventory._sources:
-                hostvars = combine_vars(hostvars, self.get_plugin_vars(inventory_dir, host))
+            hostvars = combine_vars(hostvars, get_vars_from_inventory_sources(self.loader, self.inventory._sources, [host], 'inventory'))
         else:
             # get all vars flattened by host, but skip magic hostvars
-            hostvars = self.vm.get_vars(host=host, include_hostvars=False)
+            hostvars = self.vm.get_vars(host=host, include_hostvars=False, stage='inventory')
 
         return self._remove_internal(hostvars)
 
diff --git a/lib/ansible/config/base.yml b/lib/ansible/config/base.yml
index 1a772f0423a..2539e0bf210 100644
--- a/lib/ansible/config/base.yml
+++ b/lib/ansible/config/base.yml
@@ -1757,6 +1757,19 @@ RETRY_FILES_SAVE_PATH:
   ini:
   - {key: retry_files_save_path, section: defaults}
   type: path
+RUN_VARS_PLUGINS:
+  name: When should vars plugins run relative to inventory
+  default: demand
+  description:
+    - This setting can be used to optimize vars_plugin usage depending on user's inventory size and play selection.
+    - Setting to C(demand) will run vars_plugins relative to inventory sources anytime vars are 'demanded' by tasks.
+    - Setting to C(start) will run vars_plugins relative to inventory sources after importing that inventory source.
+  env: [{name: ANSIBLE_RUN_VARS_PLUGINS}]
+  ini:
+  - {key: run_vars_plugins, section: defaults}
+  type: str
+  choices: ['demand', 'start']
+  version_added: "2.10"
 SHOW_CUSTOM_STATS:
   name: Display custom stats
   default: False
@@ -1811,6 +1824,15 @@ USE_PERSISTENT_CONNECTIONS:
   ini:
   - {key: use_persistent_connections, section: defaults}
   type: boolean
+VARIABLE_PLUGINS_ENABLED:
+  name: Vars plugin whitelist
+  default: ['host_group_vars']
+  description: Whitelist for variable plugins that require it.
+  env: [{name: ANSIBLE_VARS_ENABLED}]
+  ini:
+  - {key: vars_plugins_enabled, section: defaults}
+  type: list
+  version_added: "2.10"
 VARIABLE_PRECEDENCE:
   name: Group variable precedence
   default: ['all_inventory', 'groups_inventory', 'all_plugins_inventory', 'all_plugins_play', 'groups_plugins_inventory', 'groups_plugins_play']
diff --git a/lib/ansible/constants.py b/lib/ansible/constants.py
index 051d7258cfb..34446691b4d 100644
--- a/lib/ansible/constants.py
+++ b/lib/ansible/constants.py
@@ -102,9 +102,9 @@ DEFAULT_PASSWORD_CHARS = to_text(ascii_letters + digits + ".,:-_", errors='stric
 DEFAULT_REMOTE_PASS = None
 DEFAULT_SUBSET = None
 # FIXME: expand to other plugins, but never doc fragments
-CONFIGURABLE_PLUGINS = ('become', 'cache', 'callback', 'cliconf', 'connection', 'httpapi', 'inventory', 'lookup', 'shell')
+CONFIGURABLE_PLUGINS = ('become', 'cache', 'callback', 'cliconf', 'connection', 'httpapi', 'inventory', 'lookup', 'shell', 'vars')
 # NOTE: always update the docs/docsite/Makefile to match
-DOCUMENTABLE_PLUGINS = CONFIGURABLE_PLUGINS + ('module', 'strategy', 'vars')
+DOCUMENTABLE_PLUGINS = CONFIGURABLE_PLUGINS + ('module', 'strategy')
 IGNORE_FILES = ("COPYING", "CONTRIBUTING", "LICENSE", "README", "VERSION", "GUIDELINES")  # ignore during module search
 INTERNAL_RESULT_KEYS = ('add_host', 'add_group')
 LOCALHOST = ('127.0.0.1', 'localhost', '::1')
diff --git a/lib/ansible/inventory/manager.py b/lib/ansible/inventory/manager.py
index 72ffa6c8d70..4bb6fe4edc5 100644
--- a/lib/ansible/inventory/manager.py
+++ b/lib/ansible/inventory/manager.py
@@ -39,6 +39,8 @@ from ansible.plugins.loader import inventory_loader
 from ansible.utils.helpers import deduplicate_list
 from ansible.utils.path import unfrackpath
 from ansible.utils.display import Display
+from ansible.utils.vars import combine_vars
+from ansible.vars.plugins import get_vars_from_inventory_sources
 
 display = Display()
 
@@ -230,6 +232,11 @@ class InventoryManager(object):
             else:
                 display.warning("No inventory was parsed, only implicit localhost is available")
 
+        for group in self.groups.values():
+            group.vars = combine_vars(group.vars, get_vars_from_inventory_sources(self._loader, self._sources, [group], 'inventory'))
+        for host in self.hosts.values():
+            host.vars = combine_vars(host.vars, get_vars_from_inventory_sources(self._loader, self._sources, [host], 'inventory'))
+
     def parse_source(self, source, cache=False):
         ''' Generate or update inventory for the source provided '''
 
diff --git a/lib/ansible/plugins/__init__.py b/lib/ansible/plugins/__init__.py
index fbc386f492a..73857f45c5c 100644
--- a/lib/ansible/plugins/__init__.py
+++ b/lib/ansible/plugins/__init__.py
@@ -79,6 +79,11 @@ class AnsiblePlugin(with_metaclass(ABCMeta, object)):
         if self.allow_extras and var_options and '_extras' in var_options:
             self.set_option('_extras', var_options['_extras'])
 
+    def has_option(self, option):
+        if not self._options:
+            self.set_options()
+        return option in self._options
+
     def _check_required(self):
         # FIXME: standardize required check based on config
         pass
diff --git a/lib/ansible/plugins/doc_fragments/vars_plugin_staging.py b/lib/ansible/plugins/doc_fragments/vars_plugin_staging.py
new file mode 100644
index 00000000000..b2da29c4942
--- /dev/null
+++ b/lib/ansible/plugins/doc_fragments/vars_plugin_staging.py
@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2019,  Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+class ModuleDocFragment(object):
+
+    DOCUMENTATION = r'''
+options:
+  stage:
+    description:
+      - Control when this vars plugin may be executed.
+      - Setting this option to C(all) will run the vars plugin after importing inventory and whenever it is demanded by a task.
+      - Setting this option to C(task) will only run the vars plugin whenever it is demanded by a task.
+      - Setting this option to C(inventory) will only run the vars plugin after parsing inventory.
+      - If this option is omitted, the global I(RUN_VARS_PLUGINS) configuration is used to determine when to execute the vars plugin.
+    choices: ['all', 'task', 'inventory']
+    version_added: "2.10"
+    type: str
+'''
diff --git a/lib/ansible/plugins/vars/__init__.py b/lib/ansible/plugins/vars/__init__.py
index 668c92f5fe3..2a7bafd9ad9 100644
--- a/lib/ansible/plugins/vars/__init__.py
+++ b/lib/ansible/plugins/vars/__init__.py
@@ -18,13 +18,14 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
+from ansible.plugins import AnsiblePlugin
 from ansible.utils.path import basedir
 from ansible.utils.display import Display
 
 display = Display()
 
 
-class BaseVarsPlugin(object):
+class BaseVarsPlugin(AnsiblePlugin):
 
     """
     Loads variables for groups and/or hosts
@@ -32,6 +33,7 @@ class BaseVarsPlugin(object):
 
     def __init__(self):
         """ constructor """
+        super(BaseVarsPlugin, self).__init__()
         self._display = display
 
     def get_vars(self, loader, path, entities):
diff --git a/lib/ansible/plugins/vars/host_group_vars.py b/lib/ansible/plugins/vars/host_group_vars.py
index 5a412e215d2..377a77de7a4 100644
--- a/lib/ansible/plugins/vars/host_group_vars.py
+++ b/lib/ansible/plugins/vars/host_group_vars.py
@@ -22,12 +22,21 @@ DOCUMENTATION = '''
     vars: host_group_vars
     version_added: "2.4"
     short_description: In charge of loading group_vars and host_vars
+    requirements:
+        - whitelist in configuration
     description:
         - Loads YAML vars into corresponding groups/hosts in group_vars/ and host_vars/ directories.
         - Files are restricted by extension to one of .yaml, .json, .yml or no extension.
         - Hidden (starting with '.') and backup (ending with '~') files and directories are ignored.
         - Only applies to inventory sources that are existing paths.
+        - Starting in 2.10, this plugin requires whitelisting and is whitelisted by default.
     options:
+      stage:
+        ini:
+          - key: stage
+            section: vars_host_group_vars
+        env:
+          - name: ANSIBLE_VARS_PLUGIN_STAGE
       _valid_extensions:
         default: [".yml", ".yaml", ".json"]
         description:
@@ -39,6 +48,8 @@ DOCUMENTATION = '''
           - section: yaml_valid_extensions
             key: defaults
         type: list
+    extends_documentation_fragment:
+      - vars_plugin_staging
 '''
 
 import os
@@ -55,6 +66,8 @@ FOUND = {}
 
 class VarsModule(BaseVarsPlugin):
 
+    REQUIRES_WHITELIST = True
+
     def get_vars(self, loader, path, entities, cache=True):
         ''' parses the inventory file '''
 
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index d529047811c..21fd5814ee5 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -35,10 +35,10 @@ from ansible import constants as C
 from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVariable, AnsibleFileNotFound, AnsibleAssertionError, AnsibleTemplateError
 from ansible.inventory.host import Host
 from ansible.inventory.helpers import sort_groups, get_group_vars
-from ansible.module_utils._text import to_bytes, to_text
+from ansible.module_utils._text import to_text
 from ansible.module_utils.common._collections_compat import Mapping, MutableMapping, Sequence
 from ansible.module_utils.six import iteritems, text_type, string_types
-from ansible.plugins.loader import lookup_loader, vars_loader
+from ansible.plugins.loader import lookup_loader
 from ansible.vars.fact_cache import FactCache
 from ansible.template import Templar
 from ansible.utils.display import Display
@@ -46,6 +46,7 @@ from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.vars import combine_vars, load_extra_vars, load_options_vars
 from ansible.utils.unsafe_proxy import wrap_var
 from ansible.vars.clean import namespace_facts, clean_facts
+from ansible.vars.plugins import get_vars_from_inventory_sources, get_vars_from_path
 
 display = Display()
 
@@ -141,7 +142,7 @@ class VariableManager:
         self._inventory = inventory
 
     def get_vars(self, play=None, host=None, task=None, include_hostvars=True, include_delegate_to=True, use_cache=True,
-                 _hosts=None, _hosts_all=None):
+                 _hosts=None, _hosts_all=None, stage='task'):
         '''
         Returns the variables, with optional "context" given via the parameters
         for the play, host, and task (which could possibly result in different
@@ -231,25 +232,13 @@ class VariableManager:
             # internal fuctions that actually do the work
             def _plugins_inventory(entities):
                 ''' merges all entities by inventory source '''
-                data = {}
-                for inventory_dir in self._inventory._sources:
-                    if ',' in inventory_dir and not os.path.exists(inventory_dir):  # skip host lists
-                        continue
-                    elif not os.path.isdir(to_bytes(inventory_dir)):  # always pass 'inventory directory'
-                        inventory_dir = os.path.dirname(inventory_dir)
-
-                    for plugin in vars_loader.all():
-
-                        data = combine_vars(data, _get_plugin_vars(plugin, inventory_dir, entities))
-                return data
+                return get_vars_from_inventory_sources(self._loader, self._inventory._sources, entities, stage)
 
             def _plugins_play(entities):
                 ''' merges all entities adjacent to play '''
                 data = {}
-                for plugin in vars_loader.all():
-
-                    for path in basedirs:
-                        data = combine_vars(data, _get_plugin_vars(plugin, path, entities))
+                for path in basedirs:
+                    data = combine_vars(data, get_vars_from_path(self._loader, path, entities, stage))
                 return data
 
             # configurable functions that are sortable via config, rememer to add to _ALLOWED if expanding this list
diff --git a/lib/ansible/vars/plugins.py b/lib/ansible/vars/plugins.py
new file mode 100644
index 00000000000..1411129d4f0
--- /dev/null
+++ b/lib/ansible/vars/plugins.py
@@ -0,0 +1,95 @@
+# Copyright (c) 2018 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+
+from ansible import constants as C
+from ansible.errors import AnsibleError
+from ansible.inventory.host import Host
+from ansible.module_utils._text import to_bytes
+from ansible.plugins.loader import vars_loader
+from ansible.utils.collection_loader import AnsibleCollectionRef
+from ansible.utils.display import Display
+from ansible.utils.vars import combine_vars
+
+display = Display()
+
+
+def get_plugin_vars(loader, plugin, path, entities):
+
+    data = {}
+    try:
+        data = plugin.get_vars(loader, path, entities)
+    except AttributeError:
+        try:
+            for entity in entities:
+                if isinstance(entity, Host):
+                    data.update(plugin.get_host_vars(entity.name))
+                else:
+                    data.update(plugin.get_group_vars(entity.name))
+        except AttributeError:
+            if hasattr(plugin, 'run'):
+                raise AnsibleError("Cannot use v1 type vars plugin %s from %s" % (plugin._load_name, plugin._original_path))
+            else:
+                raise AnsibleError("Invalid vars plugin %s from %s" % (plugin._load_name, plugin._original_path))
+    return data
+
+
+def get_vars_from_path(loader, path, entities, stage):
+
+    data = {}
+
+    vars_plugin_list = list(vars_loader.all())
+    for plugin_name in C.VARIABLE_PLUGINS_ENABLED:
+        if AnsibleCollectionRef.is_valid_fqcr(plugin_name):
+            vars_plugin = vars_loader.get(plugin_name)
+            if vars_plugin is None:
+                # Error if there's no play directory or the name is wrong?
+                continue
+            if vars_plugin not in vars_plugin_list:
+                vars_plugin_list.append(vars_plugin)
+
+    for plugin in vars_plugin_list:
+        if plugin._load_name not in C.VARIABLE_PLUGINS_ENABLED and getattr(plugin, 'REQUIRES_WHITELIST', False):
+            # 2.x plugins shipped with ansible should require whitelisting, older or non shipped should load automatically
+            continue
+
+        has_stage = hasattr(plugin, 'get_option') and plugin.has_option('stage')
+
+        # if a plugin-specific setting has not been provided, use the global setting
+        # older/non shipped plugins that don't support the plugin-specific setting should also use the global setting
+        use_global = (has_stage and plugin.get_option('stage') is None) or not has_stage
+
+        if use_global:
+            if C.RUN_VARS_PLUGINS == 'demand' and stage == 'inventory':
+                continue
+            elif C.RUN_VARS_PLUGINS == 'start' and stage == 'task':
+                continue
+        elif has_stage and plugin.get_option('stage') not in ('all', stage):
+            continue
+
+        data = combine_vars(data, get_plugin_vars(loader, plugin, path, entities))
+
+    return data
+
+
+def get_vars_from_inventory_sources(loader, sources, entities, stage):
+
+    data = {}
+    for path in sources:
+
+        if path is None:
+            continue
+        if ',' in path and not os.path.exists(path):  # skip host lists
+            continue
+        elif not os.path.isdir(to_bytes(path)):
+            # always pass the directory of the inventory source file
+            path = os.path.dirname(path)
+
+        data = combine_vars(data, get_vars_from_path(loader, path, entities, stage))
+
+    return data
diff --git a/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/plugins/vars/custom_vars.py b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/plugins/vars/custom_vars.py
new file mode 100644
index 00000000000..c603d72e33b
--- /dev/null
+++ b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/plugins/vars/custom_vars.py
@@ -0,0 +1,44 @@
+# Copyright 2019 RedHat, inc
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+#############################################
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    vars: custom_vars
+    version_added: "2.10"
+    short_description: load host and group vars
+    description: test loading host and group vars from a collection
+    options:
+      stage:
+        choices: ['all', 'inventory', 'task']
+        type: str
+        ini:
+          - key: stage
+            section: custom_vars
+        env:
+          - name: ANSIBLE_VARS_PLUGIN_STAGE
+'''
+
+from ansible.plugins.vars import BaseVarsPlugin
+
+
+class VarsModule(BaseVarsPlugin):
+
+    def get_vars(self, loader, path, entities, cache=True):
+        super(VarsModule, self).get_vars(loader, path, entities)
+        return {'collection': 'collection_root_user'}
diff --git a/test/integration/targets/collections/collections/ansible_collections/testns/content_adj/plugins/vars/custom_adj_vars.py b/test/integration/targets/collections/collections/ansible_collections/testns/content_adj/plugins/vars/custom_adj_vars.py
new file mode 100644
index 00000000000..0cd9a1d5fd3
--- /dev/null
+++ b/test/integration/targets/collections/collections/ansible_collections/testns/content_adj/plugins/vars/custom_adj_vars.py
@@ -0,0 +1,45 @@
+# Copyright 2019 RedHat, inc
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+#############################################
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    vars: custom_adj_vars
+    version_added: "2.10"
+    short_description: load host and group vars
+    description: test loading host and group vars from a collection
+    options:
+      stage:
+        default: all
+        choices: ['all', 'inventory', 'task']
+        type: str
+        ini:
+          - key: stage
+            section: custom_adj_vars
+        env:
+          - name: ANSIBLE_VARS_PLUGIN_STAGE
+'''
+
+from ansible.plugins.vars import BaseVarsPlugin
+
+
+class VarsModule(BaseVarsPlugin):
+
+    def get_vars(self, loader, path, entities, cache=True):
+        super(VarsModule, self).get_vars(loader, path, entities)
+        return {'collection': 'adjacent', 'adj_var': 'value'}
diff --git a/test/integration/targets/collections/custom_vars_plugins/v1_vars_plugin.py b/test/integration/targets/collections/custom_vars_plugins/v1_vars_plugin.py
new file mode 100644
index 00000000000..b5792d88d79
--- /dev/null
+++ b/test/integration/targets/collections/custom_vars_plugins/v1_vars_plugin.py
@@ -0,0 +1,37 @@
+# Copyright 2019 RedHat, inc
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+#############################################
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    vars: v1_vars_plugin
+    version_added: "2.10"
+    short_description: load host and group vars
+    description:
+      - 3rd party vars plugin to test loading host and group vars without requiring whitelisting and without a plugin-specific stage option
+    options:
+'''
+
+from ansible.plugins.vars import BaseVarsPlugin
+
+
+class VarsModule(BaseVarsPlugin):
+
+    def get_vars(self, loader, path, entities, cache=True):
+        super(VarsModule, self).get_vars(loader, path, entities)
+        return {'collection': False, 'name': 'v1_vars_plugin', 'v1_vars_plugin': True}
diff --git a/test/integration/targets/collections/custom_vars_plugins/v2_vars_plugin.py b/test/integration/targets/collections/custom_vars_plugins/v2_vars_plugin.py
new file mode 100644
index 00000000000..fc140162167
--- /dev/null
+++ b/test/integration/targets/collections/custom_vars_plugins/v2_vars_plugin.py
@@ -0,0 +1,45 @@
+# Copyright 2019 RedHat, inc
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+#############################################
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    vars: v2_vars_plugin
+    version_added: "2.10"
+    short_description: load host and group vars
+    description:
+      - 3rd party vars plugin to test loading host and group vars without requiring whitelisting and with a plugin-specific stage option
+    options:
+      stage:
+        choices: ['all', 'inventory', 'task']
+        type: str
+        ini:
+          - key: stage
+            section: other_vars_plugin
+        env:
+          - name: ANSIBLE_VARS_PLUGIN_STAGE
+'''
+
+from ansible.plugins.vars import BaseVarsPlugin
+
+
+class VarsModule(BaseVarsPlugin):
+
+    def get_vars(self, loader, path, entities, cache=True):
+        super(VarsModule, self).get_vars(loader, path, entities)
+        return {'collection': False, 'name': 'v2_vars_plugin', 'v2_vars_plugin': True}
diff --git a/test/integration/targets/collections/custom_vars_plugins/vars_req_whitelist.py b/test/integration/targets/collections/custom_vars_plugins/vars_req_whitelist.py
new file mode 100644
index 00000000000..0ab952735aa
--- /dev/null
+++ b/test/integration/targets/collections/custom_vars_plugins/vars_req_whitelist.py
@@ -0,0 +1,46 @@
+# Copyright 2019 RedHat, inc
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
+#############################################
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    vars: vars_req_whitelist
+    version_added: "2.10"
+    short_description: load host and group vars
+    description: test loading host and group vars from a collection
+    options:
+      stage:
+        choices: ['all', 'inventory', 'task']
+        type: str
+        ini:
+          - key: stage
+            section: vars_req_whitelist
+        env:
+          - name: ANSIBLE_VARS_PLUGIN_STAGE
+'''
+
+from ansible.plugins.vars import BaseVarsPlugin
+
+
+class VarsModule(BaseVarsPlugin):
+
+    REQUIRES_WHITELIST = True
+
+    def get_vars(self, loader, path, entities, cache=True):
+        super(VarsModule, self).get_vars(loader, path, entities)
+        return {'whitelisted': True, 'collection': False}
diff --git a/test/integration/targets/collections/runme.sh b/test/integration/targets/collections/runme.sh
index d2932945042..f485836eabb 100755
--- a/test/integration/targets/collections/runme.sh
+++ b/test/integration/targets/collections/runme.sh
@@ -42,3 +42,5 @@ ansible-playbook -i "${INVENTORY_PATH}"  -i ./a.statichost.yml -v "${TEST_PLAYBO
 # test adjacent with --playbook-dir
 export ANSIBLE_COLLECTIONS_PATHS=''
 ANSIBLE_INVENTORY_ANY_UNPARSED_IS_FAILED=1 ansible-inventory -i a.statichost.yml --list --export --playbook-dir=. -v "$@"
+
+./vars_plugin_tests.sh
diff --git a/test/integration/targets/collections/vars_plugin_tests.sh b/test/integration/targets/collections/vars_plugin_tests.sh
new file mode 100755
index 00000000000..040ea1f7978
--- /dev/null
+++ b/test/integration/targets/collections/vars_plugin_tests.sh
@@ -0,0 +1,99 @@
+#!/usr/bin/env bash
+
+set -eux
+
+# Collections vars plugins must be whitelisted with FQCN because PluginLoader.all() does not search collections
+
+# Let vars plugins run for inventory by using the global setting
+export ANSIBLE_RUN_VARS_PLUGINS=start
+
+# Test vars plugin in a playbook-adjacent collection
+export ANSIBLE_VARS_ENABLED=testns.content_adj.custom_adj_vars
+
+ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep '"collection": "adjacent"' out.txt
+grep '"adj_var": "value"' out.txt
+
+# Test vars plugin in a collection path
+export ANSIBLE_VARS_ENABLED=testns.testcoll.custom_vars
+export ANSIBLE_COLLECTIONS_PATHS=$PWD/collection_root_user:$PWD/collection_root_sys
+
+ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep '"collection": "collection_root_user"' out.txt
+grep -v '"adj_var": "value"' out.txt
+
+# Test enabled vars plugins order reflects the order in which variables are merged
+export ANSIBLE_VARS_ENABLED=testns.content_adj.custom_adj_vars,testns.testcoll.custom_vars
+
+ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep '"collection": "collection_root_user"' out.txt
+grep '"adj_var": "value"' out.txt
+grep -v '"collection": "adjacent"' out.txt
+
+# Test that 3rd party plugins in plugin_path do not need to require whitelisting by default
+# Plugins shipped with Ansible and in the custom plugin dir should be used first
+export ANSIBLE_VARS_PLUGINS=./custom_vars_plugins
+
+ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep '"name": "v2_vars_plugin"' out.txt
+grep '"collection": "collection_root_user"' out.txt
+grep '"adj_var": "value"' out.txt
+grep -v '"whitelisted": true' out.txt
+
+# Test plugins in plugin paths that opt-in to require whitelisting
+unset ANSIBLE_VARS_ENABLED
+unset ANSIBLE_COLLECTIONS_PATHS
+
+ANSIBLE_VARS_ENABLED=vars_req_whitelist ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep '"whitelisted": true' out.txt
+
+# Test vars plugins that support the stage setting don't run for inventory when stage is set to 'task'
+# and that the vars plugins that don't support the stage setting don't run for inventory when the global setting is 'demand'
+ANSIBLE_VARS_PLUGIN_STAGE=task ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep -v '"v1_vars_plugin": true' out.txt
+grep -v '"v2_vars_plugin": true' out.txt
+grep -v '"vars_req_whitelist": true' out.txt
+grep -v '"collection": "adjacent"' out.txt
+grep -v '"collection": "collection_root_user"' out.txt
+grep -v '"adj_var": "value"' out.txt
+
+# Test vars plugins that support the stage setting run for inventory when stage is set to 'inventory'
+ANSIBLE_VARS_PLUGIN_STAGE=inventory ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep -v '"v1_vars_plugin": true' out.txt
+grep -v '"vars_req_whitelist": true' out.txt
+grep '"v2_vars_plugin": true' out.txt
+grep '"name": "v2_vars_plugin"' out.txt
+
+# Test that the global setting allows v1 and v2 plugins to run after importing inventory
+ANSIBLE_RUN_VARS_PLUGINS=start ansible-inventory -i a.statichost.yml --list --playbook-dir=./ | tee out.txt
+
+grep -v '"vars_req_whitelist": true' out.txt
+grep '"v1_vars_plugin": true' out.txt
+grep '"v2_vars_plugin": true' out.txt
+grep '"name": "v2_vars_plugin"' out.txt
+
+# Test that vars plugins in collections and in the vars plugin path are available for tasks
+cat << EOF > "test_task_vars.yml"
+---
+- hosts: localhost
+  connection: local
+  gather_facts: no
+  tasks:
+  - debug: msg="{{ name }}"
+  - debug: msg="{{ collection }}"
+  - debug: msg="{{ adj_var }}"
+EOF
+
+export ANSIBLE_VARS_ENABLED=testns.content_adj.custom_adj_vars
+
+ANSIBLE_VARS_PLUGIN_STAGE=task ANSIBLE_VARS_PLUGINS=./custom_vars_plugins ansible-playbook test_task_vars.yml | grep "ok=3"
+ANSIBLE_RUN_VARS_PLUGINS=start ANSIBLE_VARS_PLUGIN_STAGE=inventory ANSIBLE_VARS_PLUGINS=./custom_vars_plugins ansible-playbook test_task_vars.yml | grep "ok=3"
+ANSIBLE_RUN_VARS_PLUGINS=demand ANSIBLE_VARS_PLUGIN_STAGE=inventory ANSIBLE_VARS_PLUGINS=./custom_vars_plugins ansible-playbook test_task_vars.yml | grep "ok=3"
+ANSIBLE_VARS_PLUGINS=./custom_vars_plugins ansible-playbook test_task_vars.yml | grep "ok=3"