[manager] exclude EventReport from cache reclamation (#286)

Author: shaohuaxi

kv_cache_manager/manager/cache_reclaimer.cc

@@ -779,6 +779,10 @@ CacheReclaimer::GetWaterLevelExceed(const RequestContext *request_context,
             // skip vcns_hf3fs, because it is treated as hf3fs
             continue;
         }
+        if (IsEventReportStorageType(type)) {
+            // EventReport metadata is not reclaimed by CacheReclaimer.
+            continue;
+        }
 
         const std::size_t type_idx = ToIndex(ToBaseType(type));
         const std::uint64_t type_credit = type_idx < credited_bytes.size() ? credited_bytes[type_idx] : 0;
@@ -1418,7 +1422,8 @@ bool CacheReclaimer::FilterLocID(RequestContext *request_context,
             const bool covered = std::any_of(loc_map.begin(), loc_map.end(), [&is_pending_location, &loc_on_cold_storage, block_key, &source_spec](const auto &entry) {
                 const auto &loc_ptr = entry.second;
                 if (!loc_ptr || loc_ptr->status() != CacheLocationStatus::CLS_SERVING ||
-                    is_pending_location(block_key, loc_ptr->id()) || !loc_on_cold_storage(*loc_ptr)) {
+                    IsEventReportStorageType(loc_ptr->type()) || is_pending_location(block_key, loc_ptr->id()) ||
+                    !loc_on_cold_storage(*loc_ptr)) {
                     return false;
                 }
                 return std::any_of(loc_ptr->location_specs().begin(),
@@ -1454,6 +1459,9 @@ bool CacheReclaimer::FilterLocID(RequestContext *request_context,
                     continue;
                 }
                 const auto &loc = *loc_ptr;
+                if (IsEventReportStorageType(loc.type())) {
+                    continue;
+                }
                 if (is_pending_location(block_key, loc.id())) {
                     continue;
                 }
@@ -1482,6 +1490,12 @@ bool CacheReclaimer::FilterLocID(RequestContext *request_context,
             }
             ++valid_location_count;
             const auto &loc = *loc_ptr;
+            if (IsEventReportStorageType(loc.type())) {
+                // Generic reclamation must not remove metadata owned by a
+                // ReportEvent reporter. Dedicated lifecycle/snapshot cleanup
+                // uses metadata-only conditional deletion instead.
+                continue;
+            }
             if (is_pending_location(block_key, loc.id())) {
                 METRICS_(cache_reclaimer, duplicate_pending_location_filtered_count) += 1;
                 continue;
@@ -1987,7 +2001,7 @@ std::shared_ptr<CacheReclaimer::GroupUsageData> CacheReclaimer::GetGroupUsageDat
 
         for (std::size_t idx = 1; idx < static_cast<std::size_t>(DataStorageType::COUNT); ++idx) {
             const auto type = static_cast<DataStorageType>(idx);
-            if (type == DataStorageType::DATA_STORAGE_TYPE_VCNS_HF3FS) {
+            if (type == DataStorageType::DATA_STORAGE_TYPE_VCNS_HF3FS || IsEventReportStorageType(type)) {
                 continue;
             }
             data->AddGroupUsageByType(type, meta_indexer->GetStorageUsageByType(type));

kv_cache_manager/manager/test/cache_reclaimer_test.cc

@@ -1699,6 +1699,34 @@ TEST_F(CacheReclaimerTest, TestTriggerReclaiming17) {
     }
 }
 
+TEST_F(CacheReclaimerTest, TestEventReportUsageDoesNotTriggerStorageTypeWaterLevel) {
+    dummy_meta_indexer->SetStorageUsageByType(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5, 90);
+    dummy_meta_indexer->SetStorageUsageByType(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2, 90);
+
+    const auto instance_group = InstanceGroupFactory();
+    instance_group->quota_.set_capacity(100);
+    instance_group->quota_.set_quota_config({
+        QuotaConfig(100, DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5),
+        QuotaConfig(100, DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2),
+    });
+    instance_group->cache_config_->reclaim_strategy_->trigger_strategy_.set_used_percentage(0.8);
+    key_count = 0;
+    max_key_count = 100;
+
+    cache_reclaimer_->job_state_flag_ = true;
+    const auto water_level = cache_reclaimer_->GetWaterLevelExceed(
+        request_context_.get(),
+        instance_group->name(),
+        instance_group->quota(),
+        instance_group->cache_config()->reclaim_strategy(),
+        instance_infos);
+    ASSERT_NE(nullptr, water_level);
+    EXPECT_TRUE(water_level->GetGeneralWaterLevelExceed());
+    EXPECT_FALSE(
+        water_level->GetWaterLevelExceedByType(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5));
+    EXPECT_FALSE(water_level->GetWaterLevelExceedByType(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2));
+}
+
 TEST_F(CacheReclaimerTest, TestInsufficientSampledKeys) {
     sample_reclaim_keys = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
     get_out_properties = {
@@ -3162,6 +3190,65 @@ TEST_F(CacheReclaimerTest, TestFilterLocationCreditsNormalizeTypeAndPredictKeysC
     EXPECT_EQ(1, predicted_keys);
 }
 
+TEST_F(CacheReclaimerTest, TestFilterLocationExcludesEventReportStorage) {
+    const auto instance = InstanceInfoFactory();
+    batch_get_loc_out_maps = {
+        CacheLocationMap{
+            {"event_l1p5",
+             MakeCacheLocation("event_l1p5",
+                               CacheLocationStatus::CLS_SERVING,
+                               DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5,
+                               "nfs://reporter/l1p5?size=10")},
+            {"event_l2",
+             MakeCacheLocation("event_l2",
+                               CacheLocationStatus::CLS_SERVING,
+                               DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2,
+                               "nfs://reporter/l2?size=20")},
+        },
+        CacheLocationMap{
+            {"event_l2",
+             MakeCacheLocation("event_l2",
+                               CacheLocationStatus::CLS_SERVING,
+                               DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2,
+                               "nfs://reporter/l2?size=20")},
+            {"nfs",
+             MakeCacheLocation("nfs",
+                               CacheLocationStatus::CLS_SERVING,
+                               DataStorageType::DATA_STORAGE_TYPE_NFS,
+                               "nfs://store/nfs?size=30")},
+        },
+    };
+
+    CacheReclaimer::WaterLevelExceed water_level;
+    water_level.SetGeneralWaterLevelExceed(true);
+    std::vector<std::vector<std::string>> location_ids;
+    CacheReclaimer::BytesByStorageType bytes_by_type{};
+    CacheReclaimer::CountsByStorageType counts_by_type{};
+    std::uint64_t predicted_keys = 0;
+    CacheReclaimer::AgeStats create_age_stats;
+    ASSERT_TRUE(cache_reclaimer_->FilterLocID(request_context_.get(),
+                                              instance,
+                                              {10, 11},
+                                              water_level,
+                                              location_ids,
+                                              bytes_by_type,
+                                              counts_by_type,
+                                              predicted_keys,
+                                              create_age_stats));
+
+    ASSERT_EQ(2, location_ids.size());
+    EXPECT_TRUE(location_ids[0].empty());
+    ASSERT_EQ(1, location_ids[1].size());
+    EXPECT_EQ("nfs", location_ids[1][0]);
+    EXPECT_EQ(0, bytes_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5)]);
+    EXPECT_EQ(0, bytes_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2)]);
+    EXPECT_EQ(0, counts_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L1P5)]);
+    EXPECT_EQ(0, counts_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_EVENT_REPORT_L2)]);
+    EXPECT_EQ(30, bytes_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_NFS)]);
+    EXPECT_EQ(1, counts_by_type[ToIndex(DataStorageType::DATA_STORAGE_TYPE_NFS)]);
+    EXPECT_EQ(0, predicted_keys);
+}
+
 TEST_F(CacheReclaimerTest, TestCreditDeadlineDisablesCreditButKeepsPendingAndHardQuota) {
     CacheReclaimerAsyncDeleteConfig config;
     config.inflight_delete_timeout_ms = 1;