diff --git a/lib/ansible/config/base.yml b/lib/ansible/config/base.yml
index c6f8ba9715..327f506a62 100644
--- a/lib/ansible/config/base.yml
+++ b/lib/ansible/config/base.yml
@@ -1300,19 +1300,16 @@ NETWORK_GROUP_MODULES:
   - {key: network_group_modules, section: defaults}
   type: list
   yaml: {key: defaults.network_group_modules}
-#ONLY_NAMESPACE_FACTS:
-# Deffered to 2.5
-# FIXME: reenable when we can remove ansible_ prefix from namespaced facts
-#  default: False
-#  description:
-#    - Facts normally get injected as top level variables, this setting prevents that.
-#    - Facts are still available in the `ansible_facts` variable w/o the `ansible_` prefix.
-#  env: [{name: ANSIBLE_RESTRICT_FACTS}]
-#  ini:
-#  - {key: restrict_facts_namespace, section: defaults}
-#  type: boolean
-#  yaml: {key: defaults.restrict_facts_namespace}
-#  version_added: "2.4"
+INJECT_FACTS_AS_VARS:
+  default: True
+  description:
+    - Facts are available inside the `ansible_facts` variable, this setting also pushes them as their own vars in the main namespace.
+    - Unlike inside the `ansible_facts` dictionary, these will have an `ansible_` prefix.
+  env: [{name: ANSIBLE_INJECT_FACT_VARS}]
+  ini:
+  - {key: inject_facts_as_vars, section: defaults}
+  type: boolean
+  version_added: "2.5"
 PARAMIKO_HOST_KEY_AUTO_ADD:
   # TODO: move to plugin
   default: False
diff --git a/lib/ansible/constants.py b/lib/ansible/constants.py
index 28b1cae2ff..c6255d5c67 100644
--- a/lib/ansible/constants.py
+++ b/lib/ansible/constants.py
@@ -99,6 +99,61 @@ TREE_DIR = None
 VAULT_VERSION_MIN = 1.0
 VAULT_VERSION_MAX = 1.0
 
+# FIXME: remove once play_context mangling is removed
+MAGIC_VARIABLE_MAPPING = dict(
+
+    # base
+    connection=('ansible_connection', ),
+    module_compression=('ansible_module_compression', ),
+    shell=('ansible_shell_type', ),
+    executable=('ansible_shell_executable', ),
+    remote_tmp_dir=('ansible_remote_tmp', ),
+
+    # connection common
+    remote_addr=('ansible_ssh_host', 'ansible_host'),
+    remote_user=('ansible_ssh_user', 'ansible_user'),
+    password=('ansible_ssh_pass', 'ansible_password'),
+    port=('ansible_ssh_port', 'ansible_port'),
+    pipelining=('ansible_ssh_pipelining', 'ansible_pipelining'),
+    timeout=('ansible_ssh_timeout', 'ansible_timeout'),
+    private_key_file=('ansible_ssh_private_key_file', 'ansible_private_key_file'),
+
+    # networking modules
+    network_os=('ansible_network_os', ),
+    connection_user=('ansible_connection_user',),
+
+    # ssh TODO: remove
+    ssh_executable=('ansible_ssh_executable', ),
+    ssh_common_args=('ansible_ssh_common_args', ),
+    sftp_extra_args=('ansible_sftp_extra_args', ),
+    scp_extra_args=('ansible_scp_extra_args', ),
+    ssh_extra_args=('ansible_ssh_extra_args', ),
+    ssh_transfer_method=('ansible_ssh_transfer_method', ),
+
+    # docker TODO: remove
+    docker_extra_args=('ansible_docker_extra_args', ),
+
+    # become
+    become=('ansible_become', ),
+    become_method=('ansible_become_method', ),
+    become_user=('ansible_become_user', ),
+    become_pass=('ansible_become_password', 'ansible_become_pass'),
+    become_exe=('ansible_become_exe', ),
+    become_flags=('ansible_become_flags', ),
+
+    # deprecated
+    sudo=('ansible_sudo', ),
+    sudo_user=('ansible_sudo_user', ),
+    sudo_pass=('ansible_sudo_password', 'ansible_sudo_pass'),
+    sudo_exe=('ansible_sudo_exe', ),
+    sudo_flags=('ansible_sudo_flags', ),
+    su=('ansible_su', ),
+    su_user=('ansible_su_user', ),
+    su_pass=('ansible_su_password', 'ansible_su_pass'),
+    su_exe=('ansible_su_exe', ),
+    su_flags=('ansible_su_flags', ),
+)
+
 # POPULATE SETTINGS FROM CONFIG ###
 config = ConfigManager()
 
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 845367a425..78a1902bf6 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -33,6 +33,7 @@ from ansible.plugins.connection import ConnectionBase
 from ansible.template import Templar
 from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.unsafe_proxy import UnsafeProxy, wrap_var
