diff --git a/pkg/daemon/ceph/client/filesystem_test.go b/pkg/daemon/ceph/client/filesystem_test.go
index c0500993b..4b66dfec5 100644
--- a/pkg/daemon/ceph/client/filesystem_test.go
+++ b/pkg/daemon/ceph/client/filesystem_test.go
@@ -149,6 +149,17 @@ func TestFilesystemRemove(t *testing.T) {
 		}
 		return "", fmt.Errorf("unexpected ceph command '%v'", args)
 	}
+	executor.MockExecuteCommandWithOutput = func(debug bool, actionName string, command string, args ...string) (string, error) {
+		emptyPool := "{\"images\":{\"count\":0,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+
+		if args[0] == "pool" {
+			if args[1] == "stats" {
+				return emptyPool, nil
+			}
+		}
+		return "", fmt.Errorf("unexpected rbd command '%v'", args)
+	}
+
 	err := RemoveFilesystem(context, "ns", fs.MDSMap.FilesystemName)
 	assert.Nil(t, err)
 	assert.True(t, metadataDeleted)
diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go
index 2644139d6..a2c3b6360 100644
--- a/pkg/daemon/ceph/client/pool.go
+++ b/pkg/daemon/ceph/client/pool.go
@@ -21,10 +21,9 @@ import (
 	"strconv"
 	"strings"
 
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/model"
-
-	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 )
 
 const (
@@ -65,6 +64,19 @@ type CephStoragePoolStats struct {
 	} `json:"pools"`
 }
 
+type PoolStatistics struct {
+	Images struct {
+		Count            int `json:"count"`
+		ProvisionedBytes int `json:"provisioned_bytes"`
+		SnapCount        int `json:"snap_count"`
+	} `json:"images"`
+	Trash struct {
+		Count            int `json:"count"`
+		ProvisionedBytes int `json:"provisioned_bytes"`
+		SnapCount        int `json:"snap_count"`
+	} `json:"trash"`
+}
+
 func ListPoolSummaries(context *clusterd.Context, clusterName string) ([]CephStoragePoolSummary, error) {
 	args := []string{"osd", "lspools"}
 	buf, err := NewCephCommand(context, clusterName, args).Run()
@@ -152,6 +164,25 @@ func CreatePoolWithProfile(context *clusterd.Context, clusterName string, newPoo
 	)
 }
 
+func checkForImagesInPool(context *clusterd.Context, name, namespace string) error {
+	var err error
+	var stats = new(PoolStatistics)
+	logger.Infof("checking any images/snapshosts present in pool %s", name)
+	stats, err = GetPoolStatistics(context, name, namespace)
+	if err != nil {
+		if strings.Contains(err.Error(), "No such file or directory") {
+			return nil
+		}
+		return fmt.Errorf("failed to list images/snapshosts in pool %s. %+v", name, err)
+	}
+	if stats.Images.Count == 0 && stats.Images.SnapCount == 0 {
+		logger.Infof("no images/snapshosts present in pool %s", name)
+		return nil
+	}
+
+	return fmt.Errorf("pool %s contains images/snapshosts", name)
+}
+
 func DeletePool(context *clusterd.Context, clusterName string, name string) error {
 	// check if the pool exists
 	pool, err := GetPoolDetails(context, clusterName, name)
@@ -160,6 +191,10 @@ func DeletePool(context *clusterd.Context, clusterName string, name string) erro
 		return nil
 	}
 
+	err = checkForImagesInPool(context, name, clusterName)
+	if err != nil {
+		return fmt.Errorf("failed to delete pool %s. %+v", name, err)
+	}
 	logger.Infof("purging pool %s (id=%d)", name, pool.Number)
 	args := []string{"osd", "pool", "delete", name, name, reallyConfirmFlag}
 	_, err = NewCephCommand(context, clusterName, args).Run()
@@ -294,6 +329,23 @@ func GetPoolStats(context *clusterd.Context, clusterName string) (*CephStoragePo
 	return &poolStats, nil
 }
 
