refactor(metactl): when restore without --initial-cluster, add default single node cluster config (#17969)

* refactor(metactl): when bendsave restores, add default single node cluster config

When using `bendsave` to restore a cluster, it should fill in a default
databend-meta cluster config which has only one nodes. So that it
restores a working one node cluster at once.

Other changes: rewrite meta new-cluster test by python.

* chore: add test to ensure restore a single node cluster from 3 nodes cluster data

Author: drmingdrmer

.github/actions/test_metactl/action.yml

@@ -15,4 +15,5 @@ runs:
     - name: Test metactl restore new cluster
       shell: bash
       run: |
-        bash ./tests/metactl/test-metactl-restore-new-cluster.sh
+        python -m pip install -r ./tests/metactl/requirements.txt
+        python ./tests/metactl/test-metactl-restore-new-cluster.py

Makefile

@@ -96,7 +96,7 @@ stateless-cluster-test-tls: build
 
 metactl-test:
 	bash ./tests/metactl/test-metactl.sh
-	bash ./tests/metactl/test-metactl-restore-new-cluster.sh
+	python ./tests/metactl/test-metactl-restore-new-cluster.py
 
 meta-kvapi-test:
 	bash ./tests/meta-kvapi/test-meta-kvapi.sh

src/bendsave/src/restore.rs

@@ -58,11 +58,16 @@ pub async fn restore_meta(efs: Operator, meta_cfg: &databend_meta::configs::Conf
     local_file.sync_all().await?;
     local_file.shutdown().await?;
 
+    let id = meta_cfg.raft_config.id;
+
     let import_args = ImportArgs {
         raft_dir: Some(meta_cfg.raft_config.raft_dir.to_string()),
         db: DATABEND_META_BACKUP_PATH.to_string(),
-        initial_cluster: vec![],
-        id: meta_cfg.raft_config.id,
+
+        // when restoring, create a single node cluster, so that it will serve at once.
+        // And the address in the following will be replaced with the content in the config upon `databend-meta` startup.
+        initial_cluster: vec![format!("{id}=127.0.0.1:28004")],
+        id,
     };
     import_data(&import_args).await?;
 

src/meta/control/src/args.rs

@@ -148,7 +148,24 @@ pub struct ImportArgs {
     #[clap(long, default_value = "")]
     pub db: String,
 
-    /// initial_cluster format: node_id=endpoint,grpc_api_addr
+    /// initial_cluster format: <node_id>=<raft_api_host>:<raft_api_port>
+    ///
+    /// For example, the following command restores the node(id=1) of a cluster of two:
+    /// ```text
+    /// $0 \
+    /// --raft-dir ./meta_dir \
+    /// --db meta.db \
+    /// --id=1 \
+    /// --initial-cluster 1=localhost:29103 \
+    /// --initial-cluster 1=localhost:29103
+    /// ```
+    ///
+    /// If it is empty, the cluster information in the imported data is kept.
+    /// For example:
+    /// ```text
+    /// ["state_machine/0",{"StateMachineMeta":{"key":"LastMembership","value":{"Membership":{"log_id":null,"membership":{"configs":[[4]],"nodes":{"4":{}}}}}}}]
+    /// ["state_machine/0",{"Nodes":{"key":4,"value":{"name":"4","endpoint":{"addr":"127.0.0.1","port":28004},"grpc_api_advertise_address":null}}}]
+    /// ```
     #[clap(long)]
     pub initial_cluster: Vec<String>,
 

src/meta/control/src/import.rs

@@ -284,6 +284,8 @@ fn build_nodes(initial_cluster: Vec<String>, id: u64) -> anyhow::Result<BTreeMap
         return Ok(nodes);
     }
 
+    eprintln!("new cluster: {:?}", nodes);
+
     if !nodes.contains_key(&id) {
         return Err(anyhow::anyhow!(
             "node id ({}) has to be one of cluster member({:?})",

tests/metactl/requirements.txt

@@ -0,0 +1 @@
+requests

tests/metactl/test-metactl-restore-new-cluster.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import shutil
+import subprocess
+import time
+import sys
+import re
+import signal
+from typing import Dict
+import requests
+from pathlib import Path
+import socket
+
+# Configuration
+BUILD_PROFILE = os.environ.get("BUILD_PROFILE", "debug")
+SCRIPT_PATH = Path(__file__).parent.absolute()
+metactl_bin = f"./target/{BUILD_PROFILE}/databend-metactl"
+
+
+# Helper functions
+def run_command(cmd, check=True, shell=False):
+    """Run a command and return its output"""
+    if isinstance(cmd, str) and not shell:
+        cmd = cmd.split()
+
+    print(f"Running: {cmd}")
+    result = subprocess.run(cmd,
+                            check=check, shell=shell,
+                            text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return result.stdout
+
+
+def wait_for_port(port, timeout=10):
+    now = time.time()
+
+    while time.time() - now < timeout:
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.connect(("0.0.0.0", port))
+                print("OK :{} is listening".format(port))
+                sys.stdout.flush()
+                return
+        except Exception:
+            print("... connecting to :{}".format(port))
+            sys.stdout.flush()
+            time.sleep(1)
+
+    raise Exception("fail to connect to :{}".format(port))
+
+
+def kill_databend_meta():
+    print_step("Kill databend-meta processes")
+    try:
+        run_command("killall databend-meta", check=False)
+        time.sleep(2)
+    except subprocess.CalledProcessError:
+        pass  # It's okay if there are no processes to kill
+
+
+def start_meta_node(node_id, is_new: bool):
+    meta_bin = f"./target/{BUILD_PROFILE}/databend-meta"
+
+    ports = {
+        1: (9191, 19191),
+        2: (28202, 29191),
+        3: (28302, 39191),
+    }
+
+    if is_new:
+        config_fn = f"new-databend-meta-node-{node_id}.toml"
+        port = ports[node_id][1]
+    else:
+        config_fn = f"databend-meta-node-{node_id}.toml"
+        port = ports[node_id][0]
+
+    config_file = f"./tests/metactl/config/{config_fn}"
+
+    subprocess.Popen([meta_bin, "--config-file", config_file],
+                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+    wait_for_port(port)
+    time.sleep(1)
+
+
+# return the stdout of metactl export
+def metactl_export(meta_dir: str, output_path: str) -> str:
+    print_step(f"Start: Export meta data from {meta_dir} to {output_path}")
+
+    cmd = [metactl_bin, "export", "--raft-dir", meta_dir]
+
+    if output_path:
+        cmd.append("--db")
+        cmd.append(output_path)
+
+    result = subprocess.Popen(cmd,
+                              stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    result.wait()
+
+    print_step(f"Done: Exported meta data to {output_path}")
+    return result.stdout.read().decode()
+
+
+def metactl_import(meta_dir: str, id: int, db_path: str, initial_cluster: Dict[int, str]):
+    cmd = [metactl_bin, "import", "--raft-dir", meta_dir,
+           "--id", str(id),
+           "--db", db_path, ]
+
+    for id, addr in initial_cluster.items():
+        cmd.append("--initial-cluster")
+        cmd.append(f"{id}={addr}")
+
+    result = subprocess.Popen(cmd,
+                              stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    result.wait()
+    print(result.stdout.read().decode())
+    print(result.stderr.read().decode())
+
+
+def cluster_status(msg: str) -> str:
+    print_step(f"Check /v1/cluster/status {msg} start")
+
+    response = requests.get("http://127.0.0.1:28101/v1/cluster/status")
+    status = response.json()
+
+    print_step(f"Check /v1/cluster/status {msg} end")
+
+    return status
+
+
+def print_title(title):
+    print()
+    print(" ===")
+    print(f" === {title}")
+    print(" ===")
+
+
+def print_step(step: str):
+    print(f" === {step}")
+
+
+def main():
+    kill_databend_meta()
+
+    shutil.rmtree(".databend", ignore_errors=True)
+
+    print_title("1. Start 3 meta node cluster to generate raft logs")
+
+    start_meta_node(1, False)
+    start_meta_node(2, False)
+    start_meta_node(3, False)
+
+    old_status = cluster_status("old cluster")
+    print(json.dumps(old_status, indent=2))
+
+    kill_databend_meta()
+
+    print_title("2. Export meta node data")
+
+    metactl_export("./.databend/meta1", "meta.db")
+
+    exported = metactl_export("./.databend/meta1", None)
+    print_step("Exported meta data. start")
+    print(exported)
+    print_step("Exported meta data. end")
+
+    print_step("Clear meta service data dirs")
+    shutil.rmtree(".databend")
+
+    print_title("3. Import old meta node data to new cluster")
+
+    cluster = {
+        4: "localhost:29103",
+        5: "localhost:29203",
+        6: "localhost:29303",
+    }
+
+    metactl_import("./.databend/new_meta1", 4, "meta.db", cluster)
+    metactl_import("./.databend/new_meta2", 5, "meta.db", cluster)
+    metactl_import("./.databend/new_meta3", 6, "meta.db", cluster)
+
+    print_step("3.1 Check if state machine is complete by checking key 'LastMembership'")
+
+    meta1_data = metactl_export("./.databend/new_meta1", None)
+    print(meta1_data)
+
+    meta2_data = metactl_export("./.databend/new_meta2", None)
+    print(meta2_data)
+
+    meta3_data = metactl_export("./.databend/new_meta3", None)
+    print(meta3_data)
+
+    assert "LastMembership" in meta1_data
+    assert "LastMembership" in meta2_data
+    assert "LastMembership" in meta3_data
+
+    print_title("3.2 Start 3 new meta node cluster")
+
+    start_meta_node(1, True)
+    start_meta_node(2, True)
+    start_meta_node(3, True)
+
+    print_step("sleep 5 sec to wait for membership to commit")
+    time.sleep(5)
+
+    print_title("3.3 Check membership in new cluster")
+
+    new_status = cluster_status("new cluster")
+    print(json.dumps(new_status, indent=2))
+
+    voters = new_status["voters"]
+
+    names = dict([(voter["name"], voter) for voter in voters])
+
+    assert names["4"]["endpoint"] == {"addr": "localhost", "port": 29103}
+    assert names["5"]["endpoint"] == {"addr": "localhost", "port": 29203}
+    assert names["6"]["endpoint"] == {"addr": "localhost", "port": 29303}
+
+    kill_databend_meta()
+    shutil.rmtree(".databend")
+
+
+    print_title("4. Import with --initial-cluster of one node")
+
+    cluster = {
+        # This will be replaced when meta node starts
+        4: "127.0.0.1:12345",
+    }
+
+    metactl_import("./.databend/new_meta1", 4, "meta.db", cluster)
+
+    print_step("4.1 Check if state machine is complete by checking key 'LastMembership'")
+    meta1_data = metactl_export("./.databend/new_meta1", None)
+    print(meta1_data)
+    assert "LastMembership" in meta1_data
+
+    print_title("4.2 Start new meta node cluster")
+
+    start_meta_node(1, True)
+
+    print_step("sleep 3 sec to wait for membership to commit")
+    time.sleep(3)
+
+    print_title("4.3 Check membership in new cluster")
+
+    new_status = cluster_status("new single node cluster")
+    print(json.dumps(new_status, indent=2))
+
+    voters = new_status["voters"]
+
+    names = dict([(voter["name"], voter) for voter in voters])
+
+    # The address is replaced with the content in config.
+    assert names["4"]["endpoint"] == {"addr": "localhost", "port": 29103}
+
+    kill_databend_meta()
+    shutil.rmtree(".databend")
+
+if __name__ == "__main__":
+    main()