diff --git a/releasenotes/notes/cluster-notifications-fd205f5f0148b052.yaml b/releasenotes/notes/cluster-notifications-fd205f5f0148b052.yaml
new file mode 100644
index 0000000000..2e5ff89666
--- /dev/null
+++ b/releasenotes/notes/cluster-notifications-fd205f5f0148b052.yaml
@@ -0,0 +1,8 @@
+---
+features:
+  - Adds new fields "instance_ids", which is supposed to contain ids of
+    cluster instances, in payloads of two cluster events -
+    DBaaSClusterShrink and DBaaSClusterGrow. Moreover, additional end
+    notifications after growing and shrinking cluster have been added.
+    It allows better integration with tools for monitoring resources
+    usage.
diff --git a/trove/cluster/models.py b/trove/cluster/models.py
index 27caacec2d..db6cfd1935 100644
--- a/trove/cluster/models.py
+++ b/trove/cluster/models.py
@@ -331,7 +331,9 @@ class Cluster(object):
                 return self.grow(instances)
         elif action == 'shrink':
             context.notification = DBaaSClusterShrink(context, request=req)
-            with StartNotification(context, cluster_id=self.id):
+            instance_ids = [instance['id'] for instance in param]
+            with StartNotification(context, cluster_id=self.id,
+                                   instance_ids=instance_ids):
                 instance_ids = [instance['id'] for instance in param]
                 return self.shrink(instance_ids)
         elif action == "reset-status":
diff --git a/trove/common/notification.py b/trove/common/notification.py
index 49a3aad519..70d2071f4b 100644
--- a/trove/common/notification.py
+++ b/trove/common/notification.py
@@ -649,6 +649,10 @@ class DBaaSClusterGrow(DBaaSAPINotification):
     def required_start_traits(self):
         return ['cluster_id']
 
+    @abc.abstractmethod
+    def required_end_traits(self):
+        return ['cluster_id']
+
 
 class DBaaSClusterShrink(DBaaSAPINotification):
 
@@ -660,6 +664,10 @@ class DBaaSClusterShrink(DBaaSAPINotification):
     def required_start_traits(self):
         return ['cluster_id']
 
+    @abc.abstractmethod
+    def required_end_traits(self):
+        return ['cluster_id']
+
 
 class DBaaSBackupCreate(DBaaSAPINotification):
 
diff --git a/trove/taskmanager/manager.py b/trove/taskmanager/manager.py
index 7559c1ecf9..8d8ac1b660 100644
--- a/trove/taskmanager/manager.py
+++ b/trove/taskmanager/manager.py
@@ -428,12 +428,16 @@ class Manager(periodic_task.PeriodicTasks):
             cluster_tasks.create_cluster(context, cluster_id)
 
     def grow_cluster(self, context, cluster_id, new_instance_ids):
-        cluster_tasks = models.load_cluster_tasks(context, cluster_id)
-        cluster_tasks.grow_cluster(context, cluster_id, new_instance_ids)
+        with EndNotification(context, cluster_id=cluster_id,
+                             instance_ids=new_instance_ids):
+            cluster_tasks = models.load_cluster_tasks(context, cluster_id)
+            cluster_tasks.grow_cluster(context, cluster_id, new_instance_ids)
 
     def shrink_cluster(self, context, cluster_id, instance_ids):
-        cluster_tasks = models.load_cluster_tasks(context, cluster_id)
-        cluster_tasks.shrink_cluster(context, cluster_id, instance_ids)
+        with EndNotification(context, cluster_id=cluster_id,
+                             instance_ids=instance_ids):
+            cluster_tasks = models.load_cluster_tasks(context, cluster_id)
+            cluster_tasks.shrink_cluster(context, cluster_id, instance_ids)
 
     def restart_cluster(self, context, cluster_id):
         cluster_tasks = models.load_cluster_tasks(context, cluster_id)
diff --git a/trove/tests/unittests/taskmanager/test_manager.py b/trove/tests/unittests/taskmanager/test_manager.py
index 396941044c..e5bda5971c 100644
--- a/trove/tests/unittests/taskmanager/test_manager.py
+++ b/trove/tests/unittests/taskmanager/test_manager.py
@@ -14,7 +14,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from mock import Mock, patch, PropertyMock
+from mock import MagicMock, Mock, patch, PropertyMock
 from proboscis.asserts import assert_equal
 
 from trove.backup.models import Backup
@@ -274,6 +274,42 @@ class TestManager(trove_testtools.TestCase):
         mock_tasks.delete_cluster.assert_called_with(self.context,
                                                      'some-cluster-id')
 
+    def test_shrink_cluster_with_success(self):
+        self._assert_shrink_cluster(True)
+
+    def test_shrink_cluster_with_error(self):
+        self._assert_shrink_cluster(False)
+
+    @patch('trove.taskmanager.manager.EndNotification')
+    @patch('trove.taskmanager.manager.models.load_cluster_tasks')
+    def _assert_shrink_cluster(self, success, mock_load, mock_notification):
+        if success:
+            mock_load.side_effect = Mock()
+        else:
+            mock_load.side_effect = Exception
+
+        end_notification = MagicMock()
+        mock_notification.return_value = end_notification
+        context = Mock()
+        cluster_id = Mock()
+        instance_ids = Mock()
+
+        try:
+            self.manager.shrink_cluster(context, cluster_id, instance_ids)
+            self.assertTrue(success)
+        except Exception:
+            self.assertFalse(success)
+
+        mock_load.assert_called_once_with(context, cluster_id)
+        mock_notification.assert_called_once_with(context,
+                                                  cluster_id=cluster_id,
+                                                  instance_ids=instance_ids)
+        exit_error_type = end_notification.__exit__.call_args_list[0][0][0]
+        if success:
+            self.assertFalse(exit_error_type)
+        else:
+            self.assertTrue(exit_error_type)
+
 
 class TestTaskManagerService(trove_testtools.TestCase):
     def test_app_factory(self):