diff --git a/neutron_lib/constants.py b/neutron_lib/constants.py
index 3af197b61..d4633abb1 100644
--- a/neutron_lib/constants.py
+++ b/neutron_lib/constants.py
@@ -284,3 +284,13 @@ VHOST_USER_DEVICE_PREFIX = 'vhu'
 VETH_DEVICE_PREFIX = 'qvo'
 # prefix for SNAT interface in DVR
 SNAT_INT_DEV_PREFIX = 'sg-'
+
+
+##########################
+# Plugin related constants
+##########################
+# Plugin constants that are universally used across all neutron repos.
+# The alias for the core plugin.
+CORE = 'CORE'
+# The alias for the L3 plugin.
+L3 = 'L3_ROUTER_NAT'
diff --git a/neutron_lib/plugins/__init__.py b/neutron_lib/plugins/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/neutron_lib/plugins/directory.py b/neutron_lib/plugins/directory.py
new file mode 100644
index 000000000..276fff950
--- /dev/null
+++ b/neutron_lib/plugins/directory.py
@@ -0,0 +1,92 @@
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import weakref
+
+from oslo_concurrency import lockutils
+
+from neutron_lib import constants
+
+
+_synchronized = lockutils.synchronized_with_prefix("neutron-")
+
+
+class _PluginDirectory(object):
+    """A directory of activated plugins in a Neutron Deployment.
+
+    The directory is bootstrapped by a Neutron Manager running in
+    the context of a Neutron Server process.
+    """
+
+    def __init__(self):
+        self._plugins = {}
+
+    def add_plugin(self, alias, plugin):
+        """Add a plugin of type 'alias'."""
+        self._plugins[alias] = plugin
+
+    def get_plugin(self, alias):
+        """Get a plugin for a given alias or None if not present."""
+        return self.plugins.get(alias)
+
+    @property
+    def plugins(self):
+        """The mapping alias -> weak reference to the plugin."""
+        return dict((x, weakref.proxy(y))
+                    for x, y in self._plugins.items())
+
+    @property
+    def unique_plugins(self):
+        """A sequence of the unique plugins activated in the environments."""
+        return tuple(weakref.proxy(x) for x in set(self._plugins.values()))
+
+    @property
+    def is_loaded(self):
+        """True if the directory is non empty."""
+        return len(self._plugins) > 0
+
+
+# Create a singleton plugins directory for the Neutron server instance.
+# Accessing these methods before a Neutron Manager has had the chance
+# to load the environment may result in callers handling an empty directory.
+_PLUGIN_DIRECTORY = None
+
+
+@_synchronized("plugin-directory")
+def _create_plugin_directory():
+    global _PLUGIN_DIRECTORY
+    _PLUGIN_DIRECTORY = _PluginDirectory()
+    return _PLUGIN_DIRECTORY
+
+
+def _get_plugin_directory():
+    if _PLUGIN_DIRECTORY is None:
+        return _create_plugin_directory()
+    return _PLUGIN_DIRECTORY
+
+
+def add_plugin(alias, plugin):
+    _get_plugin_directory().add_plugin(alias, plugin)
+
+
+def get_plugin(alias=constants.CORE):
+    return _get_plugin_directory().get_plugin(alias)
+
+
+def get_plugins():
+    return _get_plugin_directory().plugins
+
+
+def get_unique_plugins():
+    return _get_plugin_directory().unique_plugins
diff --git a/neutron_lib/tests/_base.py b/neutron_lib/tests/_base.py
index 7facdf04f..56b549e53 100644
--- a/neutron_lib/tests/_base.py
+++ b/neutron_lib/tests/_base.py
@@ -30,6 +30,7 @@ import testtools
 from neutron_lib._i18n import _
 from neutron_lib import constants
 from neutron_lib import exceptions
+from neutron_lib.plugins import directory
 
 from neutron_lib.tests import _post_mortem_debug as post_mortem_debug
 from neutron_lib.tests import _tools as tools
@@ -117,6 +118,7 @@ class BaseTestCase(testtools.TestCase):
 
     def setUp(self):
         super(BaseTestCase, self).setUp()
+        self.setup_test_directory_instance()
 
         # Enabling 'use_fatal_exceptions' allows us to catch string
         # substitution format errors in exception messages.
@@ -182,6 +184,12 @@ class BaseTestCase(testtools.TestCase):
         self.addOnException(self.check_for_systemexit)
         self.orig_pid = os.getpid()
 
