Add cluster test cases (#2166)

### What problem does this PR solve?
Add cluster test cases
### Type of change
- [x] Test cases

---------

Co-authored-by: shenyushi <[email protected]>

conf/learner2.toml

@@ -0,0 +1,69 @@
+[general]
+version                  = "0.5.0"
+time_zone                = "utc-8"
+server_mode              = "cluster"
+
+[network]
+server_address           = "0.0.0.0"
+postgres_port            = 5436
+http_port                = 23825
+client_port              = 23824
+connection_pool_size     = 128
+peer_ip                  = "0.0.0.0"
+peer_port                = 23854
+
+[log]
+log_filename             = "infinity.log"
+log_dir                  = "/var/infinity/learner2/log"
+log_to_stdout            = true
+log_file_max_size        = "10GB"
+log_file_rotate_count    = 10
+
+# trace/debug/info/warning/error/critical 6 log levels, default: info
+log_level               = "debug"
+
+[storage]
+persistence_dir         = "/var/infinity/learner2/persistence"
+data_dir                = "/var/infinity/learner2/data"
+# periodically activates garbage collection:
+# 0 means real-time,
+# s means seconds, for example "60s", 60 seconds
+# m means minutes, for example "60m", 60 minutes
+# h means hours, for example "1h", 1 hour
+optimize_interval        = "10s"
+cleanup_interval         = "60s"
+compact_interval         = "120s"
+
+# dump memory index entry when it reachs the capacity
+mem_index_capacity       = 1048576
+
+storage_type             = "minio"
+
+[storage.object_storage]
+url                      = "127.0.0.1:9000"
+bucket_name              = "infinity"
+access_key               = "minioadmin"
+secret_key               = "minioadmin"
+enable_https             = false
+
+[buffer]
+buffer_manager_size      = "4GB"
+lru_num                  = 7
+temp_dir                 = "/var/infinity/learner2/tmp"
+
+memindex_memory_quota   = "1GB"
+
+[wal]
+wal_dir                       = "/var/infinity/learner2/wal"
+full_checkpoint_interval      = "86400s"
+delta_checkpoint_interval     = "60s"
+# delta_checkpoint_threshold  = 1000000000
+wal_compact_threshold         = "1GB"
+
+# flush_at_once: write and flush log each commit
+# only_write: write log, OS control when to flush the log, default
+# flush_per_second: logs are written after each commit and flushed to disk per second.
+wal_flush                     = "only_write"
+
+[resource]
+resource_dir                  = "/var/infinity/learner2/resource"

python/infinity_http.py

@@ -146,6 +146,13 @@ def set_role_follower(self, node_name, leader_addr):
         self.net.raise_exception(r)
         return database_result()
 
+    def set_role_learner(self, node_name, leader_addr):
+        url = f"admin/node/current"
+        h = self.net.set_up_header(["accept", "content-type"])
+        d = self.net.set_up_data([], {"role": "learner", "name": node_name, "address": leader_addr})
+        r = self.net.request(url, "post", h, d)
+        self.net.raise_exception(r)
+        return database_result()
 
     # database
     def create_database(self, db_name, opt=ConflictType.Error):
@@ -432,9 +439,9 @@ def list_indexes(self):
         self.net.raise_exception(r)
         r_json = r.json()
         index_list = []
-        exists = r_json.get("tables", None)
+        exists = r_json.get("indexes", None)
         if exists is not None:
-            for t in r_json["tables"]:
+            for t in r_json["indexes"]:
                 index_list.append(t)
         return database_result(index_list=index_list)
 

python/test_cluster/common_values.py

@@ -0,0 +1,2 @@
+TEST_DATA_DIR = "test/data/"
+TEST_TMP_DIR = "/var/infinity/test_data/"

python/test_cluster/conftest.py

@@ -21,21 +21,31 @@ def pytest_addoption(parser):
         action="store",
         default=9001,
     )
-    parser.addoption("--infinity_dir", action="store", required=True)
+    parser.addoption(
+        "--infinity_dir",
+        action="store",
+        required=True,
+        help="Path to infinity directory. For local test, $pwd is ok",
+    )
     parser.addoption("--docker", action="store_true", default=False)
 
