feat: meta-service transfer_leader: add response and default value (#…

…16201)

- The query `to=<next_leader_id>` can be absent in the transfer
  leadership request, such as:
  ```
  curl -vqs '<admin_api_address>/v1/ctrl/trigger_transfer_leader'
  ```

  In this case, the next node in the cluster will be used as the default
  value for query `to`.

- Add 200 OK response indicating an accepted tranfer leadership
  request. The response is a json contains current node id `from`, the
  node id to transfer leadership to: `to`, and all of the voter ids in
  the cluster `voter_ids`, such as:

  ```
  curl -vqs '127.0.0.1:28101/v1/ctrl/trigger_transfer_leader'
  < HTTP/1.1 200 OK
  < content-type: application/json; charset=utf-8
  < content-length: 37
  {"from":1,"to":2,"voter_ids":[1,2,3]}
  ```

- Add 404 NOT_FOUND response indicating an rejected transfer leadership
  request if the node is not the Leader. In such case it returns a
  string describing the state:

  ```
  curl -vqs '127.0.0.1:28301/v1/ctrl/trigger_transfer_leader'
  < HTTP/1.1 404 Not Found
  < content-length: 85
  This node is not leader, can not transfer leadership;
  id=1
  current_leader=Some(2)
  voter_ids=[1, 2, 3]
  ```

src/meta/service/src/api/http/v1/ctrl.rs

@@ -15,7 +15,11 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use databend_common_meta_sled_store::openraft::async_runtime::watch::WatchReceiver;
+use databend_common_meta_types::NodeId;
 use http::StatusCode;
+use log::info;
+use log::warn;
 use poem::web::Data;
 use poem::web::IntoResponse;
 use poem::web::Json;
@@ -40,27 +44,76 @@ pub struct TransferLeaderQuery {
     pub(crate) to: Option<u64>,
 }
 
-/// Transfer this Leader to another specified node.
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct TransferLeaderResponse {
+    pub from: NodeId,
+    pub to: NodeId,
+    pub voter_ids: Vec<NodeId>,
+}
+
+/// Transfer this Leader to another specified node, or next node in the cluster if not specified.
 ///
-/// If this node is not a Leader this request will be just ignored.
+/// If this node is not a Leader, 404 NOT_FOUND will be returned.
+/// Note that a 200 OK response does not mean the transfer is successful, it only means the request is accepted.
 #[poem::handler]
 pub async fn trigger_transfer_leader(
     meta_node: Data<&Arc<MetaNode>>,
     query: Option<Query<TransferLeaderQuery>>,
 ) -> poem::Result<impl IntoResponse> {
-    let Some(query) = query else {
-        return Err(poem::Error::from_string(
-            "missing query to=<node_id_to_transfer_leader_to>",
-            StatusCode::BAD_REQUEST,
-        ));
+    let metrics = meta_node.raft.metrics().borrow_watched().clone();
+
+    let id = metrics.id;
+    let current_leader = metrics.current_leader;
+    let voter_ids = metrics
+        .membership_config
+        .membership()
+        .voter_ids()
+        .collect::<Vec<_>>();
+
+    info!(
+        "id={} Received trigger_transfer_leader request: {:?}, \
+        this node: current_leader={:?} voter_ids={:?}",
+        id, &query, current_leader, voter_ids
+    );
+
+    let to = {
+        if let Some(query) = query {
+            query.to
+        } else {
+            None
+        }
     };
 
-    let Some(to) = query.to else {
+    let to = if let Some(to) = to {
+        to
+    } else {
+        // If `to` node is not specified, find the next node id in the voter list.
+
+        // There is still chance this Leader is not a voter,
+        // e.g., when Leader commit a membership without it.
+        let index = voter_ids.iter().position(|&x| x == id).unwrap_or_default();
+        voter_ids[(index + 1) % voter_ids.len()]
+    };
+
+    if current_leader != Some(id) {
+        warn!(
+            "id={} This node is not leader, can not transfer leadership; Current leader is: {:?} voter_ids={:?}",
+            id, current_leader, voter_ids,
+        );
+
         return Err(poem::Error::from_string(
-            "missing query to=<node_id_to_transfer_leader_to>",
-            StatusCode::BAD_REQUEST,
+            format!(
+                "This node is not leader, can not transfer leadership;\n\
+                 id={}\n\
+                 current_leader={:?}\n\
+                 voter_ids={:?}",
+                id, current_leader, voter_ids,
+            ),
+            StatusCode::NOT_FOUND,
         ));
-    };
+    }
+
+    info!("id={} Begin to transfer leadership to node: {}", id, to);
 
     meta_node
         .raft
@@ -69,7 +122,11 @@ pub async fn trigger_transfer_leader(
         .await
         .map_err(|e| poem::Error::from_string(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR))?;
 
-    Ok(Json(()))
+    Ok(Json(TransferLeaderResponse {
+        from: id,
+        to,
+        voter_ids,
+    }))
 }
 
 #[poem::handler]