+from ansible.vars.clean import namespace_facts, clean_facts
 
 try:
     from __main__ import display
@@ -574,7 +575,9 @@ class TaskExecutor:
                 return failed_when_result
 
             if 'ansible_facts' in result:
-                vars_copy.update(result['ansible_facts'])
+                vars_copy.update(namespace_facts(result['ansible_facts']))
+                if C.INJECT_FACTS_AS_VARS:
+                    vars_copy.update(clean_facts(result['ansible_facts']))
 
             # set the failed property if it was missing.
             if 'failed' not in result:
@@ -620,7 +623,9 @@ class TaskExecutor:
             variables[self._task.register] = wrap_var(result)
 
         if 'ansible_facts' in result:
-            variables.update(result['ansible_facts'])
+            variables.update(namespace_facts(result['ansible_facts']))
+            if C.INJECT_FACTS_AS_VARS:
+                variables.update(clean_facts(result['ansible_facts']))
 
         # save the notification target in the result, if it was specified, as
         # this task may be running in a loop in which case the notification
diff --git a/lib/ansible/executor/task_result.py b/lib/ansible/executor/task_result.py
index efac78f945..ef35a7f118 100644
--- a/lib/ansible/executor/task_result.py
+++ b/lib/ansible/executor/task_result.py
@@ -8,7 +8,7 @@ __metaclass__ = type
 from copy import deepcopy
 
 from ansible.parsing.dataloader import DataLoader
-from ansible.vars.manager import strip_internal_keys
+from ansible.vars.clean import strip_internal_keys
 
 _IGNORE = ('failed', 'skipped')
 
diff --git a/lib/ansible/module_utils/facts/compat.py b/lib/ansible/module_utils/facts/compat.py
index 3f28c4102c..289d36965f 100644
--- a/lib/ansible/module_utils/facts/compat.py
+++ b/lib/ansible/module_utils/facts/compat.py
@@ -72,8 +72,7 @@ def ansible_facts(module, gather_subset=None):
     all_collector_classes = default_collectors.collectors
 
     # don't add a prefix
