mds/quiesce: validate awaitability of quiesce requests.

Signed-off-by: Leonid Usov <leonid.usov@ibm.com>

src/mds/MDSRankQuiesce.cc

@@ -61,6 +61,16 @@ void MDSRank::command_quiesce_db(const cmdmap_t& cmdmap, std::function<void(int,
   bool op_cancel = cmd_getval_or<bool>(cmdmap, "cancel", false);
   bool all = cmd_getval_or<bool>(cmdmap, "all", false);
   std::optional<std::string> set_id = cmd_getval<std::string>(cmdmap, "set_id");
+  auto await = [&]() -> std::optional<QuiesceTimeInterval> {
+    double timeout;
+    if (cmd_getval(cmdmap, "await_for", timeout)) {
+      return duration_cast<QuiesceTimeInterval>(dd(timeout));
+    } else if (cmd_getval_or<bool>(cmdmap, "await", false)) {
+      return QuiesceTimeInterval::max();
+    } else {
+      return std::nullopt;
+    }
+  }();
 
   auto roots = cmd_getval_or<std::vector<std::string>>(cmdmap, "roots", std::vector<std::string> {});
 
@@ -71,7 +81,11 @@ void MDSRank::command_quiesce_db(const cmdmap_t& cmdmap, std::function<void(int,
     on_finish(-EINVAL, "Operations [include, exclude, reset, release, cancel, query] are mutually exclusive", bl);
     return;
   } else if (all_ops == 0) {
-    op_include = true;
+    if (roots.empty()) {
+      op_query = true;
+    } else {
+      op_include = true;
+    }
   }
 
   if ((op_release || op_cancel) && roots.size() > 0) {
@@ -110,6 +124,12 @@ void MDSRank::command_quiesce_db(const cmdmap_t& cmdmap, std::function<void(int,
     return;
   }
 
+  if ((op_query || op_cancel) && await) {
+    bufferlist bl;
+    on_finish(-EINVAL, "Operations [query, cancel] don't support `--await`", bl);
+    return;
+  }
+
   struct Ctx : public QuiesceDbManager::RequestContext {
     std::function<void(int, const std::string&, bufferlist&)> on_finish;
     bool all = false;
@@ -203,12 +223,7 @@ void MDSRank::command_quiesce_db(const cmdmap_t& cmdmap, std::function<void(int,
     }
 
     double timeout;
-
-    if (cmd_getval(cmdmap, "await_for", timeout)) {
-      r.await = duration_cast<QuiesceTimeInterval>(dd(timeout));
-    } else if (cmd_getval_or<bool>(cmdmap, "await", false)) {
-      r.await = QuiesceTimeInterval::max();
-    }
+    r.await = await;
 
     if (cmd_getval(cmdmap, "expiration", timeout)) {
       r.expiration = duration_cast<QuiesceTimeInterval>(dd(timeout));

src/mds/QuiesceDb.h

@@ -414,17 +414,21 @@ struct QuiesceDbRequest {
   bool operator==(const QuiesceDbRequest&) const = default;
 
   bool is_valid() const {
-    return control.roots_op < __INVALID && (
-        // Everything goes if a set id is provided
-        set_id
-        // or it's a new set creation, in which case the request should be including roots
-        || includes_roots()
-        // Otherwise, the allowed wildcard operations are: query and cancel all.
-        // Also, one can't await a wildcard
-        || ((is_cancel_all() || is_query()) && !await && !timeout && !expiration && !if_version)
-    );
+    return control.roots_op < __INVALID
+        && (is_awaitable() || !await) 
+        && (
+          // Everything goes if a set id is provided
+          set_id
+          // or it's a new set creation, in which case the request should be including roots
+          || includes_roots()
+          // Otherwise, the allowed wildcard operations are: query and cancel all.
+          // Also, one can't await a wildcard
+          || ((is_cancel_all() || is_query()) && !await && !timeout && !expiration && !if_version)
+        )
+    ;
   }
 
+  bool is_awaitable() const { return !(is_query() || is_cancel()); }
   bool is_mutating() const { return (control.roots_op != INCLUDE_OR_QUERY) || !roots.empty() || timeout || expiration; }
   bool is_cancel_all() const { return !set_id && is_cancel(); }
   bool excludes_roots() const { return control.roots_op == RESET_OR_CANCEL || (control.roots_op == EXCLUDE_OR_RELEASE && !roots.empty()); }

src/test/mds/TestQuiesceDb.cc

@@ -514,8 +514,10 @@ TEST_F(QuiesceDbTest, QuiesceRequestValidation)
           << bool(expiration) << ", await: " 
           << bool(await) << ", roots.size(): " << roots.size();
       } else {
-        // if set id is provided, all goes
-        if (set_id) {
+        if (!r.is_awaitable() && r.await) {
+          EXPECT_FALSE(r.is_valid());
+        } else if (set_id) {
+          // if set id is provided, all goes
           EXPECT_TRUE(r.is_valid())
             << "op: " << r.op_string() << ", set_id: " << bool(set_id) 
             << ", if_version: " << bool(if_version)