[meta] validate MetaIndexer metadata persistence & recovery (#98)

* [meta] introducing MetaDummyBackend for testing purpose

MetaDummyBackend is a meta storage backend implementation with in-memory
data store, and optional local filesystem persistence capability, whose
*sole* purpose is for testing & validating.

WARNING:
THIS IS A TEST FACILITY WITHOUT ANY FUNCTIONALITY COMPLETENESS WARRANTY;
DO NOT EVER USE IT UNDER ANY PRODUCTION CIRCUMSTANCE.

* [meta] [test] enable & update MetaIndexer persistence & recovery test

* [proto] add the missing persist_metadata_interval_time_ms field
  * MetaIndexer's metadata persistence interval can be adjusted via API
    calls

* [test] add reclaiming test cases
  * validate MetaIndexer metadata persistence & recovery behavior #77
  * validate MetaIndexer metadata persistence & recovery backward
    compatibility: v0 and v1 behavior validation #103
  * switch meta storage backend: local -> dummy

Author: oldsharp

integration_test/reclaimer/reclaiming_test.py

@@ -161,6 +161,294 @@ def test_reclaiming_01(self):
         # now the writing of key=16 should success
         self._write(16)
 
+    def test_persist_recover_00(self):
+        """Test e2e persist/recover: cache locations and metadata
+        survive a normal server restart.
+
+        The meta indexer is configured with the dummy backend with
+        filesystem persistence enabled so that
+
+        1. cache locations are always flushed to disk, and
+        2. metadata like key_count and storage usage accounting data are
+           flushed to disk *before* every metadata READ.
+
+        After a controlled server restart the instance group is
+        re-registered (which reinitialise the MetaIndexer from the
+        persisted file), and the test verifies:
+
+        1. All block meta written before the restart are still
+           addressable.
+        2. key_count was recovered (not reset to zero), so the capacity
+           limit is enforced correctly.
+        """
+        add_storage_req = {
+            "trace_id": self._trace_id,
+            "storage": self._make_dummy_storage(),
+        }
+        self._admin_client.add_storage(add_storage_req)
+
+        # add instance group
+        # start with the reclaiming trigger would not happen
+        ig = self._make_dummy_instance_group()
+        create_ig_req = {
+            "trace_id": self._trace_id,
+            "instance_group": ig,
+        }
+        self._admin_client.create_instance_group(create_ig_req)
+
+        # register instance
+        reg_ins_data_req = self._make_dummy_ins_req()
+        self._client.register_instance(reg_ins_data_req)
+
+        # write 8 blocks (half of max_key_count=16)
+        write_count = 8
+        for i in range(write_count):
+            self._write(i)
+
+        # --- restart ---
+        self.worker_manager.stop_worker(0)
+        self.assertTrue(self.worker_manager.start_worker(0))
+
+        # reconnect clients: update_ports() assigns fresh ports on every
+        # start
+        self._admin_client.close()
+        self._client.close()
+        self._admin_client, self._client = self._get_manager_client()
+
+        # the registry is in-memory only, so re-add the storage and
+        # instance group after restart
+        # crucially the same storage_uri is used so that
+        # MetaIndexer.Init -> RecoverMetaData will reload key_count and
+        # storage_usage_data from the persisted file
+        self._admin_client.add_storage(add_storage_req)
+        create_ig_req["trace_id"] = self._trace_id + "_restart"
+        self._admin_client.create_instance_group(create_ig_req)
+        self._client.register_instance(reg_ins_data_req)
+
+        # 1. verify that all blocks written before the restart are still
+        #    addressable (cache locations was persisted and recovered)
+        for i in range(write_count):
+            get_location_req = {
+                "trace_id": f"{self._trace_id}_verify_{i}",
+                "query_type": "QT_PREFIX_MATCH",
+                "block_keys": [i],
+                "instance_id": self._instance_id,
+                "block_mask": {"offset": 0},
+            }
+            resp = self._client.get_cache_location(get_location_req)
+            self.assertGreater(
+                len(resp["locations"]),
+                0,
+                f"block {i} should be accessible after restart",
+            )
+
+        # 2. verify key_count was recovered (not reset to zero)
+        #    write the remaining 8 blocks (keys 8-15) to reach
+        #    max_key_count=16
+        for i in range(write_count, 16):
+            self._write(i)
+        # key_count is now:
+        # 8 (recovered) + 8 (just written) = 16 = max_key_count
+        # so the next write must be rejected
+        self._start_write_expect_fail(16)
+
+    def test_persist_recover_01(self):
+        """Test e2e persist/recover: storage usage data survives a
+        normal server restart.
+
+        After writing all 16 blocks (filling key_count to
+        max_key_count=16), each block contributes 1024 bytes to
+        StorageUsageData for the NFS storage type.  StorageUsageData is
+        persisted to the dummy backend together with key_count.
+
+        After a controlled server restart the instance group is
+        re-registered, which reloads StorageUsageData from the persisted
+        file.  The test verifies recovery by observing the reclaimer's
+        byte-usage trigger:
+
+        * If StorageUsageData IS recovered (grp_used_byte_sz_ > 0):
+          the group byte-usage ratio is 0.5 which exceeds the 0.1
+          threshold → the reclaimer fires → some blocks are freed → a
+          new write succeeds.
+
+        * If StorageUsageData is NOT recovered (grp_used_byte_sz_ == 0):
+          cache_reclaimer.cc line 480 returns early with water-level
+          = false even though key_count was recovered; the reclaimer
+          does NOT trigger → the write still fails.
+        """
+        add_storage_req = {
+            "trace_id": self._trace_id,
+            "storage": self._make_dummy_storage(),
+        }
+        self._admin_client.add_storage(add_storage_req)
+
+        # add instance group
+        # start with the reclaiming trigger would not happen
+        ig = self._make_dummy_instance_group()
+        create_ig_req = {
+            "trace_id": self._trace_id,
+            "instance_group": ig,
+        }
+        self._admin_client.create_instance_group(create_ig_req)
+
+        # register instance
+        reg_ins_data_req = self._make_dummy_ins_req()
+        self._client.register_instance(reg_ins_data_req)
+
+        # write all 16 blocks to fill key_count to max_key_count=16
+        # each block also adds 1024 bytes to StorageUsageData (NFS type)
+        for i in range(16):
+            self._write(i)
+
+        # key_count is now at max; the next write must be rejected
+        self._start_write_expect_fail(16)
+
+        # --- restart ---
+        self.worker_manager.stop_worker(0)
+        self.assertTrue(self.worker_manager.start_worker(0))
+
+        # reconnect clients after restart
+        self._admin_client.close()
+        self._client.close()
+        self._admin_client, self._client = self._get_manager_client()
+
+        # re-register storage and instance group using the same storage_uri
+        # so that MetaIndexer.Init -> RecoverMetaData reloads both key_count
+        # and storage_usage_data from the persisted file
+        self._admin_client.add_storage(add_storage_req)
+        create_ig_req["trace_id"] = self._trace_id + "_restart"
+        self._admin_client.create_instance_group(create_ig_req)
+        self._client.register_instance(reg_ins_data_req)
+
+        # key_count is recovered to 16 = max, so write must still fail
+        self._start_write_expect_fail(16)
+
+        # lower the reclaim trigger so the reclaimer fires ONLY if
+        # storage_usage_data was recovered:
+        #   not recovered → grp_used_byte_sz_ == 0 → trigger checker
+        #                   early-return false → reclaimer does NOT run
+        #                   → write fails
+        #   recovered     → grp_used_byte_sz_ == 16 * 1024 = 16384 bytes
+        #                   group ratio = 16384 / 32768 = 0.5 > 0.1
+        #                   → reclaimer fires → blocks freed → write
+        #                   succeeds
+        curr_ver = ig["version"]
+        ig["version"] = curr_ver + 1
+        ig[
+            "cache_config"
+        ][
+            "reclaim_strategy"
+        ][
+            "trigger_strategy"
+        ][
+            "used_percentage"
+        ] = 0.1
+        update_ig_req = {
+            "trace_id": self._trace_id + "_update_ig",
+            "instance_group": ig,
+            "current_version": curr_ver,
+        }
+        self._admin_client.update_instance_group(update_ig_req)
+
+        # 2 seconds is enough for the background reclaimer to run
+        time.sleep(2)
+
+        # write should succeed only if reclaiming happened, which
+        # requires storage_usage_data to have survived the restart
+        self._write(16)
+
+    def test_persist_recover_02(self):
+        """Test metadata persistence-recovery backward compatibility
+        handling.
+
+        Scenario
+        --------
+        1. Setup: Fill test_instance_01 with 16 blocks but do not
+           trigger reclaiming.
+        2. Restart the server with intentionally crafted version 0
+           metadata file, that is, only key_count is included.
+        3. Re-register test_instance_01, lower the threshold for
+           test_group_01; make the reclaimer fires.
+        4. Assert: a new write to test_instance_01 should success, which
+           means the reclaiming worked as expected, thus the backward
+           compatibility is properly handled.
+        """
+        # add storage
+        add_storage_req = {
+            "trace_id": self._trace_id,
+            "storage": self._make_dummy_storage(),
+        }
+        self._admin_client.add_storage(add_storage_req)
+
+        # add instance group; reclaim trigger is intentionally too high
+        # to fire at this point
+        ig = self._make_dummy_instance_group()
+        create_ig_req = {
+            "trace_id": self._trace_id,
+            "instance_group": ig,
+        }
+        self._admin_client.create_instance_group(create_ig_req)
+
+        # register test_instance_01
+        reg_ins_01_req = self._make_dummy_ins_req()
+        self._client.register_instance(reg_ins_01_req)
+
+        # write 16 keys into test_instance_01
+        for i in range(16):
+            self._write(i)
+
+        # storage quota is full
+        self._start_write_expect_fail(16)
+
+        # restart the server
+        self.worker_manager.stop_worker(0)
+        meta_storage_backend_config = ig[
+            "cache_config"
+        ][
+            "meta_indexer_config"
+        ][
+            "meta_storage_backend_config"
+        ]
+        self._make_v0_persist_data(meta_storage_backend_config)
+        self.assertTrue(self.worker_manager.start_worker(0))
+
+        # reconnect clients: update_ports() assigns fresh ports on every
+        # start
+        self._admin_client.close()
+        self._client.close()
+        self._admin_client, self._client = self._get_manager_client()
+
+        # re-add the storage and instance group after restart
+        # re-register test_instance_01 to bring the recovered indexer
+        # back online
+        self._admin_client.add_storage(add_storage_req)
+        create_ig_req["trace_id"] = self._trace_id + "_restart"
+        self._admin_client.create_instance_group(create_ig_req)
+        self._client.register_instance(reg_ins_01_req)
+
+        # fire the reclaimer for test_group_01
+        curr_ver = ig["version"]
+        ig["version"] = curr_ver + 1
+        ig[
+            "cache_config"
+        ][
+            "reclaim_strategy"
+        ][
+            "trigger_strategy"
+        ][
+            "used_percentage"
+        ] = 0.1
+        self._admin_client.update_instance_group({
+            "trace_id": self._trace_id + "_update_ig",
+            "instance_group": ig,
+            "current_version": curr_ver,
+        })
+
+        time.sleep(2)
+
+        # this write can succeed only when v0 meta is properly handled
+        self._write(16)
+
     def _get_manager_client(self):
         self._admin_http_port = self.worker_manager.get_worker(
             0).env.admin_http_port
@@ -276,7 +564,7 @@ def _make_dummy_storage(self) -> Dict:
         return {
             "global_unique_name": self._storage_name,
             "nfs": {
-                "root_path": f"/tmp/{self._storage_name}",
+                "root_path": f"{self.get_workdir()}/{self._storage_name}/",
             }
         }
 
@@ -312,13 +600,14 @@ def _make_dummy_instance_group(self) -> Dict:
                     "max_key_count": 16,  # start with 16 max key
                     "mutex_shard_num": 16,
                     "meta_storage_backend_config": {
-                        "storage_type": "local",
-                        "storage_uri": "",  # disable persistence
+                        "storage_type": "dummy",
+                        "storage_uri": f"file://{self.get_workdir()}/meta_storage_{self._instance_group_name}",
                     },
                     "meta_cache_policy_config": {
                         "capacity": 1024 * 1024 * 1024,
                         "type": "LRU",
-                    }
+                    },
+                    "persist_metadata_interval_time_ms": 0,
                 }
             },
             "user_data": "user-defined info",