-    namespace = PrefixFactNamespace(namespace_name='ansible',
-                                    prefix='')
+    namespace = PrefixFactNamespace(namespace_name='ansible', prefix='')
 
     fact_collector = \
         ansible_collector.get_ansible_collector(all_collector_classes=all_collector_classes,
diff --git a/lib/ansible/playbook/play_context.py b/lib/ansible/playbook/play_context.py
index 120794536e..4f0f86afc8 100644
--- a/lib/ansible/playbook/play_context.py
+++ b/lib/ansible/playbook/play_context.py
@@ -54,59 +54,6 @@ __all__ = ['PlayContext']
 # object. The dictionary values are tuples, to account for aliases
 # in variable names.
 
-MAGIC_VARIABLE_MAPPING = dict(
-
-    # base
-    connection=('ansible_connection', ),
-    module_compression=('ansible_module_compression', ),
-    shell=('ansible_shell_type', ),
-    executable=('ansible_shell_executable', ),
-    remote_tmp_dir=('ansible_remote_tmp', ),
-
-    # connection common
-    remote_addr=('ansible_ssh_host', 'ansible_host'),
-    remote_user=('ansible_ssh_user', 'ansible_user'),
-    password=('ansible_ssh_pass', 'ansible_password'),
-    port=('ansible_ssh_port', 'ansible_port'),
-    pipelining=('ansible_ssh_pipelining', 'ansible_pipelining'),
-    timeout=('ansible_ssh_timeout', 'ansible_timeout'),
-    private_key_file=('ansible_ssh_private_key_file', 'ansible_private_key_file'),
-
-    # networking modules
-    network_os=('ansible_network_os', ),
-    connection_user=('ansible_connection_user',),
-
-    # ssh TODO: remove
-    ssh_executable=('ansible_ssh_executable', ),
-    ssh_common_args=('ansible_ssh_common_args', ),
-    sftp_extra_args=('ansible_sftp_extra_args', ),
-    scp_extra_args=('ansible_scp_extra_args', ),
-    ssh_extra_args=('ansible_ssh_extra_args', ),
-    ssh_transfer_method=('ansible_ssh_transfer_method', ),
-
-    # docker TODO: remove
-    docker_extra_args=('ansible_docker_extra_args', ),
-
-    # become
-    become=('ansible_become', ),
-    become_method=('ansible_become_method', ),
-    become_user=('ansible_become_user', ),
-    become_pass=('ansible_become_password', 'ansible_become_pass'),
-    become_exe=('ansible_become_exe', ),
-    become_flags=('ansible_become_flags', ),
-
-    # deprecated
-    sudo=('ansible_sudo', ),
-    sudo_user=('ansible_sudo_user', ),
-    sudo_pass=('ansible_sudo_password', 'ansible_sudo_pass'),
-    sudo_exe=('ansible_sudo_exe', ),
-    sudo_flags=('ansible_sudo_flags', ),
-    su=('ansible_su', ),
-    su_user=('ansible_su_user', ),
-    su_pass=('ansible_su_password', 'ansible_su_pass'),
-    su_exe=('ansible_su_exe', ),
-    su_flags=('ansible_su_flags', ),
-)
 
 # TODO: needs to be configurable
 b_SU_PROMPT_LOCALIZATIONS = [
@@ -382,7 +329,7 @@ class PlayContext(Base):
             delegated_vars = variables.get('ansible_delegated_vars', dict()).get(delegated_host_name, dict())
 
             delegated_transport = C.DEFAULT_TRANSPORT
-            for transport_var in MAGIC_VARIABLE_MAPPING.get('connection'):
+            for transport_var in C.MAGIC_VARIABLE_MAPPING.get('connection'):
                 if transport_var in delegated_vars:
                     delegated_transport = delegated_vars[transport_var]
                     break
@@ -391,7 +338,7 @@ class PlayContext(Base):
             # address, otherwise we default to connecting to it by name. This
             # may happen when users put an IP entry into their inventory, or if
             # they rely on DNS for a non-inventory hostname
-            for address_var in ('ansible_%s_host' % transport_var,) + MAGIC_VARIABLE_MAPPING.get('remote_addr'):
+            for address_var in ('ansible_%s_host' % transport_var,) + C.MAGIC_VARIABLE_MAPPING.get('remote_addr'):
                 if address_var in delegated_vars:
                     break
             else:
@@ -400,7 +347,7 @@ class PlayContext(Base):
 
             # reset the port back to the default if none was specified, to prevent
             # the delegated host from inheriting the original host's setting
-            for port_var in ('ansible_%s_port' % transport_var,) + MAGIC_VARIABLE_MAPPING.get('port'):
+            for port_var in ('ansible_%s_port' % transport_var,) + C.MAGIC_VARIABLE_MAPPING.get('port'):
                 if port_var in delegated_vars:
                     break
             else:
@@ -410,7 +357,7 @@ class PlayContext(Base):
                     delegated_vars['ansible_port'] = C.DEFAULT_REMOTE_PORT
 
             # and likewise for the remote user
-            for user_var in ('ansible_%s_user' % transport_var,) + MAGIC_VARIABLE_MAPPING.get('remote_user'):
+            for user_var in ('ansible_%s_user' % transport_var,) + C.MAGIC_VARIABLE_MAPPING.get('remote_user'):
                 if user_var in delegated_vars and delegated_vars[user_var]:
                     break
             else:
@@ -419,12 +366,12 @@ class PlayContext(Base):
             delegated_vars = dict()
 
             # setup shell
-            for exe_var in MAGIC_VARIABLE_MAPPING.get('executable'):
+            for exe_var in C.MAGIC_VARIABLE_MAPPING.get('executable'):
                 if exe_var in variables:
                     setattr(new_info, 'executable', variables.get(exe_var))
 
         attrs_considered = []
-        for (attr, variable_names) in iteritems(MAGIC_VARIABLE_MAPPING):
+        for (attr, variable_names) in iteritems(C.MAGIC_VARIABLE_MAPPING):
             for variable_name in variable_names:
                 if attr in attrs_considered:
                     continue
@@ -447,17 +394,17 @@ class PlayContext(Base):
 
         # become legacy updates -- from inventory file (inventory overrides
         # commandline)
-        for become_pass_name in MAGIC_VARIABLE_MAPPING.get('become_pass'):
+        for become_pass_name in C.MAGIC_VARIABLE_MAPPING.get('become_pass'):
             if become_pass_name in variables:
                 break
         else:  # This is a for-else
             if new_info.become_method == 'sudo':
-                for sudo_pass_name in MAGIC_VARIABLE_MAPPING.get('sudo_pass'):
+                for sudo_pass_name in C.MAGIC_VARIABLE_MAPPING.get('sudo_pass'):
                     if sudo_pass_name in variables:
                         setattr(new_info, 'become_pass', variables[sudo_pass_name])
                         break
             elif new_info.become_method == 'su':
-                for su_pass_name in MAGIC_VARIABLE_MAPPING.get('su_pass'):
+                for su_pass_name in C.MAGIC_VARIABLE_MAPPING.get('su_pass'):
                     if su_pass_name in variables:
                         setattr(new_info, 'become_pass', variables[su_pass_name])
                         break
@@ -471,7 +418,7 @@ class PlayContext(Base):
             # in the event that we were using local before make sure to reset the
             # connection type to the default transport for the delegated-to host,
             # if not otherwise specified
-            for connection_type in MAGIC_VARIABLE_MAPPING.get('connection'):
+            for connection_type in C.MAGIC_VARIABLE_MAPPING.get('connection'):
                 if connection_type in delegated_vars:
                     break
             else:
@@ -636,7 +583,7 @@ class PlayContext(Base):
         In case users need to access from the play, this is a legacy from runner.
         '''
 
-        for prop, var_list in MAGIC_VARIABLE_MAPPING.items():
+        for prop, var_list in C.MAGIC_VARIABLE_MAPPING.items():
             try:
                 if 'become' in prop:
                     continue
diff --git a/lib/ansible/plugins/action/__init__.py b/lib/ansible/plugins/action/__init__.py
index 308f0797b0..6324d009ac 100644
--- a/lib/ansible/plugins/action/__init__.py
+++ b/lib/ansible/plugins/action/__init__.py
@@ -37,10 +37,9 @@ from ansible.module_utils.six import binary_type, string_types, text_type, iteri
 from ansible.module_utils.six.moves import shlex_quote
 from ansible.module_utils._text import to_bytes, to_native, to_text
 from ansible.parsing.utils.jsonify import jsonify
-from ansible.playbook.play_context import MAGIC_VARIABLE_MAPPING
 from ansible.release import __version__
 from ansible.utils.unsafe_proxy import wrap_var
-from ansible.vars.manager import remove_internal_keys
+from ansible.vars.clean import remove_internal_keys
 
 
 try:
@@ -765,49 +764,6 @@ class ActionBase(with_metaclass(ABCMeta, object)):
         display.debug("done with _execute_module (%s, %s)" % (module_name, module_args))
         return data
 
-    def _clean_returned_data(self, data):
-        remove_keys = set()
-        fact_keys = set(data.keys())
-        # first we add all of our magic variable names to the set of
-        # keys we want to remove from facts
-        for magic_var in MAGIC_VARIABLE_MAPPING:
-            remove_keys.update(fact_keys.intersection(MAGIC_VARIABLE_MAPPING[magic_var]))
-        # next we remove any connection plugin specific vars
-        for conn_path in self._shared_loader_obj.connection_loader.all(path_only=True):
-            try:
-                conn_name = os.path.splitext(os.path.basename(conn_path))[0]
-                re_key = re.compile('^ansible_%s_' % conn_name)
-                for fact_key in fact_keys:
-                    # exception for lvm tech, whic normally returns asnible_x_bridge facts that get filterd out (docker,lxc, etc)
-                    if re_key.match(fact_key) and not fact_key.endswith(('_bridge', '_gwbridge')):
-                        remove_keys.add(fact_key)
-            except AttributeError:
-                pass
-
-        # remove some KNOWN keys
-        for hard in C.RESTRICTED_RESULT_KEYS + C.INTERNAL_RESULT_KEYS:
-            if hard in fact_keys:
-                remove_keys.add(hard)
-
-        # finally, we search for interpreter keys to remove
-        re_interp = re.compile('^ansible_.*_interpreter$')
-        for fact_key in fact_keys:
-            if re_interp.match(fact_key):
-                remove_keys.add(fact_key)
-        # then we remove them (except for ssh host keys)
-        for r_key in remove_keys:
-            if not r_key.startswith('ansible_ssh_host_key_'):
-                try:
-                    r_val = to_text(data[r_key])
-                    if len(r_val) > 24:
-                        r_val = '%s ... %s' % (r_val[:13], r_val[-6:])
-                except:
-                    r_val = ' <failed to convert value to a string> '
-                display.warning("Removed restricted key from module data: %s = %s" % (r_key, r_val))
-                del data[r_key]
-
-        remove_internal_keys(data)
-
     def _parse_returned_data(self, res):
         try:
             filtered_output, warnings = _filter_non_json_lines(res.get('stdout', u''))
@@ -817,7 +773,6 @@ class ActionBase(with_metaclass(ABCMeta, object)):
             data = json.loads(filtered_output)
 
             if 'ansible_facts' in data and isinstance(data['ansible_facts'], dict):
-                self._clean_returned_data(data['ansible_facts'])
                 data['ansible_facts'] = wrap_var(data['ansible_facts'])
             data['_ansible_parsed'] = True
         except ValueError:
diff --git a/lib/ansible/plugins/action/package.py b/lib/ansible/plugins/action/package.py
index 97660028f9..a72880bf5b 100644
--- a/lib/ansible/plugins/action/package.py
+++ b/lib/ansible/plugins/action/package.py
@@ -43,9 +43,9 @@ class ActionModule(ActionBase):
         if module == 'auto':
             try:
                 if self._task.delegate_to:  # if we delegate, we should use delegated host's facts
-                    module = self._templar.template("{{hostvars['%s']['ansible_pkg_mgr']}}" % self._task.delegate_to)
+                    module = self._templar.template("{{hostvars['%s']['ansible_facts']['pkg_mgr']}}" % self._task.delegate_to)
                 else:
-                    module = self._templar.template('{{ansible_pkg_mgr}}')
+                    module = self._templar.template('{{ansible_facts.pkg_mgr}}')
             except:
                 pass  # could not get it from template!
 
diff --git a/lib/ansible/plugins/action/service.py b/lib/ansible/plugins/action/service.py
index dd0c796539..a7c67fdee6 100644
--- a/lib/ansible/plugins/action/service.py
+++ b/lib/ansible/plugins/action/service.py
@@ -42,9 +42,9 @@ class ActionModule(ActionBase):
         if module == 'auto':
             try:
                 if self._task.delegate_to:  # if we delegate, we should use delegated host's facts
-                    module = self._templar.template("{{hostvars['%s']['ansible_service_mgr']}}" % self._task.delegate_to)
+                    module = self._templar.template("{{hostvars['%s']['ansible_facts']['service_mgr']}}" % self._task.delegate_to)
                 else:
-                    module = self._templar.template('{{ansible_service_mgr}}')
+                    module = self._templar.template('{{ansible_facts.service_mgr}}')
             except:
                 pass  # could not get it from template!
 
diff --git a/lib/ansible/plugins/action/synchronize.py b/lib/ansible/plugins/action/synchronize.py
index 94ba0e1e6d..6c6613cd7a 100644
--- a/lib/ansible/plugins/action/synchronize.py
+++ b/lib/ansible/plugins/action/synchronize.py
@@ -24,7 +24,6 @@ from ansible import constants as C
 from ansible.module_utils.six import string_types
 from ansible.module_utils._text import to_text
 from ansible.module_utils.parsing.convert_bool import boolean
-from ansible.playbook.play_context import MAGIC_VARIABLE_MAPPING
 from ansible.plugins.action import ActionBase
 from ansible.plugins.loader import connection_loader
 
@@ -223,7 +222,7 @@ class ActionModule(ActionBase):
         localhost_ports = set()
         for host in C.LOCALHOST:
             localhost_vars = task_vars['hostvars'].get(host, {})
-            for port_var in MAGIC_VARIABLE_MAPPING['port']:
+            for port_var in C.MAGIC_VARIABLE_MAPPING['port']:
                 port = localhost_vars.get(port_var, None)
                 if port:
                     break
@@ -271,7 +270,7 @@ class ActionModule(ActionBase):
             localhost_shell = None
             for host in C.LOCALHOST:
                 localhost_vars = task_vars['hostvars'].get(host, {})
-                for shell_var in MAGIC_VARIABLE_MAPPING['shell']:
+                for shell_var in C.MAGIC_VARIABLE_MAPPING['shell']:
                     localhost_shell = localhost_vars.get(shell_var, None)
                     if localhost_shell:
                         break
@@ -285,7 +284,7 @@ class ActionModule(ActionBase):
             localhost_executable = None
             for host in C.LOCALHOST:
                 localhost_vars = task_vars['hostvars'].get(host, {})
-                for executable_var in MAGIC_VARIABLE_MAPPING['executable']:
+                for executable_var in C.MAGIC_VARIABLE_MAPPING['executable']:
                     localhost_executable = localhost_vars.get(executable_var, None)
                     if localhost_executable:
                         break
diff --git a/lib/ansible/plugins/callback/__init__.py b/lib/ansible/plugins/callback/__init__.py
index 0376253a23..d9f40a91e3 100644
--- a/lib/ansible/plugins/callback/__init__.py
+++ b/lib/ansible/plugins/callback/__init__.py
@@ -29,7 +29,7 @@ from ansible import constants as C
 from ansible.plugins import AnsiblePlugin
 from ansible.module_utils._text import to_text
 from ansible.utils.color import stringc
-from ansible.vars.manager import strip_internal_keys
+from ansible.vars.clean import strip_internal_keys
 
 try:
     from __main__ import display as global_display
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index ae7e1e8d28..0980b5cadb 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -43,7 +43,7 @@ from ansible.playbook.role_include import IncludeRole
 from ansible.plugins.loader import action_loader, connection_loader, filter_loader, lookup_loader, module_loader, test_loader
 from ansible.template import Templar
 from ansible.utils.vars import combine_vars
-from ansible.vars.manager import strip_internal_keys
+from ansible.vars.clean import strip_internal_keys
 
 
 try:
diff --git a/lib/ansible/vars/clean.py b/lib/ansible/vars/clean.py
new file mode 100644
index 0000000000..5a34997f72
--- /dev/null
+++ b/lib/ansible/vars/clean.py
@@ -0,0 +1,128 @@
+# Copyright (c) 2017 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
+import re
+
+from copy import deepcopy
+
+from ansible import constants as C
+from ansible.module_utils._text import to_text
+from ansible.module_utils.six import string_types
+from ansible.plugins.loader import connection_loader
+
+try:
+    from __main__ import display
+except ImportError:
+    from ansible.utils.display import Display
+    display = Display()
+
+
+def strip_internal_keys(dirty, exceptions=None):
+    '''
+    All keys stating with _ansible_ are internal, so create a copy of the 'dirty' dict
+    and remove them from the clean one before returning it
+    '''
+
+    if exceptions is None:
+        exceptions = ()
+    clean = dirty.copy()
+    for k in dirty.keys():
+        if isinstance(k, string_types) and k.startswith('_ansible_'):
+            if k not in exceptions:
+                del clean[k]
+        elif isinstance(dirty[k], dict):
+            clean[k] = strip_internal_keys(dirty[k])
+    return clean
+
+
+def remove_internal_keys(data):
+    '''
+    More nuanced version of strip_internal_keys
+    '''
+    for key in list(data.keys()):
+        if (key.startswith('_ansible_') and key != '_ansible_parsed') or key in C.INTERNAL_RESULT_KEYS:
+            display.warning("Removed unexpected internal key in module return: %s = %s" % (key, data[key]))
+            del data[key]
+
+    # remove bad/empty internal keys
+    for key in ['warnings', 'deprecations']:
+        if key in data and not data[key]:
+            del data[key]
+
+
+def clean_facts(facts):
+    ''' remove facts that can override internal keys or othewise deemed unsafe '''
+    data = deepcopy(facts)
+
+    remove_keys = set()
+    fact_keys = set(data.keys())
+    # first we add all of our magic variable names to the set of
+    # keys we want to remove from facts
+    for magic_var in C.MAGIC_VARIABLE_MAPPING:
+        remove_keys.update(fact_keys.intersection(C.MAGIC_VARIABLE_MAPPING[magic_var]))
+    # next we remove any connection plugin specific vars
+    for conn_path in connection_loader.all(path_only=True):
+        try:
+            conn_name = os.path.splitext(os.path.basename(conn_path))[0]
+            re_key = re.compile('^ansible_%s_' % conn_name)
+            for fact_key in fact_keys:
+                # exception for lvm tech, whic normally returns asnible_x_bridge facts that get filterd out (docker,lxc, etc)
+                if re_key.match(fact_key) and not fact_key.endswith(('_bridge', '_gwbridge')):
+                    remove_keys.add(fact_key)
+        except AttributeError:
+            pass
+
+    # remove some KNOWN keys
+    for hard in C.RESTRICTED_RESULT_KEYS + C.INTERNAL_RESULT_KEYS:
+        if hard in fact_keys:
+            remove_keys.add(hard)
+
+    # finally, we search for interpreter keys to remove
+    re_interp = re.compile('^ansible_.*_interpreter$')
+    for fact_key in fact_keys:
+        if re_interp.match(fact_key):
+            remove_keys.add(fact_key)
+    # then we remove them (except for ssh host keys)
+    for r_key in remove_keys:
+        if not r_key.startswith('ansible_ssh_host_key_'):
+            try:
+                r_val = to_text(data[r_key])
+                if len(r_val) > 24:
+                    r_val = '%s ... %s' % (r_val[:13], r_val[-6:])
+            except Exception:
+                r_val = ' <failed to convert value to a string> '
+            display.warning("Removed restricted key from module data: %s = %s" % (r_key, r_val))
+            del data[r_key]
+
+    return strip_internal_keys(data)
+
+
+def inject_facts(facts):
+    ''' return clean facts inside with an ansible_ prefix '''
+    injected = {}
+    for k in facts:
+        if k.startswith('ansible_') or k == 'module_setup':
+            new = k
+        else:
+            new = 'ansilbe_%s' % k
+        injected[new] = deepcopy(facts[k])
+
+    return clean_facts(injected)
+
+
+def namespace_facts(facts):
+    ''' return all facts inside 'ansible_facts' w/o an ansible_ prefix '''
+    deprefixed = {}
+    for k in facts:
+        if k in ('ansible_local'):
+            # exceptions to 'deprefixing'
+            deprefixed[k] = deepcopy(facts[k])
+        else:
+            deprefixed[k.replace('ansible_', '', 1)] = deepcopy(facts[k])
+
+    return {'ansible_facts': deprefixed}
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index 8a36f496e2..8f2986a548 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -36,13 +36,14 @@ from ansible.errors import AnsibleError, AnsibleParserError, AnsibleUndefinedVar
 from ansible.inventory.host import Host
 from ansible.inventory.helpers import sort_groups, get_group_vars
 from ansible.module_utils._text import to_native
-from ansible.module_utils.six import iteritems, string_types, text_type
+from ansible.module_utils.six import iteritems, text_type
 from ansible.plugins.loader import lookup_loader, vars_loader
 from ansible.plugins.cache import FactCache
 from ansible.template import Templar
 from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.vars import combine_vars
 from ansible.utils.unsafe_proxy import wrap_var
+from ansible.vars.clean import namespace_facts, clean_facts
 
 try:
     from __main__ import display
@@ -72,39 +73,6 @@ def preprocess_vars(a):
     return data
 
 
-def strip_internal_keys(dirty, exceptions=None):
-    '''
-    All keys stating with _ansible_ are internal, so create a copy of the 'dirty' dict
-    and remove them from the clean one before returning it
-    '''
-
-    if exceptions is None:
-        exceptions = ()
-    clean = dirty.copy()
-    for k in dirty.keys():
-        if isinstance(k, string_types) and k.startswith('_ansible_'):
-            if k not in exceptions:
-                del clean[k]
-        elif isinstance(dirty[k], dict):
-            clean[k] = strip_internal_keys(dirty[k])
-    return clean
-
-
-def remove_internal_keys(data):
-    '''
-    More nuanced version of strip_internal_keys
-    '''
-    for key in list(data.keys()):
-        if (key.startswith('_ansible_') and key != '_ansible_parsed') or key in C.INTERNAL_RESULT_KEYS:
-            display.warning("Removed unexpected internal key in module return: %s = %s" % (key, data[key]))
-            del data[key]
-
-    # remove bad/empty internal keys
-    for key in ['warnings', 'deprecations']:
-        if key in data and not data[key]:
-            del data[key]
-
-
 class VariableManager:
 
     _ALLOWED = frozenset(['plugins_by_group', 'groups_plugins_play', 'groups_plugins_inventory', 'groups_inventory',
@@ -351,10 +319,15 @@ class VariableManager:
 
             # finally, the facts caches for this host, if it exists
             try:
-                host_facts = wrap_var(self._fact_cache.get(host.name, {}))
+                facts = self._fact_cache.get(host.name, {})
+                all_vars.update(namespace_facts(facts))
 
                 # push facts to main namespace
-                all_vars = combine_vars(all_vars, host_facts)
+                if C.INJECT_FACTS_AS_VARS:
+                    all_vars = combine_vars(all_vars, wrap_var(facts))
+                else:
+                    # always 'promote' ansible_local
+                    all_vars = combine_vars(all_vars, wrap_var({'ansible_local': facts.get('ansible_local', {})}))
             except KeyError:
                 pass
 
@@ -431,7 +404,9 @@ class VariableManager:
         # next, we merge in the vars cache (include vars) and nonpersistent
         # facts cache (set_fact/register), in that order
         if host:
+            # include_vars non-persistent cache
             all_vars = combine_vars(all_vars, self._vars_cache.get(host.get_name(), dict()))
+            # fact non-persistent cache
             all_vars = combine_vars(all_vars, self._nonpersistent_fact_cache.get(host.name, dict()))
 
         # next, we merge in role params and task include params
diff --git a/test/units/plugins/action/test_action.py b/test/units/plugins/action/test_action.py
index 788401c33d..6adaff32e2 100644
--- a/test/units/plugins/action/test_action.py
+++ b/test/units/plugins/action/test_action.py
@@ -33,6 +33,7 @@ from ansible.module_utils._text import to_bytes
 from ansible.playbook.play_context import PlayContext
 from ansible.plugins.action import ActionBase
 from ansible.template import Templar
+from ansible.vars.clean import clean_facts
 
 from units.mock.loader import DictDataLoader
 
@@ -550,7 +551,7 @@ class TestActionBaseCleanReturnedData(unittest.TestCase):
                 'ansible_ssh_some_var': 'whatever',
                 'ansible_ssh_host_key_somehost': 'some key here',
                 'some_other_var': 'foo bar'}
-        action_base._clean_returned_data(data)
+        data = clean_facts(data)
         self.assertNotIn('ansible_playbook_python', data)
         self.assertNotIn('ansible_python_interpreter', data)
         self.assertIn('ansible_ssh_host_key_somehost', data)