diff --git a/delfin/api/v1/storages.py b/delfin/api/v1/storages.py
old mode 100644
new mode 100755
index 563e48897..79cadd3c5
--- a/delfin/api/v1/storages.py
+++ b/delfin/api/v1/storages.py
@@ -13,7 +13,9 @@
 # limitations under the License.
 
 import copy
+import json
 
+import six
 from oslo_config import cfg
 from oslo_log import log
 from oslo_utils import timeutils
@@ -26,7 +28,7 @@
 from delfin.api.common import wsgi
 from delfin.api.schemas import storages as schema_storages
 from delfin.api.views import storages as storage_view
-from delfin.common import constants
+from delfin.common import constants, config
 from delfin.drivers import api as driverapi
 from delfin.i18n import _
 from delfin.task_manager import rpcapi as task_rpcapi
@@ -114,6 +116,7 @@ def delete(self, req, id):
                 storage['id'],
                 subclass.__module__ + '.' + subclass.__name__)
         self.task_rpcapi.remove_storage_in_cache(ctxt, storage['id'])
+        self._unregister_perf_collection(storage['id'])
 
     @wsgi.response(202)
     def sync_all(self, req):
@@ -192,6 +195,48 @@ def _storage_exist(self, context, access_info):
 
         return False
 
+    def _unregister_perf_collection(self, storage_id):
+
+        schedule = config.Scheduler.getInstance()
+
+        # The path of scheduler config file
+        config_file = CONF.scheduler.config_path
+
+        try:
+            # Load the scheduler configuration file
+            data = config.load_json_file(config_file)
+            for storage in data.get("storages"):
+                config_storage_id = storage.get('id')
+                if config_storage_id == storage_id:
+                    for resource in storage.keys():
+                        # Skip storage id attribute and
+                        # check for all metric collection jobs
+                        if resource == 'id':
+                            continue
+                        job_id = storage_id + resource
+
+                        if schedule.get_job(job_id):
+                            schedule.remove_job(job_id)
+
+                    # Remove the entry for storage being deleted and
+                    # update schedular config file
+                    data['storages'].remove(storage)
+                    with open(config_file, "w") as jsonFile:
+                        json.dump(data, jsonFile)
+                        jsonFile.close()
+                    break
+        except TypeError:
+            LOG.error("Failed to unregister performance collection. Error "
+                      "occurred during parsing of config file")
+        except json.decoder.JSONDecodeError:
+            msg = ("Failed to unregister performance collection. Not able to "
+                   "open the config file: {0} ".format(config_file))
+            LOG.error(msg)
+        except Exception as e:
+            msg = _('Failed to unregister performance collection. Reason: {0}'
+                    .format(six.text_type(e)))
+            LOG.error(msg)
+
 
 def create_resource():
     return wsgi.Resource(StorageController())
diff --git a/delfin/tests/unit/api/v1/test_storages.py b/delfin/tests/unit/api/v1/test_storages.py
index 9bac91b14..f6ca67e68 100644
--- a/delfin/tests/unit/api/v1/test_storages.py
+++ b/delfin/tests/unit/api/v1/test_storages.py
@@ -22,6 +22,15 @@
 from delfin.api.v1.storages import StorageController
 from delfin.tests.unit.api import fakes
 
+fake_schedular_config = {
+    "storages": [{"id": "fake_id",
+                  "array_polling": {"perf_collection": True, "interval": 12,
+                                    "is_historic": True}}]}
+
+invalid_schedular_config = '"storages": [{"id": "fake_id","array_polling": {' \
+                           '"perf_collection": True, "interval": 12,' \
+                           '"is_historic": True}}]} '
+
 
 class TestStorageController(test.TestCase):
 
@@ -35,8 +44,24 @@ def setUp(self):
 
     @mock.patch.object(db, 'storage_get',
                        mock.Mock(return_value={'id': 'fake_id'}))
-    def test_delete(self):
+    @mock.patch('delfin.common.config.load_json_file')
+    def test_delete(self, mock_load_json_file):
+        req = fakes.HTTPRequest.blank('/storages/fake_id')
+        mock_load_json_file.return_value = fake_schedular_config
+        self.controller.delete(req, 'fake_id')
+        ctxt = req.environ['delfin.context']
+        db.storage_get.assert_called_once_with(ctxt, 'fake_id')
+        self.task_rpcapi.remove_storage_resource.assert_called_with(
+            ctxt, 'fake_id', mock.ANY)
+        self.task_rpcapi.remove_storage_in_cache.assert_called_once_with(
+            ctxt, 'fake_id')
+
+    @mock.patch.object(db, 'storage_get',
+                       mock.Mock(return_value={'id': 'fake_id'}))
+    @mock.patch('delfin.common.config.load_json_file')
+    def test_delete_invalid_schedular_conf(self, mock_load_json_file):
         req = fakes.HTTPRequest.blank('/storages/fake_id')
+        mock_load_json_file.return_value = invalid_schedular_config
         self.controller.delete(req, 'fake_id')
         ctxt = req.environ['delfin.context']
         db.storage_get.assert_called_once_with(ctxt, 'fake_id')