@@ -350,6 +639,34 @@ def _make_dummy_model_deployment(self):
             "pp_size": 1,
         }
 
+    def _make_v0_persist_data(self, meta_storage_backend_conf):
+        if meta_storage_backend_conf["storage_type"] == "dummy":
+            # rewrite the persisted metadata file
+            import json
+            d = {}
+
+            persist_path = meta_storage_backend_conf["storage_uri"]
+            persist_path = persist_path.removeprefix("file://")
+            persist_path = "_".join((persist_path, self._instance_id,))
+
+            with open(persist_path, 'r') as f:
+                meta_key = "__metadata__"
+                key = "__storage_usage_data__"
+                d = json.load(f)
+                d1 = json.loads(d[meta_key])
+                if key in d1:
+                    del d1[key]
+                d[meta_key] = json.dumps(d1)
+
+            with open(persist_path, 'w') as f:
+                json.dump(d, f)
+
+        elif meta_storage_backend_conf["storage_type"] == "local":
+            pass
+        elif meta_storage_backend_conf["storage_type"] == "redis":
+            # TODO
+            pass
+
 
 if __name__ == "__main__":
     unittest.main()

kv_cache_manager/config/meta_indexer_config.h

@@ -34,6 +34,7 @@ class MetaIndexerConfig : public Jsonizable {
         : max_key_count_(max_key_count)
         , mutex_shard_num_(mutex_shard_num)
         , batch_key_size_(batch_key_size)
+        , persist_metadata_interval_time_ms_(kDefaultPersistMetaDataIntervalTimeMs)
         , meta_storage_backend_config_(meta_storage_backend_config)
         , meta_cache_policy_config_(meta_cache_policy_config) {}
 
@@ -83,6 +84,7 @@ class MetaIndexerConfig : public Jsonizable {
         Put(writer, "max_key_count", max_key_count_);
         Put(writer, "mutex_shard_num", mutex_shard_num_);
         Put(writer, "batch_key_size", batch_key_size_);
+        Put(writer, "persist_metadata_interval_time_ms", persist_metadata_interval_time_ms_);
         Put(writer, "meta_storage_backend_config", meta_storage_backend_config_);
         Put(writer, "meta_cache_policy_config", meta_cache_policy_config_);
     }