+    def setup_test_directory_instance(self):
+        """Give a private copy of the directory to each test."""
+        self._plugin_directory = directory._PluginDirectory()
+        mock.patch.object(directory, '_get_plugin_directory',
+                          return_value=self._plugin_directory).start()
+
     def get_new_temp_dir(self):
         """Create a new temporary directory.
 
diff --git a/neutron_lib/tests/unit/plugins/__init__.py b/neutron_lib/tests/unit/plugins/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/neutron_lib/tests/unit/plugins/test_directory.py b/neutron_lib/tests/unit/plugins/test_directory.py
new file mode 100644
index 000000000..761f50f4d
--- /dev/null
+++ b/neutron_lib/tests/unit/plugins/test_directory.py
@@ -0,0 +1,95 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from neutron_lib.plugins import directory
+from neutron_lib.tests import _base as base
+
+
+def fake_plugin():
+    pass
+
+
+class DirectoryTestCase(base.BaseTestCase):
+
+    def test__create_plugin_directory(self):
+        self.assertIsNotNone(directory._create_plugin_directory())
+
+    def test__get_plugin_directory(self):
+        self.assertIsNotNone(directory._get_plugin_directory())
+
+    def test_add_plugin(self):
+        directory.add_plugin('foo', fake_plugin)
+        self.assertIn('foo', directory.get_plugins())
+
+    def test_get_plugin_core_none(self):
+        self.assertIsNone(directory.get_plugin())
+
+    def test_get_plugin_alias_none(self):
+        self.assertIsNone(directory.get_plugin('foo'))
+
+    def test_get_plugin_core(self):
+        directory.add_plugin('CORE', fake_plugin)
+        self.assertIsNotNone(directory.get_plugin())
+
+    def test_get_plugin_alias(self):
+        directory.add_plugin('foo', fake_plugin)
+        self.assertIsNotNone(directory.get_plugin('foo'))
+
+    def test_get_plugins_none(self):
+        self.assertFalse(directory.get_plugins())
+
+    def test_get_unique_plugins_none(self):
+        self.assertFalse(directory.get_unique_plugins())
+
+    def test_get_plugins(self):
+        directory.add_plugin('CORE', fake_plugin)
+        self.assertIsNotNone(directory.get_plugins())
+
+    def test_get_unique_plugins(self):
+        directory.add_plugin('foo1', fake_plugin)
+        directory.add_plugin('foo2', fake_plugin)
+        self.assertEqual(1, len(directory.get_unique_plugins()))
+
+
+class PluginDirectoryTestCase(base.BaseTestCase):
+
+    def setUp(self):
+        super(PluginDirectoryTestCase, self).setUp()
+        self.plugin_directory = directory._PluginDirectory()
+
+    def test_add_plugin(self):
+        self.plugin_directory.add_plugin('foo', 'bar')
+        self.assertEqual(1, len(self.plugin_directory._plugins))
+
+    def test_get_plugin_not_found(self):
+        self.assertIsNone(self.plugin_directory.get_plugin('foo'))
+
+    def test_get_plugin_found(self):
+        self.plugin_directory._plugins = {'foo': lambda *x, **y: 'bar'}
+        plugin = self.plugin_directory.get_plugin('foo')
+        self.assertEqual('bar', plugin())
+
+    def test_plugins(self):
+        self.plugin_directory._plugins = {'foo': lambda *x, **y: 'bar'}
+        self.assertIsNotNone(self.plugin_directory.plugins)
+
+    def test_unique_plugins(self):
+        self.plugin_directory._plugins = {
+            'foo1': fake_plugin,
+            'foo2': fake_plugin,
+        }
+        self.assertEqual(1, len(self.plugin_directory.unique_plugins))
+
+    def test_is_loaded(self):
+        self.assertFalse(self.plugin_directory.is_loaded)
+        self.plugin_directory._plugins = {'foo': lambda *x, **y: 'bar'}
+        self.assertTrue(self.plugin_directory.is_loaded)
diff --git a/requirements.txt b/requirements.txt
index 6b003b718..788adf853 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -6,6 +6,7 @@ pbr>=1.6 # Apache-2.0
 
 SQLAlchemy<1.1.0,>=1.0.10 # MIT
 debtcollector>=1.2.0 # Apache-2.0
+oslo.concurrency>=3.8.0 # Apache-2.0
 oslo.config!=3.18.0,>=3.14.0 # Apache-2.0
 oslo.context>=2.9.0 # Apache-2.0
 oslo.db!=4.13.1,!=4.13.2,>=4.10.0 # Apache-2.0