diff --git a/src/sonic-bmpcfgd/__init__.py b/src/sonic-bmpcfgd/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/src/sonic-bmpcfgd/data/debian/sonic-services-data.bmpcfgd.service b/src/sonic-bmpcfgd/data/debian/sonic-services-data.bmpcfgd.service
new file mode 100644
index 000000000000..0a16edf77fe1
--- /dev/null
+++ b/src/sonic-bmpcfgd/data/debian/sonic-services-data.bmpcfgd.service
@@ -0,0 +1,14 @@
+[Unit]
+Description=Process which monitors config_db and manage openbmpd daemon
+Requires=database.service config-setup.service
+After=database.service config-setup.service
+BindsTo=sonic.target
+After=sonic.target
+
+[Service]
+Type=simple
+ExecStart=/usr/local/bin/bmpcfgd
+Restart=always
+
+[Install]
+WantedBy=sonic.target
diff --git a/src/sonic-bmpcfgd/scripts/bmpcfgd b/src/sonic-bmpcfgd/scripts/bmpcfgd
new file mode 100644
index 000000000000..448ca5b24617
--- /dev/null
+++ b/src/sonic-bmpcfgd/scripts/bmpcfgd
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+'''
+bmpcfgd
+Daemon which monitors bmp relevant table enablement from CONFIG_DB, and reset BMP states
+'''
+
+import os
+import sys
+import subprocess
+import syslog
+import signal
+from shutil import copy2
+from datetime import datetime
+from sonic_py_common import device_info
+from sonic_py_common.general import check_output_pipe
+from swsscommon.swsscommon import ConfigDBConnector, DBConnector, Table
+from swsscommon import swsscommon
+from sonic_py_common.daemon_base import DaemonBase
+
+CFG_DB = "CONFIG_DB"
+BMP_STATE_DB = "BMP_STATE_DB"
+REDIS_HOSTIP = "127.0.0.1"
+BMP_TABLE = "BMP"
+
+def is_true(val):
+    return str(val).lower() == 'true'
+
+class BMPCfg(DaemonBase):
+    def __init__(self, state_db_conn):
+        DaemonBase.__init__(self, SYSLOG_IDENTIFIER)
+        self.bgp_neighbor_table  = False
+        self.bgp_rib_in_table  = False
+        self.bgp_rib_out_table  = False
+        self.state_db_conn = state_db_conn
+
+
+    def load(self, data={}):
+        common_config = data.get('table', {})
+
+        self.bgp_neighbor_table = is_true(common_config.get('bgp_neighbor_table', 'false'))
+        self.bgp_rib_in_table = is_true(common_config.get('bgp_rib_in_table', 'false'))
+        self.bgp_rib_out_table = is_true(common_config.get('bgp_rib_out_table', 'false'))
+        self.log_info(f'BMPCfg: update : {self.bgp_neighbor_table}, {self.bgp_rib_in_table}, {self.bgp_rib_out_table}')
+
+        # reset bmp table state once config is changed.
+        self.stop_bmp()
+        self.reset_bmp_table()
+        self.start_bmp()
+
+
+    def cfg_handler(self, data):
+        self.load(data)
+
+
+    def stop_bmp(self):
+        self.log_info('BMPCfg: stop bmp daemon')
+        subprocess.call(["service", "openbmpd", "stop"])
+
+
+    def reset_bmp_table(self):
+        self.log_info('BMPCfg: Reset bmp table from state_db')
+        self.state_db_conn.delete_all_by_pattern(BMP_STATE_DB, 'BGP_NEIGHBOR*')
+        self.state_db_conn.delete_all_by_pattern(BMP_STATE_DB, 'BGP_RIB_IN_TABLE*')
+        self.state_db_conn.delete_all_by_pattern(BMP_STATE_DB, 'BGP_RIB_OUT_TABLE*')
+
+
+    def start_bmp(self):
+        self.log_info('BMPCfg: start bmp daemon')
+        subprocess.call(["service", "openbmpd", "start"])
+
+
+class BMPCfgDaemon:
+    def __init__(self):
+        self.state_db_conn = swsscommon.SonicV2Connector(host=REDIS_HOSTIP)
+        self.state_db_conn.connect(BMP_STATE_DB)
+        self.config_db = ConfigDBConnector()
+        self.config_db.connect(wait_for_init=True, retry_on=True)
+        self.bmpcfg = BMPCfg(self.state_db_conn)
+
+    def bmp_handler(self, key, op, data):
+        data = self.config_db.get_table(BMP_TABLE)
+        self.bmpcfg.cfg_handler(data)
+
+    def register_callbacks(self):
+        self.config_db.subscribe(BMP_TABLE,
+                                 lambda table, key, data:
+                                     self.bmp_handler(key, op, data))
+
+def signal_handler(sig, frame):
+    if sig == signal.SIGHUP:
+        self.log_info("bmpcfgd: signal 'SIGHUP' is caught and ignoring..")
+    elif sig == signal.SIGINT:
+        self.log_info("bmpcfgd: signal 'SIGINT' is caught and exiting...")
+        sys.exit(128 + sig)
+    elif sig == signal.SIGTERM:
+        self.log_info("bmpcfgd: signal 'SIGTERM' is caught and exiting...")
+        sys.exit(128 + sig)
+    else:
+        self.log_info("bmpcfgd: invalid signal - ignoring..")
+
+
+def main():
+    signal.signal(signal.SIGTERM, signal_handler)
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGHUP, signal_handler)
+    daemon = BMPCfgDaemon()
+    daemon.register_callbacks()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/sonic-bmpcfgd/setup.cfg b/src/sonic-bmpcfgd/setup.cfg
new file mode 100644
index 000000000000..1ce2f2f07257
--- /dev/null
+++ b/src/sonic-bmpcfgd/setup.cfg
@@ -0,0 +1,5 @@
+[aliases]
+test=pytest
+[tool:pytest]
+addopts = --verbose
+python_files = tests/*.py
\ No newline at end of file
diff --git a/src/sonic-bmpcfgd/setup.py b/src/sonic-bmpcfgd/setup.py
new file mode 100644
index 000000000000..33c3fd9803e7
--- /dev/null
+++ b/src/sonic-bmpcfgd/setup.py
@@ -0,0 +1,75 @@
+from __future__ import print_function
+import sys
+from setuptools import setup
+import pkg_resources
+from packaging import version
+
+# sonic_dependencies, version requirement only supports '>='
+sonic_dependencies = ['sonic-py-common', 'sonic-utilities']
+for package in sonic_dependencies:
+    try:
+        package_dist = pkg_resources.get_distribution(package.split(">=")[0])
+    except pkg_resources.DistributionNotFound:
+        print(package + " is not found!", file=sys.stderr)
+        print("Please build and install SONiC python wheels dependencies from sonic-buildimage", file=sys.stderr)
+        exit(1)
+    if ">=" in package:
+        if version.parse(package_dist.version) >= version.parse(package.split(">=")[1]):
+            continue
+        print(package + " version not match!", file=sys.stderr)
+        exit(1)
+
+setup(
+    name = 'sonic-bmpcfgd-services',
+    version = '1.0',
+    description = 'Python services which run in the bmp container',
+    license = 'Apache 2.0',
+    author = 'SONiC Team',
+    author_email = 'linuxnetdev@microsoft.com',
+    url = 'https://github.com/Azure/sonic-buildimage',
+    maintainer = 'Feng Pan',
+    maintainer_email = 'fenpan@microsoft.com',
+    packages = setuptools.find_packages(),
+    scripts = [
+        'scripts/bmpcfgd'
+    ],
+    install_requires = [
+        'jinja2>=2.10',
+        'netaddr==0.8.0',
+        'pyyaml==6.0.1',
+        'ipaddress==1.0.23'
+    ] + sonic_dependencies,
+    setup_requires = [
+        'pytest-runner',
+        'wheel'
+    ],
+    tests_require = [
+        'parameterized',
+        'pytest',
+        'pyfakefs',
+        'sonic-py-common',
+        'pytest-cov'
+    ],
+    extras_require = {
+        "testing": [
+            'parameterized',
+            'pytest',
+            'pyfakefs',
+            'sonic-py-common'
+        ]
+    },
+    classifiers = [
+        'Development Status :: 3 - Alpha',
+        'Environment :: Console',
+        'Intended Audience :: Developers',
+        'Intended Audience :: Information Technology',
+        'Intended Audience :: System Administrators',
+        'License :: OSI Approved :: Apache Software License',
+        'Natural Language :: English',
+        'Operating System :: POSIX :: Linux',
+        'Programming Language :: Python :: 3.7',
+        'Topic :: System',
+    ],
+    keywords = 'sonic SONiC bmp services',
+    test_suite = 'setup.get_test_suite'
+)
diff --git a/src/sonic-bmpcfgd/tests/__init__.py b/src/sonic-bmpcfgd/tests/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/src/sonic-bmpcfgd/tests/bmpcfgd_test.py b/src/sonic-bmpcfgd/tests/bmpcfgd_test.py
new file mode 100644
index 000000000000..a6abc6c3b703
--- /dev/null
+++ b/src/sonic-bmpcfgd/tests/bmpcfgd_test.py
@@ -0,0 +1,116 @@
+import importlib.machinery
+import importlib.util
+import filecmp
+import json
+import shutil
+import os
+import sys
+import signal
+from swsscommon import swsscommon
+
+from parameterized import parameterized
+from unittest import TestCase, mock
+from tests.common.mock_configdb import MockConfigDb, MockDBConnector
+from tests.common.mock_bootloader import MockBootloader
+from sonic_py_common.general import getstatusoutput_noshell
+from .mock_connector import MockConnector
+from sonic_py_common.general import load_module_from_source
+from mock import patch
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(test_path)
+scripts_path = os.path.join(modules_path, "scripts")
+sys.path.insert(0, modules_path)
+
+# Load the file under test
+bmpcfgd_path = os.path.join(scripts_path, 'bmpcfgd')
+bmpcfgd = load_module_from_source('bmpcfgd', bmpcfgd_path)
+
+
+from bmpcfgd import signal_handler
+
+original_syslog = bmpcfgd.syslog
+
+# Mock swsscommon classes
+bmpcfgd.ConfigDBConnector = MockConfigDb
+bmpcfgd.DBConnector = MockDBConnector
+bmpcfgd.Table = mock.Mock()
+swsscommon.SonicV2Connector = MockConnector
+
+class TestBMPCfgDaemon(TestCase):
+    """
+        Test bmpcfgd daemon
+    """
+    def setUp(self):
+        self.test_data = {}
+        self.test_data['BMP'] = {}
+        self.test_data['BMP']['table'] = {'bgp_neighbor_table': 'false', 'bgp_rib_in_table': 'false', 'bgp_rib_out_table': 'false'}
+
+    @mock.patch('sonic_installer.bootloader.get_bootloader', side_effect=[MockBootloader()])
+    @mock.patch('syslog.syslog')
+    @mock.patch('subprocess.call')
+    def test_bmpcfgd_neighbor_enable(self, mock_check_call, mock_syslog, mock_get_bootloader):
+        self.test_data['BMP']['table']['bgp_neighbor_table'] = 'true'
+        MockConfigDb.set_config_db(self.test_data)
+        bmp_config_daemon = bmpcfgd.BMPCfgDaemon()
+        bmp_config_daemon.register_callbacks()
+        bmp_config_daemon.bmp_handler("BMP", '', self.test_data)
+        expected_calls = [
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: update : True, False, False'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: stop bmp daemon'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: Reset bmp table from state_db'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: start bmp daemon'),
+        ]
+        mock_syslog.assert_has_calls(expected_calls)
+
+    @mock.patch('sonic_installer.bootloader.get_bootloader', side_effect=[MockBootloader()])
+    @mock.patch('syslog.syslog')
+    @mock.patch('subprocess.check_call')
+    def test_bmpcfgd_bgp_rib_in_enable(self, mock_check_call, mock_syslog, mock_get_bootloader):
+        self.test_data['BMP']['table']['bgp_rib_in_table'] = 'true'
+        MockConfigDb.set_config_db(self.test_data)
+        bmp_config_daemon = bmpcfgd.BMPCfgDaemon()
+        bmp_config_daemon.bmp_handler("BMP", '', self.test_data)
+        expected_calls = [
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: update : False, True, False'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: stop bmp daemon'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: Reset bmp table from state_db'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: start bmp daemon'),
+        ]
+        mock_syslog.assert_has_calls(expected_calls)
+
+    @mock.patch('sonic_installer.bootloader.get_bootloader', side_effect=[MockBootloader()])
+    @mock.patch('syslog.syslog')
+    @mock.patch('subprocess.check_call')
+    def test_bmpcfgd_bgp_rib_out_enable(self, mock_check_call, mock_syslog, mock_get_bootloader):
+        self.test_data['BMP']['table']['bgp_rib_out_table'] = 'true'
+        MockConfigDb.set_config_db(self.test_data)
+        bmp_config_daemon = bmpcfgd.BMPCfgDaemon()
+        bmp_config_daemon.bmp_handler("BMP", '', self.test_data)
+        expected_calls = [
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: update : False, False, True'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: stop bmp daemon'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: Reset bmp table from state_db'),
+            mock.call(original_syslog.LOG_INFO, 'BMPCfg: start bmp daemon'),
+        ]
+        mock_syslog.assert_has_calls(expected_calls)
+
+
+    @mock.patch('syslog.syslog')
+    @mock.patch.object(sys, 'exit')
+    def test_signal_handler(self, mock_exit, mock_syslog):
+        # Test SIGHUP signal
+        signal_handler(signal.SIGHUP, None)
+        mock_syslog.assert_called_with(original_syslog.LOG_INFO, "bmpcfgd: signal 'SIGHUP' is caught and ignoring..")
+        mock_exit.assert_not_called()
+        # Test SIGINT signal
+        signal_handler(signal.SIGINT, None)
+        mock_syslog.assert_called_with(original_syslog.LOG_INFO, "bmpcfgd: signal 'SIGINT' is caught and exiting...")
+        mock_exit.assert_called_once_with(128 + signal.SIGINT)
+        # Test SIGTERM signal
+        signal_handler(signal.SIGTERM, None)
+        mock_syslog.assert_called_with(original_syslog.LOG_INFO, "bmpcfgd: signal 'SIGTERM' is caught and exiting...")
+        mock_exit.assert_called_with(128 + signal.SIGTERM)
+        # Test invalid signal
+        signal_handler(999, None)
+        mock_syslog.assert_called_with(original_syslog.LOG_INFO, "bmpcfgd: invalid signal - ignoring..")
diff --git a/src/sonic-bmpcfgd/tests/common/mock_configdb.py b/src/sonic-bmpcfgd/tests/common/mock_configdb.py
new file mode 100644
index 000000000000..d6c3c055bc57
--- /dev/null
+++ b/src/sonic-bmpcfgd/tests/common/mock_configdb.py
@@ -0,0 +1,162 @@
+class MockConfigDb(object):
+    """
+        Mock Config DB which responds to data tables requests and store updates to the data table
+    """
+    STATE_DB = None
+    CONFIG_DB = None
+    event_queue = []
+
+    def __init__(self, **kwargs):
+        self.handlers = {}
+
+    @staticmethod
+    def set_config_db(test_config_db):
+        MockConfigDb.CONFIG_DB = test_config_db
+
+    @staticmethod
+    def mod_config_db(test_config_db):
+        MockConfigDb.CONFIG_DB.update(test_config_db)
+
+    @staticmethod
+    def deserialize_key(key, separator="|"):
+        tokens = key.split(separator)
+        if len(tokens) > 1:
+            return tuple(tokens)
+        else:
+            return key
+
+    @staticmethod
+    def get_config_db():
+        return MockConfigDb.CONFIG_DB
+
+    def connect(self, wait_for_init=True, retry_on=True):
+        pass
+
+    def close(self, db_name):
+        pass
+
+    def get(self, db_id, key, field):
+        return MockConfigDb.CONFIG_DB[key][field]
+
+    def get_entry(self, key, field):
+        return MockConfigDb.CONFIG_DB[key][field]
+
+    def mod_entry(self, key, field, data):
+        existing_data = self.get_entry(key, field)
+        existing_data.update(data)
+        self.set_entry(key, field, existing_data)
+
+    def set_entry(self, key, field, data):
+        MockConfigDb.CONFIG_DB[key][field] = data
+
+    def get_table(self, table_name):
+        data = {}
+        if table_name in MockConfigDb.CONFIG_DB:
+            for k, v in MockConfigDb.CONFIG_DB[table_name].items():
+                data[self.deserialize_key(k)] = v
+        return data
+
+    def subscribe(self, table_name, callback):
+        self.handlers[table_name] = callback
+
+    def publish(self, table_name, key, op, data):
+        self.handlers[table_name](key, op, data)
+
+    def listen(self, init_data_handler=None):
+        for e in MockConfigDb.event_queue:
+            self.handlers[e[0]](e[0], e[1], self.get_entry(e[0], e[1]))
+
+
+class MockSelect():
+
+    event_queue = []
+    OBJECT = "OBJECT"
+    TIMEOUT = "TIMEOUT"
+    ERROR = ""
+    NUM_TIMEOUT_TRIES = 0
+
+    @staticmethod
+    def set_event_queue(Q):
+        MockSelect.event_queue = Q
+
+    @staticmethod
+    def get_event_queue():
+        return MockSelect.event_queue
+
+    @staticmethod
+    def reset_event_queue():
+        MockSelect.event_queue = []
+
+    def __init__(self):
+        self.sub_map = {}
+        self.TIMEOUT = "TIMEOUT"
+        self.ERROR = "ERROR"
+
+    def addSelectable(self, subscriber):
+        self.sub_map[subscriber.table] = subscriber
+
+    def select(self, TIMEOUT):
+        if not MockSelect.get_event_queue() and MockSelect.NUM_TIMEOUT_TRIES == 0:
+            raise TimeoutError
+        elif MockSelect.NUM_TIMEOUT_TRIES != 0:
+            MockSelect.NUM_TIMEOUT_TRIES = MockSelect.NUM_TIMEOUT_TRIES - 1
+            return MockSelect.TIMEOUT, 0
+        
+        table, key = MockSelect.get_event_queue().pop(0)
+        self.sub_map[table].nextKey(key)
+        return "OBJECT", self.sub_map[table]
+
+
+class MockSubscriberStateTable():
+
+    FD_INIT = 0
+
+    @staticmethod
+    def generate_fd():
+        curr = MockSubscriberStateTable.FD_INIT
+        MockSubscriberStateTable.FD_INIT = curr + 1
+        return curr
+
+    @staticmethod
+    def reset_fd():
+        MockSubscriberStateTable.FD_INIT = 0
+
+    def __init__(self, conn, table, pop=None, pri=None):
+        self.fd = MockSubscriberStateTable.generate_fd()
+        self.next_key = ''
+        self.table = table
+
+    def getFd(self):
+        return self.fd
+
+    def nextKey(self, key):
+        print("next key")
+        self.next_key = key
+
+    def pop(self):
+        table = MockConfigDb.CONFIG_DB.get(self.table, {})
+        print(self.next_key)
+        if self.next_key not in table:
+            op = "DEL"
+            fvs = {}
+        else:
+            op = "SET"
+            fvs = table.get(self.next_key, {})
+        return self.next_key, op, fvs
+
+
+class MockDBConnector():
+    def __init__(self, db, val, tcpFlag=False, name=None):
+        self.data = {}
+
+    def hget(self, key, field):
+        if key not in self.data:
+            return None
+        if field not in self.data[key]:
+            return None
+        return self.data[key][field]
+
+    def hset(self, key, field, value):
+        if key not in self.data:
+            self.data[key] = {}
+        self.data[key][field] = value
diff --git a/src/sonic-bmpcfgd/tests/mock_connector.py b/src/sonic-bmpcfgd/tests/mock_connector.py
new file mode 100644
index 000000000000..71490796c823
--- /dev/null
+++ b/src/sonic-bmpcfgd/tests/mock_connector.py
@@ -0,0 +1,39 @@
+class MockConnector(object):
+    STATE_DB = None
+    data = {}
+
+    def __init__(self, host):
+        pass
+
+    def connect(self, db_id):
+        pass
+
+    def get(self, db_id, key, field):
+        return MockConnector.data[key][field]
+
+    def set(self, db_id, key, field, value):
+        if key not in MockConnector.data:
+            MockConnector.data[key] = {}
+        MockConnector.data[key][field] = value
+
+    def keys(self, db_id, pattern):
+        match = pattern.split('*')[0]
+        ret = []
+        for key in MockConnector.data.keys():
+            if match in key:
+                ret.append(key)
+
+        return ret
+
+    def get_all(self, db_id, key):
+        return MockConnector.data[key]
+
+    def delete(self, db_id, key):
+        return MockConnector.data.delete(key)
+
+    def delete_all_by_pattern(self, db_id, pattern):
+        keys = self.keys(db_id, pattern)
+        for key in keys:
+            self.delete(db_id, key)
+
+