diff --git a/docs/api/meta_service.md b/docs/api/meta_service.md index e2a8fcd3b..fd99abfe8 100644 --- a/docs/api/meta_service.md +++ b/docs/api/meta_service.md @@ -227,3 +227,82 @@ curl -g -vvv -X POST http://localhost:6382/api/getCacheMeta \ "detail_level": 1 }' ``` + +## Get Cache Meta Detail +```bash +curl -g -vvv -X POST http://localhost:6382/api/getCacheMetaDetail \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{ + "trace_id": "trace_id_131", + "instance_id": "test_instance", + "block_keys": [123, 456], + "block_mask": { + "offset": 0 + }, + "detail_level": 1 +}' +``` + +Example response: +```json +{ + "header": { + "status": { + "code": "OK", + "message": "Cache metadata detail retrieved successfully" + }, + "request_id": "request_id" + }, + "items": [ + { + "status": { + "code": "OK" + }, + "request_index": 0, + "block_key": 123, + "prev_block_key": "", + "properties": { + "BP#prev_key": "" + }, + "locations": [ + { + "location_id": "loc_a", + "status": "CLS_SERVING", + "type": "ST_3FS", + "spec_size": 2, + "create_time": 1710000000000000, + "location_specs": [ + {"name": "tp0", "uri": "3fs://cluster/root/key_123_tp0?offset=0&size=1024"}, + {"name": "tp1", "uri": "3fs://cluster/root/key_123_tp1?offset=0&size=1024"} + ] + }, + { + "location_id": "loc_b", + "status": "CLS_WRITING", + "type": "ST_NFS", + "spec_size": 1, + "create_time": 1710000001000000, + "location_specs": [ + {"name": "tp0", "uri": "file://nfs/root/key_123_tp0?offset=0&size=1024"} + ] + } + ] + }, + { + "status": { + "code": "OK" + }, + "request_index": 1, + "block_key": 456, + "locations": [ + { + "status": "CLS_NOT_FOUND" + } + ] + } + ] +} +``` + +This diagnostic API returns raw metadata for every unmasked requested key. It does not apply location selection, data-file existence filtering, or lazy prune. `prev_block_key` is a convenience projection of `properties["BP#prev_key"]`; the raw property is intentionally retained. A missing key is reported as a successful item with `CLS_NOT_FOUND`. Backend or deserialization failures are reported in that item's `status` while other successful items are still returned. If every item fails to read, the request returns an overall error. diff --git a/integration_test/meta_service/grpc_interface_test.py b/integration_test/meta_service/grpc_interface_test.py index 8f7df994b..3111ef02d 100644 --- a/integration_test/meta_service/grpc_interface_test.py +++ b/integration_test/meta_service/grpc_interface_test.py @@ -7,6 +7,7 @@ RegisterInstanceRequest, GetInstanceInfoRequest, GetCacheLocationRequest, + GetCacheMetaDetailRequest, StartWriteCacheRequest, FinishWriteCacheRequest, RemoveCacheRequest, @@ -76,6 +77,18 @@ def get_cache_location(self, data, check_response=True): f"Request to get_cache_location failed with error: {response_dict['header']['status']['message']}") return response_dict + def get_cache_meta_detail(self, data, check_response=True): + """Get full raw metadata detail for specified block keys""" + request = self._convert_dict_to_proto(GetCacheMetaDetailRequest, data) + response = self._stub.GetCacheMetaDetail(request, timeout=self._timeout) + response_dict = self._convert_proto_to_dict(response) + if check_response: + if response_dict['header']['status']['code'] != "OK": + raise AssertionError( + f"Request to get_cache_meta_detail failed with error: " + f"{response_dict['header']['status']['message']}") + return response_dict + def start_write_cache(self, data, check_response=True): """Start writing cache data""" request = self._convert_dict_to_proto(StartWriteCacheRequest, data) diff --git a/integration_test/meta_service/http_interface_test.py b/integration_test/meta_service/http_interface_test.py index a585bb8a8..0aadb5c04 100644 --- a/integration_test/meta_service/http_interface_test.py +++ b/integration_test/meta_service/http_interface_test.py @@ -53,6 +53,10 @@ def get_cache_location(self, data, check_response=True): """Get cache location for specified block keys""" return self._make_api_request('/api/getCacheLocation', data, check_response) + def get_cache_meta_detail(self, data, check_response=True): + """Get full raw metadata detail for specified block keys""" + return self._make_api_request('/api/getCacheMetaDetail', data, check_response) + def start_write_cache(self, data, check_response=True): """Start writing cache data""" return self._make_api_request('/api/startWriteCache', data, check_response) diff --git a/integration_test/meta_service/meta_interface_cases.py b/integration_test/meta_service/meta_interface_cases.py index 186eb9a21..3aa889a97 100644 --- a/integration_test/meta_service/meta_interface_cases.py +++ b/integration_test/meta_service/meta_interface_cases.py @@ -41,6 +41,11 @@ def get_cache_location(self, data, check_response=True) -> Dict: """Get cache location for specified block keys""" return {} + @abc.abstractmethod + def get_cache_meta_detail(self, data, check_response=True) -> Dict: + """Get full raw metadata detail for specified block keys""" + return {} + @abc.abstractmethod def start_write_cache(self, data, check_response=True) -> Dict: """Start writing cache data""" @@ -149,7 +154,35 @@ def test_basic_smoke(self): self._client.finish_write_cache(finish_write_data) - # Step 4: Get cache location to verify it was added correctly + # Step 4: Query raw metadata detail for one existing and one missing key. + detail_data = { + "trace_id": self._trace_id, + "instance_id": self._instance_id, + "block_keys": [123, 999], + "block_mask": { + "offset": 0 + }, + "detail_level": 1 + } + detail_response = self._client.get_cache_meta_detail(detail_data) + self.assertIn('items', detail_response) + self.assertEqual(2, len(detail_response['items'])) + + found_item, missing_item = detail_response['items'] + self.assertEqual(0, found_item['request_index']) + self.assertEqual(123, int(found_item['block_key'])) + self.assertEqual('OK', found_item['status']['code']) + self.assertGreater(len(found_item['locations']), 0) + self.assertTrue(any(location['status'] == 'CLS_SERVING' + for location in found_item['locations'])) + + self.assertEqual(1, missing_item['request_index']) + self.assertEqual(999, int(missing_item['block_key'])) + self.assertEqual('OK', missing_item['status']['code']) + self.assertEqual(['CLS_NOT_FOUND'], + [location['status'] for location in missing_item['locations']]) + + # Step 5: Get cache location to verify it was added correctly get_location_data = { "trace_id": self._trace_id, "query_type": "QT_PREFIX_MATCH", diff --git a/kv_cache_manager/client/include/common.h b/kv_cache_manager/client/include/common.h index f696efb26..6b9242541 100644 --- a/kv_cache_manager/client/include/common.h +++ b/kv_cache_manager/client/include/common.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -83,6 +84,47 @@ struct Metas { std::vector metas; }; +enum class CacheMetaLocationStatus : int32_t { + CLS_NOT_FOUND = 0, + CLS_NEW = 1, + CLS_WRITING = 2, + CLS_SERVING = 3, + CLS_DELETING = 4, +}; + +enum class CacheMetaStorageType : int32_t { + ST_UNSPECIFIED = 0, + ST_3FS = 1, + ST_MOONCAKE = 2, + ST_TAIRMEMPOOL = 3, + ST_NFS = 4, + ST_VCNS_3FS = 5, + ST_DUMMY = 6, + ST_EVENT_REPORT_L1P5 = 7, + ST_EVENT_REPORT_L2 = 8, +}; + +struct CacheMetaLocationDetail { + std::string location_id; + CacheMetaLocationStatus status{CacheMetaLocationStatus::CLS_NOT_FOUND}; + CacheMetaStorageType storage_type{CacheMetaStorageType::ST_UNSPECIFIED}; + int32_t spec_size{0}; + int64_t create_time{0}; + Location location_specs; +}; + +struct CacheMetaDetailItem { + ClientErrorCode error_code{ClientErrorCode::ER_OK}; + std::string error_message; + int32_t request_index{0}; + int64_t block_key{0}; + std::string prev_block_key; + std::map properties; + std::vector locations; +}; + +using CacheMetaDetails = std::vector; + using BlockMaskVector = std::vector; using BlockMaskOffset = size_t; using BlockMask = std::variant; @@ -189,4 +231,4 @@ struct TransferTraceInfo { std::vector block_ids; // block_ids.size() must be equal to block_buffer.size() }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/include/manager_client.h b/kv_cache_manager/client/include/manager_client.h index fecffb423..0230dd918 100644 --- a/kv_cache_manager/client/include/manager_client.h +++ b/kv_cache_manager/client/include/manager_client.h @@ -43,6 +43,12 @@ class ManagerClient { const BlockMask &block_mask, int32_t detail_level) = 0; + virtual std::pair MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) = 0; + virtual ClientErrorCode RemoveCache(const std::string &trace_id, const std::vector &keys, const std::vector &tokens, @@ -59,4 +65,4 @@ class ManagerClient { virtual void Shutdown() = 0; }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/include/meta_client.h b/kv_cache_manager/client/include/meta_client.h index eada8c57a..a44ac1487 100644 --- a/kv_cache_manager/client/include/meta_client.h +++ b/kv_cache_manager/client/include/meta_client.h @@ -42,6 +42,12 @@ class MetaClient { const BlockMask &block_mask, int32_t detail_level) = 0; + virtual std::pair MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) = 0; + virtual std::pair MatchLocationLen(const std::string &trace_id, QueryType query_type, const std::vector &keys, @@ -60,4 +66,4 @@ class MetaClient { virtual ClientErrorCode Init(const std::string &config, const InitParams &init_params) = 0; virtual void Shutdown() = 0; }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/src/internal/stub/grpc_stub.cc b/kv_cache_manager/client/src/internal/stub/grpc_stub.cc index 156223688..9ea85de8d 100644 --- a/kv_cache_manager/client/src/internal/stub/grpc_stub.cc +++ b/kv_cache_manager/client/src/internal/stub/grpc_stub.cc @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include "kv_cache_manager/client/src/internal/util/debug_string_util.h" #include "kv_cache_manager/common/logger.h" @@ -63,6 +65,8 @@ namespace { +kv_cache_manager::ClientErrorCode ToClientError(kv_cache_manager::proto::meta::ErrorCode service_error); + kv_cache_manager::Locations GenLocations( const google::protobuf::RepeatedPtrField<::kv_cache_manager::proto::meta::CacheLocation> &proto_locations) { kv_cache_manager::Locations locations; @@ -78,6 +82,42 @@ kv_cache_manager::Locations GenLocations( return locations; } +kv_cache_manager::CacheMetaDetails +GenCacheMetaDetails(const google::protobuf::RepeatedPtrField<::kv_cache_manager::proto::meta::CacheMetaDetailItem> + &proto_cache_meta_details) { + kv_cache_manager::CacheMetaDetails cache_meta_details; + cache_meta_details.reserve(proto_cache_meta_details.size()); + for (const auto &proto_item : proto_cache_meta_details) { + kv_cache_manager::CacheMetaDetailItem item; + if (proto_item.has_status() && proto_item.status().code() != kv_cache_manager::proto::meta::OK) { + item.error_code = ToClientError(proto_item.status().code()); + item.error_message = proto_item.status().message(); + } + item.request_index = proto_item.request_index(); + item.block_key = proto_item.block_key(); + item.prev_block_key = proto_item.prev_block_key(); + for (const auto &[property_name, property_value] : proto_item.properties()) { + item.properties[property_name] = property_value; + } + item.locations.reserve(proto_item.locations_size()); + for (const auto &proto_location : proto_item.locations()) { + kv_cache_manager::CacheMetaLocationDetail location; + location.location_id = proto_location.location_id(); + location.status = static_cast(proto_location.status()); + location.storage_type = static_cast(proto_location.type()); + location.spec_size = proto_location.spec_size(); + location.create_time = proto_location.create_time(); + location.location_specs.reserve(proto_location.location_specs_size()); + for (const auto &proto_spec : proto_location.location_specs()) { + location.location_specs.push_back({proto_spec.name(), proto_spec.uri()}); + } + item.locations.push_back(std::move(location)); + } + cache_meta_details.push_back(std::move(item)); + } + return cache_meta_details; +} + kv_cache_manager::ClientErrorCode GenCacheLocation(const kv_cache_manager::Locations &locations, google::protobuf::RepeatedPtrField<::kv_cache_manager::proto::meta::CacheLocation> *proto_locations) { @@ -322,6 +362,27 @@ std::pair GrpcStub::GetCacheMeta(const std::string &trac return {ER_OK, {locations, metas}}; } +std::pair GrpcStub::GetCacheMetaDetail(const std::string &trace_id, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level) { + auto stub = GET_AND_CHECK_STUB_WITH_TYPE(); + proto::meta::GetCacheMetaDetailRequest request; + SetKeysAndTokens(request, trace_id, instance_id, keys, tokens); + ProtoConvert::BlockMaskToProto(block_mask, request.mutable_block_mask()); + request.set_detail_level(detail_level); + grpc::ClientContext context; + proto::meta::GetCacheMetaDetailResponse response; + auto grpc_status = stub->GetCacheMetaDetail(&context, request, &response); + CHECK_GRPC_STATUS_WITH_TYPE(grpc_status); + CHECK_COMMON_HEADER_WITH_TYPE(response); + auto cache_meta_details = GenCacheMetaDetails(response.items()); + KVCM_LOG_DEBUG("get cache meta detail success, items: %lu", cache_meta_details.size()); + return {ER_OK, cache_meta_details}; +} + std::pair GrpcStub::GetCacheLocation(const std::string &trace_id, const std::string &instance_id, QueryType query_type, diff --git a/kv_cache_manager/client/src/internal/stub/grpc_stub.h b/kv_cache_manager/client/src/internal/stub/grpc_stub.h index a2e74fa3e..14869ba34 100644 --- a/kv_cache_manager/client/src/internal/stub/grpc_stub.h +++ b/kv_cache_manager/client/src/internal/stub/grpc_stub.h @@ -36,6 +36,13 @@ class GrpcStub : public Stub { const BlockMask &block_mask, int32_t detail_level) override; + std::pair GetCacheMetaDetail(const std::string &trace_id, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level) override; + std::pair GetCacheLocation(const std::string &trace_id, const std::string &instance_id, diff --git a/kv_cache_manager/client/src/internal/stub/stub.h b/kv_cache_manager/client/src/internal/stub/stub.h index 16b7cf64b..7ed61db71 100644 --- a/kv_cache_manager/client/src/internal/stub/stub.h +++ b/kv_cache_manager/client/src/internal/stub/stub.h @@ -40,6 +40,13 @@ class Stub { const BlockMask &block_mask, int32_t detail_level) = 0; + virtual std::pair GetCacheMetaDetail(const std::string &trace_id, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level) = 0; + virtual std::pair GetCacheLocation(const std::string &trace_id, const std::string &instance_id, diff --git a/kv_cache_manager/client/src/manager_client_impl.cc b/kv_cache_manager/client/src/manager_client_impl.cc index 2deae2b43..2c6e696d4 100644 --- a/kv_cache_manager/client/src/manager_client_impl.cc +++ b/kv_cache_manager/client/src/manager_client_impl.cc @@ -92,6 +92,15 @@ std::pair ManagerClientImpl::MatchMeta(const std::string return meta_client_->MatchMeta(trace_id, keys, tokens, block_mask, detail_level); } +std::pair ManagerClientImpl::MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) { + CHECK_CLIENT_WITH_TYPE(meta_client_); + return meta_client_->MatchMetaDetail(trace_id, keys, tokens, block_mask, detail_level); +} + ClientErrorCode ManagerClientImpl::RemoveCache(const std::string &trace_id, const std::vector &keys, const std::vector &tokens, @@ -122,4 +131,4 @@ std::unique_ptr ManagerClient::Create(const std::string &client_c return nullptr; } -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/src/manager_client_impl.h b/kv_cache_manager/client/src/manager_client_impl.h index 05080b089..8ad707d9a 100644 --- a/kv_cache_manager/client/src/manager_client_impl.h +++ b/kv_cache_manager/client/src/manager_client_impl.h @@ -34,6 +34,12 @@ class ManagerClientImpl : public ManagerClient { const BlockMask &block_mask, int32_t detail_level) override; + std::pair MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) override; + ClientErrorCode RemoveCache(const std::string &trace_id, const std::vector &keys, const std::vector &tokens, @@ -56,4 +62,4 @@ class ManagerClientImpl : public ManagerClient { std::unique_ptr meta_client_; std::unique_ptr transfer_client_; }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/src/meta_client_impl.cc b/kv_cache_manager/client/src/meta_client_impl.cc index dfa401287..e72305836 100644 --- a/kv_cache_manager/client/src/meta_client_impl.cc +++ b/kv_cache_manager/client/src/meta_client_impl.cc @@ -149,6 +149,21 @@ std::pair MetaClientImpl::MatchMeta(const std::string &t return stub_->GetCacheMeta(trace_id, instance_id, keys, tokens, block_mask, detail_level); } +std::pair MetaClientImpl::MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) { + KVCM_LOG_DEBUG("match meta detail with trace_id [%s], keys %s, tokens %s, block_mask %s, detail_level [%d]", + trace_id.c_str(), + DebugStringUtil::ToString(keys).c_str(), + DebugStringUtil::ToString(tokens).c_str(), + DebugStringUtil::ToString(block_mask).c_str(), + detail_level); + const std::string &instance_id = CHECK_INSTANCE_STUB_WITH_TYPE(); + return stub_->GetCacheMetaDetail(trace_id, instance_id, keys, tokens, block_mask, detail_level); +} + std::pair MetaClientImpl::StartWrite(const std::string &trace_id, const std::vector &keys, diff --git a/kv_cache_manager/client/src/meta_client_impl.h b/kv_cache_manager/client/src/meta_client_impl.h index df8a65cbb..c71a7c1e5 100644 --- a/kv_cache_manager/client/src/meta_client_impl.h +++ b/kv_cache_manager/client/src/meta_client_impl.h @@ -44,6 +44,12 @@ class MetaClientImpl : public MetaClient { const BlockMask &block_mask, int32_t detail_level) override; + std::pair MatchMetaDetail(const std::string &trace_id, + const std::vector &keys, + const std::vector &tokens, + const BlockMask &block_mask, + int32_t detail_level) override; + ClientErrorCode RemoveCache(const std::string &trace_id, const std::vector &keys, const std::vector &tokens, @@ -69,4 +75,4 @@ class MetaClientImpl : public MetaClient { std::string storage_config_; mutable std::shared_mutex config_mutex_; }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/client/test/meta_client_test.cc b/kv_cache_manager/client/test/meta_client_test.cc index d8f7e5d10..cfaec3047 100644 --- a/kv_cache_manager/client/test/meta_client_test.cc +++ b/kv_cache_manager/client/test/meta_client_test.cc @@ -89,6 +89,16 @@ class MockStub : public Stub { int32_t detail_level), (override)); + MOCK_METHOD((std::pair), + GetCacheMetaDetail, + (const std::string &trace_id, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level), + (override)); + MOCK_METHOD((std::pair), GetCacheLocation, (const std::string &trace_id, diff --git a/kv_cache_manager/manager/BUILD b/kv_cache_manager/manager/BUILD index 1005029f3..eff5603f3 100644 --- a/kv_cache_manager/manager/BUILD +++ b/kv_cache_manager/manager/BUILD @@ -66,6 +66,7 @@ cc_library( ], deps = [ "//kv_cache_manager/meta:cache_location", + "//kv_cache_manager/meta:types", ], ) diff --git a/kv_cache_manager/manager/cache_location_view.h b/kv_cache_manager/manager/cache_location_view.h index 40a07a53c..7fb65c586 100644 --- a/kv_cache_manager/manager/cache_location_view.h +++ b/kv_cache_manager/manager/cache_location_view.h @@ -1,10 +1,14 @@ #pragma once +#include +#include +#include #include #include #include "kv_cache_manager/data_storage/common_define.h" #include "kv_cache_manager/meta/cache_location.h" +#include "kv_cache_manager/meta/types.h" namespace kv_cache_manager { @@ -74,6 +78,25 @@ class CacheMetaVecWrapper { CacheLocationViewVecWrapper locations_; }; +struct CacheLocationMetaDetail { + std::string location_id; + CacheLocationStatus status = CacheLocationStatus::CLS_NOT_FOUND; + DataStorageType type = DataStorageType::DATA_STORAGE_TYPE_UNKNOWN; + int32_t spec_size = 0; + int64_t create_time = 0; + std::vector location_specs; +}; + +struct CacheKeyMetaDetail { + ErrorCode error_code = EC_OK; + size_t request_index = 0; + KeyType block_key = 0; + PropertyMap properties; + std::vector locations; +}; + +using CacheMetaDetailVec = std::vector; + class StartWriteCacheInfo { public: StartWriteCacheInfo() = default; @@ -97,4 +120,4 @@ class StartWriteCacheInfo { CacheLocationViewVecWrapper locations_; }; -} // namespace kv_cache_manager \ No newline at end of file +} // namespace kv_cache_manager diff --git a/kv_cache_manager/manager/cache_manager.cc b/kv_cache_manager/manager/cache_manager.cc index 166487f34..93a7a8151 100644 --- a/kv_cache_manager/manager/cache_manager.cc +++ b/kv_cache_manager/manager/cache_manager.cc @@ -750,6 +750,125 @@ std::pair CacheManager::GetCacheMeta(RequestCont return {ec, CacheMetaVecWrapper(std::move(metas), std::move(cache_locations))}; } +std::pair CacheManager::GetCacheMetaDetail(RequestContext *request_context, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level /*reserved*/) { + SPAN_TRACER(request_context); + (void)detail_level; + const std::string &trace_id = request_context->trace_id(); + auto *service_metrics_collector = dynamic_cast(request_context->metrics_collector()); + auto [ec, meta_searcher] = CheckInputAndGetMetaSearcher(request_context, instance_id, keys, tokens); + RETURN_IF_EC_NOT_OK_WITH_TYPE_LOG(DEBUG, ec, CacheMetaDetailVec, "get cache meta detail failed"); + + KeyVector all_keys; + if (!keys.empty()) { + all_keys = keys; + } else { + auto [ec_temp, block_size] = GetBlockSize(request_context, instance_id); + RETURN_IF_EC_NOT_OK_WITH_TYPE_LOG(DEBUG, ec_temp, CacheMetaDetailVec, "get cache meta detail failed"); + all_keys = GenKeyVector(tokens, block_size); + } + + KeyVector query_keys; + std::vector request_indices; + query_keys.reserve(all_keys.size()); + request_indices.reserve(all_keys.size()); + for (size_t idx = 0; idx < all_keys.size(); ++idx) { + if (IsIndexInMaskRange(block_mask, idx)) { + continue; + } + query_keys.push_back(all_keys[idx]); + request_indices.push_back(idx); + } + if (query_keys.empty()) { + return {EC_OK, CacheMetaDetailVec()}; + } + + CacheLocationMapVector location_maps; + PropertyMapVector properties; + std::vector per_key_ecs; + KVCM_METRICS_COLLECTOR_SET_METRICS(service_metrics_collector, manager, request_key_count, query_keys.size()); + KVCM_METRICS_COLLECTOR_CHRONO_MARK_BEGIN(service_metrics_collector, ManagerBatchGetLocation); + ec = meta_searcher->BatchGetRawMeta(request_context, query_keys, location_maps, properties, per_key_ecs); + KVCM_METRICS_COLLECTOR_CHRONO_MARK_END(service_metrics_collector, ManagerBatchGetLocation); + RETURN_IF_EC_NOT_OK_WITH_TYPE_LOG(DEBUG, ec, CacheMetaDetailVec, "get cache meta detail failed: BatchGetRawMeta"); + + CacheMetaDetailVec details; + details.reserve(query_keys.size()); + for (size_t idx = 0; idx < query_keys.size(); ++idx) { + CacheKeyMetaDetail item; + item.request_index = request_indices[idx]; + item.block_key = query_keys[idx]; + + if (idx >= per_key_ecs.size()) { + item.error_code = EC_MISMATCH; + details.push_back(std::move(item)); + continue; + } + if (per_key_ecs[idx] != EC_OK && per_key_ecs[idx] != EC_NOENT) { + item.error_code = per_key_ecs[idx]; + details.push_back(std::move(item)); + continue; + } + if (idx >= location_maps.size() || idx >= properties.size()) { + item.error_code = EC_MISMATCH; + details.push_back(std::move(item)); + continue; + } + item.properties = properties[idx]; + + const bool key_not_found = per_key_ecs[idx] == ErrorCode::EC_NOENT || location_maps[idx].empty(); + if (key_not_found) { + CacheLocationMetaDetail not_found; + not_found.status = CacheLocationStatus::CLS_NOT_FOUND; + item.locations.push_back(std::move(not_found)); + details.push_back(std::move(item)); + continue; + } + + std::vector location_ids; + location_ids.reserve(location_maps[idx].size()); + for (const auto &[location_id, location] : location_maps[idx]) { + if (location) { + location_ids.push_back(location_id); + } + } + std::sort(location_ids.begin(), location_ids.end()); + for (const auto &location_id : location_ids) { + const auto location_iter = location_maps[idx].find(location_id); + if (location_iter == location_maps[idx].end() || !location_iter->second) { + continue; + } + const auto &raw_location = *location_iter->second; + CacheLocationMetaDetail location_detail; + location_detail.location_id = raw_location.id().empty() ? location_id : raw_location.id(); + location_detail.status = raw_location.status(); + location_detail.type = raw_location.type(); + location_detail.spec_size = static_cast(raw_location.spec_size()); + location_detail.create_time = raw_location.create_time(); + location_detail.location_specs = raw_location.location_specs(); + item.locations.push_back(std::move(location_detail)); + } + if (item.locations.empty()) { + CacheLocationMetaDetail not_found; + not_found.status = CacheLocationStatus::CLS_NOT_FOUND; + item.locations.push_back(std::move(not_found)); + } + details.push_back(std::move(item)); + } + + const auto first_item_error = + std::find_if(details.begin(), details.end(), [](const auto &item) { return item.error_code != EC_OK; }); + if (first_item_error != details.end() && + std::all_of(details.begin(), details.end(), [](const auto &item) { return item.error_code != EC_OK; })) { + return {first_item_error->error_code, std::move(details)}; + } + return {EC_OK, std::move(details)}; +} + ErrorCode CacheManager::PerformCacheLocationQuery(RequestContext *request_context, ServiceMetricsCollector *service_metrics_collector, MetaSearcher *meta_searcher, diff --git a/kv_cache_manager/manager/cache_manager.h b/kv_cache_manager/manager/cache_manager.h index d42a9ae53..85f28c572 100644 --- a/kv_cache_manager/manager/cache_manager.h +++ b/kv_cache_manager/manager/cache_manager.h @@ -140,6 +140,13 @@ class CacheManager { const BlockMask &block_mask, int32_t detail_level /*TODO*/); + std::pair GetCacheMetaDetail(RequestContext *request_context, + const std::string &instance_id, + const KeyVector &keys, + const TokenIdsVector &tokens, + const BlockMask &block_mask, + int32_t detail_level /*reserved*/); + std::pair GetCacheLocation(RequestContext *request_context, const std::string &instance_id, diff --git a/kv_cache_manager/manager/meta_searcher.cc b/kv_cache_manager/manager/meta_searcher.cc index c9408d679..d5d9183fd 100644 --- a/kv_cache_manager/manager/meta_searcher.cc +++ b/kv_cache_manager/manager/meta_searcher.cc @@ -2072,6 +2072,33 @@ ErrorCode MetaSearcher::BatchGetLocation(RequestContext *request_context, return EC_OK; } +ErrorCode MetaSearcher::BatchGetRawMeta(RequestContext *request_context, + const KeyVector &keys, + CacheLocationMapVector &out_location_maps, + PropertyMapVector &out_properties, + std::vector &out_error_codes) { + out_location_maps.clear(); + out_properties.clear(); + out_error_codes.clear(); + + if (keys.empty()) { + return EC_OK; + } + + auto *service_metrics_collector = dynamic_cast(request_context->metrics_collector()); + KVCM_METRICS_COLLECTOR_CHRONO_MARK_BEGIN(service_metrics_collector, MetaSearcherIndexerGet); + auto result = meta_indexer_->Get(request_context, keys, out_location_maps, out_properties); + KVCM_METRICS_COLLECTOR_CHRONO_MARK_END(service_metrics_collector, MetaSearcherIndexerGet); + out_error_codes = std::move(result.error_codes); + for (size_t idx = 0; idx < keys.size() && idx < out_error_codes.size(); idx++) { + if (out_error_codes[idx] != ErrorCode::EC_OK && out_error_codes[idx] != ErrorCode::EC_NOENT) { + KVCM_LOG_WARN( + "get raw meta failed, key[%lu](%lu), error_code: %d", idx, keys[idx], out_error_codes[idx]); + } + } + return EC_OK; +} + ErrorCode MetaSearcher::BatchAddLocation(RequestContext *request_context, const KeyVector &keys, const CacheLocationVector &locations, diff --git a/kv_cache_manager/manager/meta_searcher.h b/kv_cache_manager/manager/meta_searcher.h index 87d1422d3..e733384c9 100644 --- a/kv_cache_manager/manager/meta_searcher.h +++ b/kv_cache_manager/manager/meta_searcher.h @@ -122,6 +122,11 @@ class MetaSearcher { // EC_OK 时是可供业务使用的 location id；失败时若非空，仅可作为回滚定位符。 std::string location_id; }; + ErrorCode BatchGetRawMeta(RequestContext *request_context, + const KeyVector &keys, + CacheLocationMapVector &out_location_maps, + PropertyMapVector &out_properties, + std::vector &out_error_codes); ErrorCode BatchAddLocation(RequestContext *request_context, const KeyVector &keys, const CacheLocationVector &locations, diff --git a/kv_cache_manager/manager/test/cache_manager_test.cc b/kv_cache_manager/manager/test/cache_manager_test.cc index 7f6e67501..e106e3813 100644 --- a/kv_cache_manager/manager/test/cache_manager_test.cc +++ b/kv_cache_manager/manager/test/cache_manager_test.cc @@ -85,6 +85,38 @@ ErrorCode ReadError_stub(void * /*obj*/, } } // namespace mark_query_read_error_stub +namespace raw_meta_partial_error_stub { +ErrorCode BatchGetRawMeta_stub(void * /*obj*/, + RequestContext * /*request_context*/, + const KeyVector &keys, + CacheLocationMapVector &out_location_maps, + PropertyMapVector &out_properties, + std::vector &out_error_codes) { + out_location_maps.assign(keys.size(), {}); + out_properties.assign(keys.size(), {}); + out_error_codes.assign(keys.size(), EC_OK); + + for (size_t idx = 0; idx < keys.size(); ++idx) { + if (keys[idx] == 22) { + out_error_codes[idx] = EC_TIMEOUT; + continue; + } + if (keys[idx] == 33) { + out_error_codes[idx] = EC_NOENT; + continue; + } + auto location = std::make_shared("location_ok", + CacheLocationStatus::CLS_SERVING, + DataStorageType::DATA_STORAGE_TYPE_NFS, + 1, + std::vector{{"tp0", "file:///tmp/key?size=1"}}); + out_location_maps[idx][location->id()] = std::move(location); + out_properties[idx][PROPERTY_PREV_BLOCK_KEY] = ""; + } + return EC_OK; +} +} // namespace raw_meta_partial_error_stub + namespace remove_instance_reclaimer_state_stub { CacheReclaimer *reclaimer = nullptr; bool called = false; @@ -2123,6 +2155,35 @@ TEST_F(CacheManagerTest, TestGetNotExistCacheMeta) { } } +TEST_F(CacheManagerTest, TestGetCacheMetaDetailPreservesPartialResults) { + Stub stub; + stub.set(ADDR(MetaSearcher, BatchGetRawMeta), raw_meta_partial_error_stub::BatchGetRawMeta_stub); + + BlockMask block_mask = static_cast(0); + auto [ec, details] = + cache_manager_->GetCacheMetaDetail(request_context_.get(), "test_instance", {11, 22, 33}, {}, block_mask, 1); + + ASSERT_EQ(EC_OK, ec); + ASSERT_EQ(3, details.size()); + + EXPECT_EQ(EC_OK, details[0].error_code); + ASSERT_EQ(1, details[0].locations.size()); + EXPECT_EQ(CacheLocationStatus::CLS_SERVING, details[0].locations[0].status); + + EXPECT_EQ(EC_TIMEOUT, details[1].error_code); + EXPECT_TRUE(details[1].locations.empty()); + + EXPECT_EQ(EC_OK, details[2].error_code); + ASSERT_EQ(1, details[2].locations.size()); + EXPECT_EQ(CacheLocationStatus::CLS_NOT_FOUND, details[2].locations[0].status); + + auto [all_failed_ec, all_failed_details] = + cache_manager_->GetCacheMetaDetail(request_context_.get(), "test_instance", {22}, {}, block_mask, 1); + EXPECT_EQ(EC_TIMEOUT, all_failed_ec); + ASSERT_EQ(1, all_failed_details.size()); + EXPECT_EQ(EC_TIMEOUT, all_failed_details[0].error_code); +} + TEST_F(CacheManagerTest, TestRemoveCache) { auto expected = std::pair(EC_OK, default_storage_configs); ASSERT_EQ(expected, diff --git a/kv_cache_manager/protocol/protobuf/meta_service.proto b/kv_cache_manager/protocol/protobuf/meta_service.proto index a63c64c32..da922862b 100644 --- a/kv_cache_manager/protocol/protobuf/meta_service.proto +++ b/kv_cache_manager/protocol/protobuf/meta_service.proto @@ -403,6 +403,51 @@ message GetCacheMetaResponse { repeated string metas = 3; // 与请求key一一对应 } +message GetCacheMetaDetailRequest { + string trace_id = 1; + string instance_id = 2; + repeated int64 block_keys = 3; + repeated int64 token_ids = 4; + BlockMask block_mask = 5; // 对block_keys或者token_ids的mask + int32 detail_level = 6; // 预留，用于控制诊断明细级别 +} + +enum CacheLocationStatus { + CLS_NOT_FOUND = 0; + CLS_NEW = 1; + CLS_WRITING = 2; + CLS_SERVING = 3; + CLS_DELETING = 4; +} + +message CacheLocationDetail { + string location_id = 1; + CacheLocationStatus status = 2; + StorageType type = 3; + int32 spec_size = 4; + int64 create_time = 5; + repeated LocationSpec location_specs = 6; +} + +message CacheMetaDetailItem { + int32 request_index = 1; + int64 block_key = 2; + // Convenience projection of properties["BP#prev_key"]. The raw property is + // intentionally retained in properties as well. + string prev_block_key = 3; + map properties = 4; + repeated CacheLocationDetail locations = 5; + // Per-key raw metadata lookup status. A missing key is a successful lookup + // represented by CLS_NOT_FOUND; backend/read failures are reported here so + // other successful items can still be returned. + Status status = 6; +} + +message GetCacheMetaDetailResponse { + CommonResponseHeader header = 1; + repeated CacheMetaDetailItem items = 2; +} + // 在每次写入 cache 前, 需要调用这个接口获得CacheLocation信息 message StartWriteCacheRequest { string trace_id = 1; @@ -496,6 +541,8 @@ service MetaService { rpc GetInstanceInfo(GetInstanceInfoRequest) returns (GetInstanceInfoResponse); // 在排查问题时获取meta完整信息，明确的; TODO : 把这个挪到debug service中 rpc GetCacheMeta(GetCacheMetaRequest) returns (GetCacheMetaResponse); + // 查询指定 key 的完整 raw metadata，不做 location 选择、数据文件检查或 lazy prune + rpc GetCacheMetaDetail(GetCacheMetaDetailRequest) returns (GetCacheMetaDetailResponse); // 返回Cache的位置，由PD节点调用 rpc GetCacheLocation(GetCacheLocationRequest) returns (GetCacheLocationResponse); // 返回命中的key的数量 diff --git a/kv_cache_manager/py_connector/common/manager_client.py b/kv_cache_manager/py_connector/common/manager_client.py index b2c9321f8..a41624120 100644 --- a/kv_cache_manager/py_connector/common/manager_client.py +++ b/kv_cache_manager/py_connector/common/manager_client.py @@ -409,6 +409,10 @@ def get_cache_locations_by_backend(self, data, check_response=True): """Get cache locations selected independently for each storage backend.""" return self._make_api_request('/api/getCacheLocationsByBackend', data, check_response) + def get_cache_meta_detail(self, data, check_response=True): + """Get full raw metadata detail for specified block keys""" + return self._make_api_request('/api/getCacheMetaDetail', data, check_response) + def start_write_cache(self, data, check_response=True): """Start writing cache data""" return self._make_api_request('/api/startWriteCache', data, check_response) diff --git a/kv_cache_manager/service/grpc_service/meta_service_grpc.cc b/kv_cache_manager/service/grpc_service/meta_service_grpc.cc index 4d9fb68cd..e797c8f57 100644 --- a/kv_cache_manager/service/grpc_service/meta_service_grpc.cc +++ b/kv_cache_manager/service/grpc_service/meta_service_grpc.cc @@ -46,6 +46,14 @@ grpc::Status MetaServiceGRpc::GetCacheMeta(grpc::ServerContext *context, return grpc::Status::OK; } +grpc::Status MetaServiceGRpc::GetCacheMetaDetail(grpc::ServerContext *context, + const proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response) { + API_CONTEXT_GET_COLLECTOR_AND_INIT_GRPC(GetCacheMetaDetail, grpc::Status::OK); + meta_service_impl_->GetCacheMetaDetail(request_context, request, response); + return grpc::Status::OK; +} + grpc::Status MetaServiceGRpc::GetCacheLocation(grpc::ServerContext *context, const proto::meta::GetCacheLocationRequest *request, proto::meta::GetCacheLocationResponse *response) { diff --git a/kv_cache_manager/service/grpc_service/meta_service_grpc.h b/kv_cache_manager/service/grpc_service/meta_service_grpc.h index 1b782e73c..13642e7fc 100644 --- a/kv_cache_manager/service/grpc_service/meta_service_grpc.h +++ b/kv_cache_manager/service/grpc_service/meta_service_grpc.h @@ -33,6 +33,10 @@ class MetaServiceGRpc final : public proto::meta::MetaService::Service, public M const proto::meta::GetCacheMetaRequest *request, proto::meta::GetCacheMetaResponse *response) override; + grpc::Status GetCacheMetaDetail(grpc::ServerContext *context, + const proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response) override; + grpc::Status GetCacheLocation(grpc::ServerContext *context, const proto::meta::GetCacheLocationRequest *request, proto::meta::GetCacheLocationResponse *response) override; diff --git a/kv_cache_manager/service/http_service/meta_service_http.cc b/kv_cache_manager/service/http_service/meta_service_http.cc index 8d45efeab..094fc46c6 100644 --- a/kv_cache_manager/service/http_service/meta_service_http.cc +++ b/kv_cache_manager/service/http_service/meta_service_http.cc @@ -30,6 +30,8 @@ void MetaServiceHttp::RegisterHandler() { Post, registerInstance, RegisterInstance, RegisterInstance, RegisterInstance); REGISTER_HTTP_HANDLER_FOR_META_SERVICE(Post, getInstanceInfo, GetInstanceInfo, GetInstanceInfo, GetInstanceInfo); REGISTER_HTTP_HANDLER_FOR_META_SERVICE(Post, getCacheMeta, GetCacheMeta, GetCacheMeta, GetCacheMeta); + REGISTER_HTTP_HANDLER_FOR_META_SERVICE( + Post, getCacheMetaDetail, GetCacheMetaDetail, GetCacheMetaDetail, GetCacheMetaDetail); REGISTER_HTTP_HANDLER_FOR_META_SERVICE( Post, getCacheLocation, GetCacheLocation, GetCacheLocation, GetCacheLocation); REGISTER_HTTP_HANDLER_FOR_META_SERVICE( @@ -114,6 +116,23 @@ void MetaServiceHttp::GetCacheMeta(coro_http::coro_http_connection *http_conn, meta_service_impl_->GetCacheMeta(request_context, request, response); } +void MetaServiceHttp::GetCacheMetaDetail(coro_http::coro_http_connection *http_conn, + proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response) { + API_CONTEXT_GET_COLLECTOR_AND_INIT_HTTP(GetCacheMetaDetail, __NOTHING__); + KVCM_LOG_INFO("[traceId: %s] GetCacheMetaDetail called with instance id: %s, block keys count: %d, " + "token ids count: %d, detail level: %d", + request->trace_id().c_str(), + request->instance_id().c_str(), + request->block_keys_size(), + request->token_ids_size(), + request->detail_level()); + KVCM_LOG_DEBUG("[traceId: %s] GetCacheMetaDetail request details: %s", + request->trace_id().c_str(), + request->ShortDebugString().c_str()); + meta_service_impl_->GetCacheMetaDetail(request_context, request, response); +} + void MetaServiceHttp::StartWriteCache(coro_http::coro_http_connection *http_conn, proto::meta::StartWriteCacheRequest *request, proto::meta::StartWriteCacheResponse *response) { diff --git a/kv_cache_manager/service/http_service/meta_service_http.h b/kv_cache_manager/service/http_service/meta_service_http.h index 26d39aa1a..39a1bf0b1 100644 --- a/kv_cache_manager/service/http_service/meta_service_http.h +++ b/kv_cache_manager/service/http_service/meta_service_http.h @@ -33,6 +33,9 @@ class MetaServiceHttp : public CoroHttpService, public MetaServiceMetricsBase { void GetCacheMeta(coro_http::coro_http_connection *http_conn, proto::meta::GetCacheMetaRequest *request, proto::meta::GetCacheMetaResponse *response); + void GetCacheMetaDetail(coro_http::coro_http_connection *http_conn, + proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response); void GetCacheLocation(coro_http::coro_http_connection *http_conn, proto::meta::GetCacheLocationRequest *request, proto::meta::GetCacheLocationResponse *response); diff --git a/kv_cache_manager/service/meta_service_impl.cc b/kv_cache_manager/service/meta_service_impl.cc index 0865c79ac..8516bb6c7 100644 --- a/kv_cache_manager/service/meta_service_impl.cc +++ b/kv_cache_manager/service/meta_service_impl.cc @@ -689,6 +689,68 @@ void MetaServiceImpl::GetCacheMeta(RequestContext *request_context, SET_SPAN_TRACER_STR_IN_HEADER(request_context); } +void MetaServiceImpl::GetCacheMetaDetail(RequestContext *request_context, + const proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response) { + SPAN_TRACER(request_context); + API_CALL_GUARD("GetCacheMetaDetail", true); + auto *header = response->mutable_header(); + auto *status = header->mutable_status(); + CHECK_FAULT_INJECTION("GetCacheMetaDetail"); + std::string invalid_fields = "missing or invalid fields: "; + if (request->instance_id().empty()) { + CHECK_REQUIRED_FIELDS_VALIDATION("GetCacheMetaDetail", "instance_id", true); + SET_SPAN_TRACER_STR_IN_HEADER(request_context); + return; + } + if (request->block_keys().empty() && request->token_ids().empty()) { + CHECK_REQUIRED_FIELDS_VALIDATION("GetCacheMetaDetail", "block_keys and token_ids", true); + SET_SPAN_TRACER_STR_IN_HEADER(request_context); + return; + } + + BlockMask block_mask_req; + ProtoConvert::BlockMaskFromProto(&request->block_mask(), block_mask_req); + auto [ec_info, cache_meta_details] = cache_manager_->GetCacheMetaDetail( + request_context, + request->instance_id(), + std::vector(request->block_keys().begin(), request->block_keys().end()), + std::vector(request->token_ids().begin(), request->token_ids().end()), + block_mask_req, + request->detail_level()); + + size_t item_error_count = 0; + for (const auto &cache_meta_detail : cache_meta_details) { + item_error_count += cache_meta_detail.error_code != EC_OK; + ProtoConvert::CacheKeyMetaDetailToProto(cache_meta_detail, response->add_items()); + } + + if (ec_info != EC_OK) { + status->set_code(ToMetaPbError(ec_info)); + request_context->set_status_code(status->code()); + status->set_message("Failed to get cache metadata detail : " + request_context->error_tracer()->ToJsonString()); + KVCM_LOG_ERROR("[traceId: %s] GetCacheMetaDetail failed, ec: %d, returned %d items with %zu item errors", + request->trace_id().c_str(), + ec_info, + response->items_size(), + item_error_count); + } else { + status->set_code(proto::meta::OK); + request_context->set_status_code(status->code()); + if (item_error_count == 0) { + status->set_message("Cache metadata detail retrieved successfully"); + } else { + status->set_message("Cache metadata detail retrieved with " + std::to_string(item_error_count) + + " item errors"); + } + KVCM_LOG_INFO("[traceId: %s] GetCacheMetaDetail succeeded, returned %d items with %zu item errors", + request->trace_id().c_str(), + response->items_size(), + item_error_count); + } + SET_SPAN_TRACER_STR_IN_HEADER(request_context); +} + void MetaServiceImpl::StartWriteCache(RequestContext *request_context, const proto::meta::StartWriteCacheRequest *request, proto::meta::StartWriteCacheResponse *response) { diff --git a/kv_cache_manager/service/meta_service_impl.h b/kv_cache_manager/service/meta_service_impl.h index 02eb1af03..f5f91b683 100644 --- a/kv_cache_manager/service/meta_service_impl.h +++ b/kv_cache_manager/service/meta_service_impl.h @@ -45,6 +45,10 @@ class MetaServiceImpl : public ServiceImplBase { const proto::meta::GetCacheMetaRequest *request, proto::meta::GetCacheMetaResponse *response); + void GetCacheMetaDetail(RequestContext *request_context, + const proto::meta::GetCacheMetaDetailRequest *request, + proto::meta::GetCacheMetaDetailResponse *response); + void StartWriteCache(RequestContext *request_context, const proto::meta::StartWriteCacheRequest *request, proto::meta::StartWriteCacheResponse *response); diff --git a/kv_cache_manager/service/meta_service_metrics_base.cc b/kv_cache_manager/service/meta_service_metrics_base.cc index 54a52c4c4..70b3c7b3f 100644 --- a/kv_cache_manager/service/meta_service_metrics_base.cc +++ b/kv_cache_manager/service/meta_service_metrics_base.cc @@ -60,6 +60,7 @@ void MetaServiceMetricsBase::InvalidateCollectorCache(const std::string &instanc } KVCM_INVALIDATE_METRICS_COLLECTOR_MAP_(GetCacheMeta, instance_id); + KVCM_INVALIDATE_METRICS_COLLECTOR_MAP_(GetCacheMetaDetail, instance_id); KVCM_INVALIDATE_METRICS_COLLECTOR_MAP_(GetCacheLocation, instance_id); KVCM_INVALIDATE_METRICS_COLLECTOR_MAP_(GetCacheLocationsByBackend, instance_id); KVCM_INVALIDATE_METRICS_COLLECTOR_MAP_(GetCacheLocationLen, instance_id); @@ -262,6 +263,7 @@ void MetaServiceMetricsBase::AttachReportEventTypeMetricsCollectors(const proto: } KVCM_DEFINE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheMeta); +KVCM_DEFINE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheMetaDetail); KVCM_DEFINE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocation); KVCM_DEFINE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocationsByBackend); KVCM_DEFINE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocationLen); diff --git a/kv_cache_manager/service/meta_service_metrics_base.h b/kv_cache_manager/service/meta_service_metrics_base.h index 72091baa8..46e708b98 100644 --- a/kv_cache_manager/service/meta_service_metrics_base.h +++ b/kv_cache_manager/service/meta_service_metrics_base.h @@ -64,6 +64,7 @@ class MetaServiceMetricsBase { void InvalidateCollectorCache(const std::string &instance_id); KVCM_DECLARE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheMeta); + KVCM_DECLARE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheMetaDetail); KVCM_DECLARE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocation); KVCM_DECLARE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocationsByBackend); KVCM_DECLARE_METRICS_COLLECTOR_MAP_METHOD_(GetCacheLocationLen); @@ -98,6 +99,7 @@ class MetaServiceMetricsBase { KVCM_DECLARE_METRICS_COLLECTOR_(GetClusterInfo); KVCM_DECLARE_METRICS_COLLECTOR_(ReportEvent); KVCM_DECLARE_METRICS_COLLECTOR_MAP_(GetCacheMeta); + KVCM_DECLARE_METRICS_COLLECTOR_MAP_(GetCacheMetaDetail); KVCM_DECLARE_METRICS_COLLECTOR_MAP_(GetCacheLocation); KVCM_DECLARE_METRICS_COLLECTOR_MAP_(GetCacheLocationsByBackend); KVCM_DECLARE_METRICS_COLLECTOR_MAP_(GetCacheLocationLen); diff --git a/kv_cache_manager/service/test/BUILD b/kv_cache_manager/service/test/BUILD index f2c934838..56e598e84 100644 --- a/kv_cache_manager/service/test/BUILD +++ b/kv_cache_manager/service/test/BUILD @@ -15,6 +15,24 @@ cc_test( ], ) +cc_test( + name = "MetaServiceImplTest", + srcs = [ + "meta_service_impl_test.cc", + ], + copts = ["-fno-access-control"], + data = [], + linkstatic = True, + deps = [ + "//kv_cache_manager/common:unittest", + "//kv_cache_manager/config", + "//kv_cache_manager/manager:cache_manager", + "//kv_cache_manager/metrics:metrics_registry", + "//kv_cache_manager/service:meta_service_impl", + "@cpp_stub", + ], +) + cc_test( name = "CommandLineTest", srcs = [ diff --git a/kv_cache_manager/service/test/meta_service_impl_test.cc b/kv_cache_manager/service/test/meta_service_impl_test.cc new file mode 100644 index 000000000..89a47d95c --- /dev/null +++ b/kv_cache_manager/service/test/meta_service_impl_test.cc @@ -0,0 +1,60 @@ +#include +#include +#include +#include + +#include "kv_cache_manager/common/request_context.h" +#include "kv_cache_manager/common/unittest.h" +#include "kv_cache_manager/config/registry_manager.h" +#include "kv_cache_manager/manager/cache_location_view.h" +#include "kv_cache_manager/manager/cache_manager.h" +#include "kv_cache_manager/metrics/metrics_registry.h" +#include "kv_cache_manager/service/meta_service_impl.h" +#include "stub.h" + +namespace kv_cache_manager { +namespace { + +std::pair +GetCacheMetaDetailAllFailedStub(void * /*obj*/, + RequestContext * /*request_context*/, + const std::string & /*instance_id*/, + const CacheManager::KeyVector & /*keys*/, + const CacheManager::TokenIdsVector & /*tokens*/, + const BlockMask & /*block_mask*/, + int32_t /*detail_level*/) { + CacheKeyMetaDetail item; + item.error_code = EC_TIMEOUT; + item.request_index = 0; + item.block_key = 22; + return {EC_TIMEOUT, {std::move(item)}}; +} + +} // namespace + +TEST(MetaServiceImplTest, PreservesPerKeyDetailsWhenEveryLookupFails) { + auto metrics_registry = std::make_shared(); + auto registry_manager = std::make_shared("", metrics_registry); + auto cache_manager = std::make_shared(metrics_registry, registry_manager); + MetaServiceImpl service(cache_manager, /*metrics_reporter*/ nullptr, /*leader_elector*/ nullptr); + + Stub stub; + stub.set(ADDR(CacheManager, GetCacheMetaDetail), GetCacheMetaDetailAllFailedStub); + + RequestContext request_context("all_failed"); + proto::meta::GetCacheMetaDetailRequest request; + request.set_trace_id("all_failed"); + request.set_instance_id("test_instance"); + request.add_block_keys(22); + proto::meta::GetCacheMetaDetailResponse response; + + service.GetCacheMetaDetail(&request_context, &request, &response); + + EXPECT_EQ(proto::meta::INTERNAL_ERROR, response.header().status().code()); + ASSERT_EQ(1, response.items_size()); + EXPECT_EQ(0, response.items(0).request_index()); + EXPECT_EQ(22, response.items(0).block_key()); + EXPECT_EQ(proto::meta::INTERNAL_ERROR, response.items(0).status().code()); +} + +} // namespace kv_cache_manager diff --git a/kv_cache_manager/service/util/manager_message_proto_util.h b/kv_cache_manager/service/util/manager_message_proto_util.h index 7258e36c2..b0007c471 100644 --- a/kv_cache_manager/service/util/manager_message_proto_util.h +++ b/kv_cache_manager/service/util/manager_message_proto_util.h @@ -18,6 +18,7 @@ #include "kv_cache_manager/data_storage/storage_config.h" #include "kv_cache_manager/manager/cache_location_view.h" #include "kv_cache_manager/meta/cache_location.h" +#include "kv_cache_manager/meta/common.h" #include "kv_cache_manager/protocol/protobuf/admin_service.pb.h" #include "kv_cache_manager/protocol/protobuf/meta_service.pb.h" namespace kv_cache_manager { @@ -44,6 +45,12 @@ class ProtoConvert { static void DataStorageTypeToProto(const DataStorageType &data_storage_type_info, T *proto_data_storage_type); template static void DataStorageTypeFromProto(const T proto_data_storage_type, DataStorageType &data_storage_type_info); + static void CacheLocationStatusToProto(const CacheLocationStatus &status_info, + proto::meta::CacheLocationStatus *proto_status); + static void CacheLocationMetaDetailToProto(const CacheLocationMetaDetail &cache_location_info, + proto::meta::CacheLocationDetail *proto_cache_location); + static void CacheKeyMetaDetailToProto(const CacheKeyMetaDetail &cache_meta_detail, + proto::meta::CacheMetaDetailItem *proto_cache_meta_detail); static void StorageConfigToProto(const StorageConfig &storage_config, proto::admin::StorageConfig *proto_storage_config); @@ -366,6 +373,68 @@ void ProtoConvert::DataStorageTypeFromProto(const T proto_data_storage_type, Dat } } +inline void ProtoConvert::CacheLocationStatusToProto(const CacheLocationStatus &status_info, + proto::meta::CacheLocationStatus *proto_status) { + switch (status_info) { + case CacheLocationStatus::CLS_NEW: + *proto_status = proto::meta::CLS_NEW; + break; + case CacheLocationStatus::CLS_WRITING: + *proto_status = proto::meta::CLS_WRITING; + break; + case CacheLocationStatus::CLS_SERVING: + *proto_status = proto::meta::CLS_SERVING; + break; + case CacheLocationStatus::CLS_DELETING: + *proto_status = proto::meta::CLS_DELETING; + break; + case CacheLocationStatus::CLS_NOT_FOUND: + default: + *proto_status = proto::meta::CLS_NOT_FOUND; + break; + } +} + +inline void ProtoConvert::CacheLocationMetaDetailToProto( + const CacheLocationMetaDetail &cache_location_info, + proto::meta::CacheLocationDetail *proto_cache_location) { + proto_cache_location->set_location_id(cache_location_info.location_id); + proto::meta::CacheLocationStatus status; + CacheLocationStatusToProto(cache_location_info.status, &status); + proto_cache_location->set_status(status); + proto::meta::StorageType type; + DataStorageTypeToProto(cache_location_info.type, &type); + proto_cache_location->set_type(type); + proto_cache_location->set_spec_size(cache_location_info.spec_size); + proto_cache_location->set_create_time(cache_location_info.create_time); + LocationSpecsToProto(cache_location_info.location_specs, proto_cache_location->mutable_location_specs()); +} + +inline void ProtoConvert::CacheKeyMetaDetailToProto(const CacheKeyMetaDetail &cache_meta_detail, + proto::meta::CacheMetaDetailItem *proto_cache_meta_detail) { + auto *status = proto_cache_meta_detail->mutable_status(); + if (cache_meta_detail.error_code == EC_OK) { + status->set_code(proto::meta::OK); + } else { + status->set_code(ToPbError(cache_meta_detail.error_code)); + status->set_message("Raw metadata lookup failed with internal error code: " + + std::to_string(static_cast(cache_meta_detail.error_code))); + } + proto_cache_meta_detail->set_request_index(static_cast(cache_meta_detail.request_index)); + proto_cache_meta_detail->set_block_key(cache_meta_detail.block_key); + auto prev_key_iter = cache_meta_detail.properties.find(PROPERTY_PREV_BLOCK_KEY); + if (prev_key_iter != cache_meta_detail.properties.end()) { + proto_cache_meta_detail->set_prev_block_key(prev_key_iter->second); + } + auto *properties = proto_cache_meta_detail->mutable_properties(); + for (const auto &[key, value] : cache_meta_detail.properties) { + (*properties)[key] = value; + } + for (const auto &location : cache_meta_detail.locations) { + CacheLocationMetaDetailToProto(location, proto_cache_meta_detail->add_locations()); + } +} + template std::enable_if_t || std::is_same_v> ProtoConvert::InstanceInfoToProto(const InstanceInfo &instance_info, T *proto_instance_info) {