+
 def pytest_configure(config):
-    config.addinivalue_line("markers", "docker: mark test to run only when --docker option is provided")
+    config.addinivalue_line(
+        "markers", "docker: mark test to run only when --docker option is provided"
+    )
+
 
 def pytest_collection_modifyitems(config, items):
     if config.getoption("--docker"):
-        return # do not skip docker test
+        return  # do not skip docker test
     skip_docker = pytest.mark.skip(reason="need --docker option to run")
     for item in items:
         if "docker" in item.keywords:
             print(f"skip {item.name}")
             item.add_marker(skip_docker)
 
+
 def pytest_generate_tests(metafunc):
     infinity_path = metafunc.config.getoption("infinity_path")
     minio_dir = metafunc.config.getoption("minio_dir")
@@ -51,7 +61,10 @@ def pytest_generate_tests(metafunc):
 
     if "docker_cluster" in metafunc.fixturenames:
         # skip if docker is in option and the testcase is marked with docker
-        if not metafunc.config.getoption("--docker") and "docker" in metafunc.definition.keywords:
+        if (
+            not metafunc.config.getoption("--docker")
+            and "docker" in metafunc.definition.keywords
+        ):
             return
 
         print("Init DockerInfinityCluster")

python/test_cluster/docker_infinity_cluster.py

@@ -100,6 +100,7 @@ def uninit(self):
             [
                 "cd /infinity",
                 f"pid=$(pgrep -f infinity || true)",
+                f"echo $pid",
                 f"bash scripts/timeout_kill.sh {timeout} $pid",
             ]
         )
@@ -228,7 +229,6 @@ def add_node(self, node_name: str, config_path: str):
 
         try:
             container = docker_client.containers.get(container_name)
-            added = False
         except docker.errors.NotFound:
             container = docker_client.containers.run(
                 image=self.image_name,
@@ -241,10 +241,11 @@ def add_node(self, node_name: str, config_path: str):
                 ],
                 environment=[f"TZ={tz}"],
             )
-            added = True
 
-        if added:
+        try:
             self.network.connect(container)
+        except docker.errors.APIError as e:
+            pass
         info = docker_client.api.inspect_network(self.network.id)
         # print(info)
         mock_ip = info["Containers"][container.id]["IPv4Address"]
@@ -293,54 +294,4 @@ def __init_docker_params(self):
 
 
 if __name__ == "__main__":
-    infinity_path = "cmake-build-debug/src/infinity"
-    minio_dir = "minio"
-    minio_port = 9001
-    cluster = DockerInfinityCluster(
-        infinity_path, minio_params=MinioParams(minio_dir, minio_port)
-    )
-    try:
-        cluster.add_node("node1", "conf/leader.toml")
-        cluster.add_node("node2", "conf/follower.toml")
-
-        print("init nodes")
-
-        cluster.init_leader("node1")
-        cluster.init_follower("node2")
-
-        time.sleep(1)
-        print("insert in node1")
-
-        infinity1 = cluster.client("node1")
-        r = infinity1.list_databases()
-
-        db1 = infinity1.get_database("default_db")
-        table1 = db1.create_table(
-            "table1", {"c1": {"type": "int"}, "c2": {"type": "vector,4,float"}}
-        )
-        table1.insert([{"c1": 1, "c2": [1.0, 2.0, 3.0, 4.0]}])
-
-        res_gt = pd.DataFrame(
-            {
-                "c1": (1),
-                "c2": ([[1.0, 2.0, 3.0, 4.0]]),
-            }
-        ).astype({"c1": dtype("int32"), "c2": dtype("object")})
-
-        res = table1.output(["*"]).to_df()
-        pd.testing.assert_frame_equal(res, res_gt)
-
-        time.sleep(1)
-        print("select in node2")
-
-        infinity2 = cluster.client("node2")
-        db2 = infinity2.get_database("default_db")
-        table2 = db2.get_table("table1")
-        res = table2.output(["*"]).to_df()
-        pd.testing.assert_frame_equal(res, res_gt)
-    except Exception as e:
-        print(e)
-        cluster.clear()
-        raise
-    else:
-        cluster.clear()
+    pass