diff --git a/database.cc b/database.cc
index b40ae4ba10..cc2c09305f 100644
--- a/database.cc
+++ b/database.cc
@@ -1798,6 +1798,36 @@ const sstring& database::get_snitch_name() const {
     return _cfg->endpoint_snitch();
 }
 
+// For the filesystem operations, this code will assume that all keyspaces are visible in all shards
+// (as we have been doing for a lot of the other operations, like the snapshot itself).
+future<> database::clear_snapshot(sstring tag, std::vector<sstring> keyspace_names) {
+    std::vector<std::reference_wrapper<keyspace>> keyspaces;
+
+    if (keyspace_names.empty()) {
+        // if keyspace names are not given - apply to all existing local keyspaces
+        for (auto& ks: _keyspaces) {
+            keyspaces.push_back(std::reference_wrapper<keyspace>(ks.second));
+        }
+    } else {
+        for (auto& ksname: keyspace_names) {
+            try {
+                keyspaces.push_back(std::reference_wrapper<keyspace>(find_keyspace(ksname)));
+            } catch (no_such_keyspace& e) {
+                return make_exception_future(std::current_exception());
+            }
+        }
+    }
+
+    return parallel_for_each(keyspaces, [this, tag] (auto& ks) {
+        return parallel_for_each(ks.get().metadata()->cf_meta_data(), [this, tag] (auto& pair) {
+            auto& cf = this->find_column_family(pair.second);
+            return cf.clear_snapshot(tag);
+         }).then_wrapped([] (future<> f) {
+            dblog.debug("Cleared out snapshot directories");
+         });
+    });
+}
+
 future<> update_schema_version_and_announce(distributed<service::storage_proxy>& proxy)
 {
     return db::schema_tables::calculate_schema_digest(proxy).then([&proxy] (utils::UUID uuid) {
diff --git a/database.hh b/database.hh
index b2cd279892..fb6889f5d1 100644
--- a/database.hh
+++ b/database.hh
@@ -621,6 +621,7 @@ public:
     future<> apply(const frozen_mutation&);
     keyspace::config make_keyspace_config(const keyspace_metadata& ksm);
     const sstring& get_snitch_name() const;
+    future<> clear_snapshot(sstring tag, std::vector<sstring> keyspace_names);
 
     friend std::ostream& operator<<(std::ostream& out, const database& db);
     const std::unordered_map<sstring, keyspace>& get_keyspaces() const {
diff --git a/service/storage_service.cc b/service/storage_service.cc
index b0895b2f8d..a776332b7a 100644
--- a/service/storage_service.cc
+++ b/service/storage_service.cc
@@ -1416,27 +1416,8 @@ future<> storage_service::take_column_family_snapshot(sstring ks_name, sstring c
     });
 }
 
-// For the filesystem operations, this code will assume that all keyspaces are visible in all shards
-// (as we have been doing for a lot of the other operations, like the snapshot itself).
 future<> storage_service::clear_snapshot(sstring tag, std::vector<sstring> keyspace_names) {
-    std::vector<std::reference_wrapper<keyspace>> keyspaces;
-    for (auto& ksname: keyspace_names) {
-        try {
-            keyspaces.push_back(std::reference_wrapper<keyspace>(_db.local().find_keyspace(ksname)));
-        } catch (no_such_keyspace& e) {
-            return make_exception_future(std::current_exception());
-        }
-    }
-
-    auto deleted_keyspaces = make_lw_shared<std::vector<sstring>>();
-    return parallel_for_each(keyspaces, [this, tag, deleted_keyspaces] (auto& ks) {
-        return parallel_for_each(ks.get().metadata()->cf_meta_data(), [this, tag] (auto& pair) {
-            auto& cf = _db.local().find_column_family(pair.second);
-            return cf.clear_snapshot(tag);
-         }).then_wrapped([] (future<> f) {
-            logger.debug("Cleared out snapshot directories");
-         });
-    });
+    return _db.local().clear_snapshot(tag, keyspace_names);
 }
 
 future<std::unordered_map<sstring, std::vector<service::storage_service::snapshot_details>>>