+func GetPoolStatistics(context *clusterd.Context, name, clusterName string) (*PoolStatistics, error) {
+	args := []string{"pool", "stats", name}
+	cmd := NewRBDCommand(context, clusterName, args)
+	cmd.JsonOutput = true
+	buf, err := cmd.Run()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pool stats: %+v", err)
+	}
+
+	var poolStats PoolStatistics
+	if err := json.Unmarshal(buf, &poolStats); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal pool stats response: %+v", err)
+	}
+
+	return &poolStats, nil
+}
+
 func GetPools(context *clusterd.Context, clusterName string) ([]model.Pool, error) {
 	// list pool summaries using the ceph client
 	cephPoolSummaries, err := ListPoolSummaries(context, clusterName)
diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go
index 005cef7a1..f9851c0a6 100644
--- a/pkg/daemon/ceph/client/pool_test.go
+++ b/pkg/daemon/ceph/client/pool_test.go
@@ -17,6 +17,7 @@ package client
 
 import (
 	"fmt"
+	"reflect"
 	"testing"
 
 	"github.com/rook/rook/pkg/daemon/ceph/model"
@@ -172,3 +173,38 @@ func testIsStringInSlice(a string, list []string) bool {
 	}
 	return false
 }
+
+func TestGetPoolStatistics(t *testing.T) {
+	p := PoolStatistics{}
+	p.Images.Count = 1
+	p.Images.ProvisionedBytes = 1024
+	p.Images.SnapCount = 1
+	p.Trash.Count = 1
+	p.Trash.ProvisionedBytes = 2048
+	p.Trash.SnapCount = 0
+	executor := &exectest.MockExecutor{}
+	context := &clusterd.Context{Executor: executor}
+	executor.MockExecuteCommandWithOutput = func(debug bool, actionName string, command string, args ...string) (string, error) {
+		a := "{\"images\":{\"count\":1,\"provisioned_bytes\":1024,\"snap_count\":1},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+		logger.Infof("Command: %s %v", command, args)
+
+		if args[0] == "pool" {
+			if args[1] == "stats" {
+				if args[2] == "replicapool" {
+					return a, nil
+				}
+				return "", fmt.Errorf("rbd:error opening pool '%s': (2) No such file or directory", args[3])
+
+			}
+		}
+		return "", fmt.Errorf("unexpected rbd command '%v'", args)
+	}
+
+	stats, err := GetPoolStatistics(context, "replicapool", "cluster")
+	assert.Nil(t, err)
+	assert.True(t, reflect.DeepEqual(stats, &p))
+
+	stats, err = GetPoolStatistics(context, "rbd", "cluster")
+	assert.NotNil(t, err)
+	assert.Nil(t, stats)
+}
diff --git a/pkg/operator/ceph/object/objectstore_test.go b/pkg/operator/ceph/object/objectstore_test.go
index c1991296f..5a2faf49a 100644
--- a/pkg/operator/ceph/object/objectstore_test.go
+++ b/pkg/operator/ceph/object/objectstore_test.go
@@ -138,6 +138,13 @@ func deleteStore(t *testing.T, name string, existingStores string, expectedDelet
 			zoneDeleted = true
 			return "", nil
 		}
+
+		if args[0] == "pool" {
+			if args[1] == "stats" {
+				emptyPool := "{\"images\":{\"count\":0,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+				return emptyPool, nil
+			}
+		}
 		return "", fmt.Errorf("unexpected ceph command '%v'", args)
 	}
 	executor.MockExecuteCommandWithOutput = executorFunc
diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go
index 0860ce25d..646be9dc0 100644
--- a/pkg/operator/ceph/pool/controller.go
+++ b/pkg/operator/ceph/pool/controller.go
@@ -143,7 +143,6 @@ func (c *PoolController) onDelete(obj interface{}) {
 		logger.Errorf("failed to get pool object: %+v", err)
 		return
 	}
-
 	if err := deletePool(c.context, pool); err != nil {
 		logger.Errorf("failed to delete pool %s. %+v", pool.ObjectMeta.Name, err)
 	}
diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go
index ac80c9c82..195730216 100644
--- a/pkg/operator/ceph/pool/controller_test.go
+++ b/pkg/operator/ceph/pool/controller_test.go
@@ -152,6 +152,7 @@ func TestUpdatePool(t *testing.T) {
 }
 
 func TestDeletePool(t *testing.T) {
+	failOnDelete := false
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutputFile: func(debug bool, actionName, command, outfile string, args ...string) (string, error) {
 			if command == "ceph" && args[1] == "lspools" {
@@ -159,8 +160,28 @@ func TestDeletePool(t *testing.T) {
 			} else if command == "ceph" && args[1] == "pool" && args[2] == "get" {
 				return `{"pool": "mypool","pool_id": 1,"size":1}`, nil
 			}
+
 			return "", nil
 		},
+		MockExecuteCommandWithOutput: func(debug bool, actionName string, command string, args ...string) (string, error) {
+			emptyPool := "{\"images\":{\"count\":0,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+			p := "{\"images\":{\"count\":1,\"provisioned_bytes\":1024,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+			logger.Infof("Command: %s %v", command, args)
+
+			if args[0] == "pool" {
+				if args[1] == "stats" {
+					if !failOnDelete {
+						return emptyPool, nil
+					}
+
+					return p, nil
+
+				}
+				return "", fmt.Errorf("rbd: error opening pool '%s': (2) No such file or directory", args[3])
+
+			}
+			return "", fmt.Errorf("unexpected rbd command '%v'", args)
+		},
 	}
 	context := &clusterd.Context{Executor: executor}
 
@@ -179,6 +200,15 @@ func TestDeletePool(t *testing.T) {
 	assert.False(t, exists)
 	err = deletePool(context, p)
 	assert.Nil(t, err)
+
+	// fail if images/snapshosts exist in the pool
+	failOnDelete = true
+	p = &cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
+	exists, err = poolExists(context, p)
+	assert.Nil(t, err)
+	assert.True(t, exists)
+	err = deletePool(context, p)
+	assert.NotNil(t, err)
 }
 
 func TestGetPoolObject(t *testing.T) {