[Store] feat: Add BatchQueryIp API for querying multiple client IPs (#1162)
* [Store] feat: Add BatchQueryIp API for querying multiple client IPs Add BatchQueryIp API with client test and master server test Signed-off-by: Vincent Gao <vincentbo@linux.alibaba.com> * [Doc] Add API documents for BatchQueryIp Signed-off-by: Vincent Gao <vincentbo@linux.alibaba.com> --------- Signed-off-by: Vincent Gao <vincentbo@linux.alibaba.com>
This commit is contained in:
Vincent-Bo-ali
GitHub
parent
d116df6c4e
commit
72bfa28c04
|
|
@ -112,6 +112,15 @@ tl::expected<void, ErrorCode> Remove(const ObjectKey& key);
|
|||
|
||||
Used to delete the object corresponding to the specified key. This interface marks all data replicas associated with the key in the storage engine as deleted, without needing to communicate with the corresponding storage node (Client).
|
||||
|
||||
### BatchQueryIp
|
||||
|
||||
```C++
|
||||
tl::expected<std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>, ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
```
|
||||
|
||||
Used to batch query the IP addresses for multiple client IDs. For each client ID in the input list, this interface retrieves the unique IP addresses from all segments mounted by that client. The operation is performed on the Master Service and returns a map from client ID to their IP address lists. Only client IDs that have successfully mounted segments are included in the result map. This is useful for discovering the network locations of storage nodes in the cluster.
|
||||
|
||||
### QueryByRegex
|
||||
|
||||
```C++
|
||||
|
|
@ -175,6 +184,9 @@ service MasterService {
|
|||
// Get replica lists for objects matching a regex
|
||||
rpc GetReplicaListByRegex(GetReplicaListByRegexRequest) returns (GetReplicaListByRegexResponse);
|
||||
|
||||
// Batch query IP addresses for multiple client IDs
|
||||
rpc BatchQueryIp(BatchQueryIpRequest) returns (BatchQueryIpResponse);
|
||||
|
||||
// Start Put operation, allocate storage space
|
||||
rpc PutStart(PutStartRequest) returns (PutStartResponse);
|
||||
|
||||
|
|
@ -233,7 +245,28 @@ message GetReplicaListByRegexResponse {
|
|||
- **Response**: GetReplicaListByRegexResponse, which contains a status_code and an object_map. The keys of this map are the successfully matched object keys, and the values are the lists of replica information for each key.
|
||||
- **Description**: Used to query for all keys and their replica information that match the specified regular expression. This interface facilitates bulk queries and management.
|
||||
|
||||
3. PutStart
|
||||
3. BatchQueryIp
|
||||
|
||||
```protobuf
|
||||
message BatchQueryIpRequest {
|
||||
repeated UUID client_ids = 1; // List of client IDs to query
|
||||
};
|
||||
|
||||
message BatchQueryIpResponse {
|
||||
required int32 status_code = 1;
|
||||
map<UUID, IPAddressList> client_ip_map = 2; // Map from client ID to their IP address lists
|
||||
};
|
||||
|
||||
message IPAddressList {
|
||||
repeated string ip_addresses = 1; // List of unique IP addresses
|
||||
};
|
||||
```
|
||||
|
||||
- **Request**: `BatchQueryIpRequest` containing a list of client IDs to query.
|
||||
- **Response**: `BatchQueryIpResponse` containing the status code `status_code` and a `client_ip_map`. The keys of this map are the client IDs that have successfully mounted segments, and the values are lists of unique IP addresses extracted from all segments mounted by each client. Client IDs that have no mounted segments or are not found are silently skipped and not included in the result map.
|
||||
- **Description**: Used to batch query the IP addresses for multiple client IDs. For each client ID in the input list, this interface retrieves the unique IP addresses from all segments mounted by that client.
|
||||
|
||||
4. PutStart
|
||||
|
||||
```protobuf
|
||||
message PutStartRequest {
|
||||
|
|
@ -253,7 +286,7 @@ message PutStartResponse {
|
|||
- **Response**: `PutStartResponse` containing the status code status_code and the allocated replica information replica_list.
|
||||
- **Description**: Before writing an object, the Client must call PutStart to request storage space from the Master Service. The Master Service allocates space based on the config and returns the allocation results (`replica_list`) to the Client. The allocation strategy ensures that each slice of the object is placed in different segments, while operating on a best-effort basis - if insufficient space is available for all requested replicas, as many replicas as possible will be allocated. The Client then writes data to the storage nodes where the allocated replicas are located. The need for both start and end steps ensures that other Clients do not read partially written values, preventing dirty reads.
|
||||
|
||||
4. PutEnd
|
||||
5. PutEnd
|
||||
|
||||
```protobuf
|
||||
message PutEndRequest {
|
||||
|
|
@ -269,7 +302,7 @@ message PutEndResponse {
|
|||
- **Response**: `PutEndResponse` containing the status code status_code.
|
||||
- **Description**: After the Client completes data writing, it calls `PutEnd` to notify the Master Service. The Master Service updates the object's metadata, marking the replica status as `COMPLETE`, indicating that the object is readable.
|
||||
|
||||
5. Remove
|
||||
6. Remove
|
||||
|
||||
```protobuf
|
||||
message RemoveRequest {
|
||||
|
|
@ -285,7 +318,7 @@ message RemoveResponse {
|
|||
- **Response**: `RemoveResponse` containing the status code `status_code`.
|
||||
- **Description**: Used to delete the object and all its replicas corresponding to the specified key. The Master Service marks all replicas of the corresponding object as deleted.
|
||||
|
||||
6. RemoveByRegex
|
||||
7. RemoveByRegex
|
||||
|
||||
```protobuf
|
||||
message RemoveByRegexRequest {
|
||||
|
|
@ -302,7 +335,7 @@ message RemoveByRegexResponse {
|
|||
- **Response**: RemoveByRegexResponse, which contains a status_code and the number of objects that were removed, removed_count.
|
||||
- **Description**: Used to delete all objects and their corresponding replicas for keys that match the specified regular expression. Similar to the Remove interface, this is a metadata operation where the Master Service marks the status of all matched object replicas as removed.
|
||||
|
||||
7. MountSegment
|
||||
8. MountSegment
|
||||
|
||||
```protobuf
|
||||
message MountSegmentRequest {
|
||||
|
|
@ -318,7 +351,7 @@ message MountSegmentResponse {
|
|||
|
||||
The storage node (Client) allocates a segment of memory and, after calling `TransferEngine::registerLocalMemory` to complete local mounting, calls this interface to mount the allocated continuous address space to the Master Service for allocation.
|
||||
|
||||
8. UnmountSegment
|
||||
9. UnmountSegment
|
||||
|
||||
```protobuf
|
||||
message UnmountSegmentRequest {
|
||||
|
|
|
|||
|
|
@ -124,6 +124,15 @@ tl::expected<long, ErrorCode> RemoveByRegex(const ObjectKey& str);
|
|||
|
||||
用于删除与正则表达式匹配的所有 key 对应的对象。其余能力类似 Remove。
|
||||
|
||||
### BatchQueryIp
|
||||
|
||||
```C++
|
||||
tl::expected<std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>, ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
```
|
||||
|
||||
用于批量查询多个客户端 ID 的 IP 地址。对于输入列表中的每个客户端 ID，此接口会检索该客户端挂载的所有网段中的唯一 IP 地址。此操作在主服务器上执行，并返回一个从客户端 ID 到其 IP 地址列表的映射。
|
||||
|
||||
### QueryByRegex
|
||||
|
||||
```C++
|
||||
|
|
@ -177,6 +186,9 @@ service MasterService {
|
|||
// 获取与正则表达式匹配的对性的副本列表
|
||||
rpc GetReplicaListByRegex(GetReplicaListByRegexRequest) returns (GetReplicaListByRegexResponse);
|
||||
|
||||
// 批量查询多个客户端 ID 的 IP 地址
|
||||
rpc BatchQueryIp(BatchQueryIpRequest) returns (BatchQueryIpResponse);
|
||||
|
||||
// 开始 Put 操作，分配存储空间
|
||||
rpc PutStart(PutStartRequest) returns (PutStartResponse);
|
||||
|
||||
|
|
@ -237,7 +249,29 @@ message GetReplicaListByRegexResponse {
|
|||
|
||||
说明: 用于查询与指定正则表达式匹配的所有 key 及其副本信息。该接口方便进行批量查询和管理。
|
||||
|
||||
3. PutStart
|
||||
3. BatchQueryIp
|
||||
|
||||
```protobuf
|
||||
message BatchQueryIpRequest {
|
||||
repeated UUID client_ids = 1; // 客户端ID列表
|
||||
};
|
||||
|
||||
message BatchQueryIpResponse {
|
||||
required int32 status_code = 1;
|
||||
map<UUID, IPAddressList> client_ip_map = 2; // 从客户端 ID 到其 IP 地址列表的映射
|
||||
};
|
||||
|
||||
message IPAddressList {
|
||||
repeated string ip_addresses = 1; // 唯一 IP 地址列表
|
||||
};
|
||||
```
|
||||
|
||||
* 请求: BatchQueryIpRequest 包含要查询的客户端 ID 列表。
|
||||
* 响应: BatchQueryIpResponse 包含状态码 status_code 和 client_ip_map。该映射的键是已成功挂载网段的客户端 ID，值是从每个客户端挂载的所有网段中提取的唯一 IP 地址列表。未挂载网段或未找到的客户端 ID 将被静默跳过，不包含在结果映射中。
|
||||
|
||||
说明: 用于批量查询多个客户端 ID 的 IP 地址。对于输入列表中的每个客户端 ID，此接口会从该客户端挂载的所有网段中检索唯一的 IP 地址。
|
||||
|
||||
4. PutStart
|
||||
|
||||
```protobuf
|
||||
message PutStartRequest {
|
||||
|
|
@ -258,7 +292,7 @@ message PutStartResponse {
|
|||
|
||||
说明: Client 在写入对象前，需要先调用 PutStart 向 `Master Service` 申请存储空间。`Master Service` 会根据 config 分配空间，并将分配结果（replica_list）返回给 Client。分配策略确保对象的每个slice被放置在不同的segment中，同时采用尽力而为的方式运行——如果没有足够的空间来分配所有请求的副本，将分配尽可能多的副本。Client 随后将数据写入到分配副本所在的存储节点。 之所以需要 start 和 end 两步，是为确保其他Client不会读到正在写的值，进而造成脏读。
|
||||
|
||||
4. PutEnd
|
||||
5. PutEnd
|
||||
|
||||
```protobuf
|
||||
message PutEndRequest {
|
||||
|
|
@ -275,7 +309,7 @@ message PutEndResponse {
|
|||
|
||||
Client 完成数据写入后，调用 PutEnd 通知 `Master Service`。`Master Service` 将更新对象的元数据信息，将副本状态标记为 COMPLETE，表示该对象可以被读取。
|
||||
|
||||
5. Remove
|
||||
6. Remove
|
||||
|
||||
```protobuf
|
||||
message RemoveRequest {
|
||||
|
|
@ -292,7 +326,7 @@ message RemoveResponse {
|
|||
|
||||
用于删除指定 key 对应的对象及其所有副本。Master Service 将对应对象的所有副本状态标记为删除。
|
||||
|
||||
6. RemoveByRegex
|
||||
7. RemoveByRegex
|
||||
|
||||
```protobuf
|
||||
message RemoveByRegexRequest {
|
||||
|
|
@ -310,7 +344,7 @@ message RemoveByRegexResponse {
|
|||
|
||||
说明: 用于删除与指定正则表达式匹配的所有对象及其全部副本。与 Remove 接口类似，这是一个元数据操作，Master Service 将所有匹配对象的副本状态标记为删除。
|
||||
|
||||
7. MountSegment
|
||||
8. MountSegment
|
||||
|
||||
```protobuf
|
||||
message MountSegmentRequest {
|
||||
|
|
@ -326,7 +360,7 @@ message MountSegmentResponse {
|
|||
|
||||
存储节点(Client)自己分配一段内存，然后在调用`TransferEngine::registerLoalMemory` 完成本地挂载后，调用该接口，将分配好的一段连续的地址空间挂载到`Master Service`用于分配。
|
||||
|
||||
8. UnmountSegment
|
||||
9. UnmountSegment
|
||||
|
||||
```protobuf
|
||||
message UnmountSegmentRequest {
|
||||
|
|
|
|||
|
|
@ -112,6 +112,15 @@ tl::expected<void, ErrorCode> Remove(const ObjectKey& key);
|
|||
|
||||
Used to delete the object corresponding to the specified key. This interface marks all data replicas associated with the key in the storage engine as deleted, without needing to communicate with the corresponding storage node (Client).
|
||||
|
||||
### BatchQueryIp
|
||||
|
||||
```C++
|
||||
tl::expected<std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>, ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
```
|
||||
|
||||
Used to batch query the IP addresses for multiple client IDs. For each client ID in the input list, this interface retrieves the unique IP addresses from all segments mounted by that client. The operation is performed on the Master Service and returns a map from client ID to their IP address lists. Only client IDs that have successfully mounted segments are included in the result map. This is useful for discovering the network locations of storage nodes in the cluster.
|
||||
|
||||
### QueryByRegex
|
||||
|
||||
```C++
|
||||
|
|
@ -175,6 +184,9 @@ service MasterService {
|
|||
// Get replica lists for objects matching a regex
|
||||
rpc GetReplicaListByRegex(GetReplicaListByRegexRequest) returns (GetReplicaListByRegexResponse);
|
||||
|
||||
// Batch query IP addresses for multiple client IDs
|
||||
rpc BatchQueryIp(BatchQueryIpRequest) returns (BatchQueryIpResponse);
|
||||
|
||||
// Start Put operation, allocate storage space
|
||||
rpc PutStart(PutStartRequest) returns (PutStartResponse);
|
||||
|
||||
|
|
@ -233,7 +245,28 @@ message GetReplicaListByRegexResponse {
|
|||
- **Response**: GetReplicaListByRegexResponse, which contains a status_code and an object_map. The keys of this map are the successfully matched object keys, and the values are the lists of replica information for each key.
|
||||
- **Description**: Used to query for all keys and their replica information that match the specified regular expression. This interface facilitates bulk queries and management.
|
||||
|
||||
3. PutStart
|
||||
3. BatchQueryIp
|
||||
|
||||
```protobuf
|
||||
message BatchQueryIpRequest {
|
||||
repeated UUID client_ids = 1; // List of client IDs to query
|
||||
};
|
||||
|
||||
message BatchQueryIpResponse {
|
||||
required int32 status_code = 1;
|
||||
map<UUID, IPAddressList> client_ip_map = 2; // Map from client ID to their IP address lists
|
||||
};
|
||||
|
||||
message IPAddressList {
|
||||
repeated string ip_addresses = 1; // List of unique IP addresses
|
||||
};
|
||||
```
|
||||
|
||||
- **Request**: `BatchQueryIpRequest` containing a list of client IDs to query.
|
||||
- **Response**: `BatchQueryIpResponse` containing the status code `status_code` and a `client_ip_map`. The keys of this map are the client IDs that have successfully mounted segments, and the values are lists of unique IP addresses extracted from all segments mounted by each client. Client IDs that have no mounted segments or are not found are silently skipped and not included in the result map.
|
||||
- **Description**: Used to batch query the IP addresses for multiple client IDs. For each client ID in the input list, this interface retrieves the unique IP addresses from all segments mounted by that client.
|
||||
|
||||
4. PutStart
|
||||
|
||||
```protobuf
|
||||
message PutStartRequest {
|
||||
|
|
@ -253,7 +286,7 @@ message PutStartResponse {
|
|||
- **Response**: `PutStartResponse` containing the status code status_code and the allocated replica information replica_list.
|
||||
- **Description**: Before writing an object, the Client must call PutStart to request storage space from the Master Service. The Master Service allocates space based on the config and returns the allocation results (`replica_list`) to the Client. The allocation strategy ensures that each slice of the object is placed in different segments, while operating on a best-effort basis - if insufficient space is available for all requested replicas, as many replicas as possible will be allocated. The Client then writes data to the storage nodes where the allocated replicas are located. The need for both start and end steps ensures that other Clients do not read partially written values, preventing dirty reads.
|
||||
|
||||
4. PutEnd
|
||||
5. PutEnd
|
||||
|
||||
```protobuf
|
||||
message PutEndRequest {
|
||||
|
|
@ -269,7 +302,7 @@ message PutEndResponse {
|
|||
- **Response**: `PutEndResponse` containing the status code status_code.
|
||||
- **Description**: After the Client completes data writing, it calls `PutEnd` to notify the Master Service. The Master Service updates the object's metadata, marking the replica status as `COMPLETE`, indicating that the object is readable.
|
||||
|
||||
5. Remove
|
||||
6. Remove
|
||||
|
||||
```protobuf
|
||||
message RemoveRequest {
|
||||
|
|
@ -285,7 +318,7 @@ message RemoveResponse {
|
|||
- **Response**: `RemoveResponse` containing the status code `status_code`.
|
||||
- **Description**: Used to delete the object and all its replicas corresponding to the specified key. The Master Service marks all replicas of the corresponding object as deleted.
|
||||
|
||||
6. RemoveByRegex
|
||||
7. RemoveByRegex
|
||||
|
||||
```protobuf
|
||||
message RemoveByRegexRequest {
|
||||
|
|
@ -302,7 +335,7 @@ message RemoveByRegexResponse {
|
|||
- **Response**: RemoveByRegexResponse, which contains a status_code and the number of objects that were removed, removed_count.
|
||||
- **Description**: Used to delete all objects and their corresponding replicas for keys that match the specified regular expression. Similar to the Remove interface, this is a metadata operation where the Master Service marks the status of all matched object replicas as removed.
|
||||
|
||||
7. MountSegment
|
||||
8. MountSegment
|
||||
|
||||
```protobuf
|
||||
message MountSegmentRequest {
|
||||
|
|
@ -318,7 +351,7 @@ message MountSegmentResponse {
|
|||
|
||||
The storage node (Client) allocates a segment of memory and, after calling `TransferEngine::registerLocalMemory` to complete local mounting, calls this interface to mount the allocated continuous address space to the Master Service for allocation.
|
||||
|
||||
8. UnmountSegment
|
||||
9. UnmountSegment
|
||||
|
||||
```protobuf
|
||||
message UnmountSegmentRequest {
|
||||
|
|
|
|||
|
|
@ -97,6 +97,17 @@ class Client {
|
|||
const std::vector<std::string>& object_keys,
|
||||
std::unordered_map<std::string, std::vector<Slice>>& slices);
|
||||
|
||||
/**
|
||||
* @brief Batch query IP addresses for multiple client IDs.
|
||||
* @param client_ids Vector of client UUIDs to query.
|
||||
* @return An expected object containing a map from client_id to their IP
|
||||
* address lists on success, or an ErrorCode on failure.
|
||||
*/
|
||||
tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
|
||||
/**
|
||||
* @brief Gets object metadata without transferring data
|
||||
* @param object_key Key to query
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
#include <string>
|
||||
#include <vector>
|
||||
#include <cstdlib>
|
||||
#include <boost/functional/hash.hpp>
|
||||
#include <ylt/coro_rpc/coro_rpc_client.hpp>
|
||||
#include <ylt/coro_io/client_pool.hpp>
|
||||
|
||||
|
|
@ -72,6 +73,17 @@ class MasterClient {
|
|||
[[nodiscard]] tl::expected<MasterMetricManager::CacheHitStatDict, ErrorCode>
|
||||
CalcCacheStats();
|
||||
|
||||
/**
|
||||
* @brief Batch query IP addresses for multiple client IDs.
|
||||
* @param client_ids Vector of client UUIDs to query.
|
||||
* @return An expected object containing a map from client_id to their IP
|
||||
* address lists on success, or an ErrorCode on failure.
|
||||
*/
|
||||
[[nodiscard]] tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
|
||||
/**
|
||||
* @brief Gets object metadata without transferring data
|
||||
* @param object_key Key to query
|
||||
|
|
|
|||
|
|
@ -118,6 +118,9 @@ class MasterMetricManager {
|
|||
void inc_batch_exist_key_requests(int64_t items);
|
||||
void inc_batch_exist_key_failures(int64_t failed_items);
|
||||
void inc_batch_exist_key_partial_success(int64_t failed_items);
|
||||
void inc_batch_query_ip_requests(int64_t items);
|
||||
void inc_batch_query_ip_failures(int64_t failed_items);
|
||||
void inc_batch_query_ip_partial_success(int64_t failed_items);
|
||||
void inc_batch_get_replica_list_requests(int64_t items);
|
||||
void inc_batch_get_replica_list_failures(int64_t failed_items);
|
||||
void inc_batch_get_replica_list_partial_success(int64_t failed_items);
|
||||
|
|
@ -165,6 +168,11 @@ class MasterMetricManager {
|
|||
int64_t get_batch_exist_key_partial_successes();
|
||||
int64_t get_batch_exist_key_items();
|
||||
int64_t get_batch_exist_key_failed_items();
|
||||
int64_t get_batch_query_ip_requests();
|
||||
int64_t get_batch_query_ip_failures();
|
||||
int64_t get_batch_query_ip_partial_successes();
|
||||
int64_t get_batch_query_ip_items();
|
||||
int64_t get_batch_query_ip_failed_items();
|
||||
int64_t get_batch_get_replica_list_requests();
|
||||
int64_t get_batch_get_replica_list_failures();
|
||||
int64_t get_batch_get_replica_list_partial_successes();
|
||||
|
|
@ -286,6 +294,11 @@ class MasterMetricManager {
|
|||
ylt::metric::counter_t batch_exist_key_partial_successes_;
|
||||
ylt::metric::counter_t batch_exist_key_items_;
|
||||
ylt::metric::counter_t batch_exist_key_failed_items_;
|
||||
ylt::metric::counter_t batch_query_ip_requests_;
|
||||
ylt::metric::counter_t batch_query_ip_failures_;
|
||||
ylt::metric::counter_t batch_query_ip_partial_successes_;
|
||||
ylt::metric::counter_t batch_query_ip_items_;
|
||||
ylt::metric::counter_t batch_query_ip_failed_items_;
|
||||
ylt::metric::counter_t batch_get_replica_list_requests_;
|
||||
ylt::metric::counter_t batch_get_replica_list_failures_;
|
||||
ylt::metric::counter_t batch_get_replica_list_partial_successes_;
|
||||
|
|
|
|||
|
|
@ -110,6 +110,28 @@ class MasterService {
|
|||
auto QuerySegments(const std::string& segment)
|
||||
-> tl::expected<std::pair<size_t, size_t>, ErrorCode>;
|
||||
|
||||
/**
|
||||
* @brief Query IP addresses for a given client ID.
|
||||
* @param client_id The UUID of the client to query.
|
||||
* @return An expected object containing a vector of IP addresses on success
|
||||
* (empty vector if client has no IPs), or ErrorCode::CLIENT_NOT_FOUND if
|
||||
* the client doesn't exist, or another ErrorCode on other failures.
|
||||
*/
|
||||
auto QueryIp(const UUID& client_id)
|
||||
-> tl::expected<std::vector<std::string>, ErrorCode>;
|
||||
|
||||
/**
|
||||
* @brief Batch query IP addresses for multiple client IDs.
|
||||
* @param client_ids Vector of client UUIDs to query.
|
||||
* @return An expected object containing a map from client_id to their IP
|
||||
* address lists on success, or an ErrorCode on failure. Non-existent
|
||||
* clients are omitted from the result map. Clients that exist but have no
|
||||
* IPs are included with empty vectors.
|
||||
*/
|
||||
auto BatchQueryIp(const std::vector<UUID>& client_ids) -> tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>;
|
||||
|
||||
/**
|
||||
* @brief Retrieves replica lists for object keys that match a regex
|
||||
* pattern.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
#pragma once
|
||||
|
||||
#include <atomic>
|
||||
#include <boost/functional/hash.hpp>
|
||||
#include <cstdint>
|
||||
#include <thread>
|
||||
#include <ylt/coro_http/coro_http_server.hpp>
|
||||
|
|
@ -32,6 +33,11 @@ class WrappedMasterService {
|
|||
std::vector<tl::expected<bool, ErrorCode>> BatchExistKey(
|
||||
const std::vector<std::string>& keys);
|
||||
|
||||
tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
BatchQueryIp(const std::vector<UUID>& client_ids);
|
||||
|
||||
tl::expected<
|
||||
std::unordered_map<std::string, std::vector<Replica::Descriptor>>,
|
||||
ErrorCode>
|
||||
|
|
|
|||
|
|
@ -100,6 +100,7 @@ enum class ErrorCode : int32_t {
|
|||
SHARD_INDEX_OUT_OF_RANGE = -100, ///< Shard index is out of bounds.
|
||||
SEGMENT_NOT_FOUND = -101, ///< No available segments found.
|
||||
SEGMENT_ALREADY_EXISTS = -102, ///< Segment already exists.
|
||||
CLIENT_NOT_FOUND = -103, ///< Client not found.
|
||||
|
||||
// Handle selection errors (Range: -200 to -299)
|
||||
NO_AVAILABLE_HANDLE =
|
||||
|
|
|
|||
|
|
@ -74,6 +74,11 @@ void to_stream(std::ostream& os, const std::vector<T>& vec) {
|
|||
|
||||
template <typename K, typename V>
|
||||
void to_stream(std::ostream& os, const std::unordered_map<K, V>& map) {
|
||||
to_stream<K, V, std::hash<K>>(os, map);
|
||||
}
|
||||
|
||||
template <typename K, typename V, typename H>
|
||||
void to_stream(std::ostream& os, const std::unordered_map<K, V, H>& map) {
|
||||
os << "{";
|
||||
auto it = map.begin();
|
||||
while (it != map.end()) {
|
||||
|
|
|
|||
|
|
@ -501,6 +501,14 @@ std::vector<tl::expected<void, ErrorCode>> Client::BatchGet(
|
|||
return results;
|
||||
}
|
||||
|
||||
tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
Client::BatchQueryIp(const std::vector<UUID>& client_ids) {
|
||||
auto result = master_client_.BatchQueryIp(client_ids);
|
||||
return result;
|
||||
}
|
||||
|
||||
tl::expected<std::unordered_map<std::string, std::vector<Replica::Descriptor>>,
|
||||
ErrorCode>
|
||||
Client::QueryByRegex(const std::string& str) {
|
||||
|
|
|
|||
|
|
@ -41,6 +41,11 @@ struct RpcNameTraits<&WrappedMasterService::CalcCacheStats> {
|
|||
static constexpr const char* value = "CalcCacheStats";
|
||||
};
|
||||
|
||||
template <>
|
||||
struct RpcNameTraits<&WrappedMasterService::BatchQueryIp> {
|
||||
static constexpr const char* value = "BatchQueryIp";
|
||||
};
|
||||
|
||||
template <>
|
||||
struct RpcNameTraits<&WrappedMasterService::GetReplicaListByRegex> {
|
||||
static constexpr const char* value = "GetReplicaListByRegex";
|
||||
|
|
@ -281,6 +286,22 @@ MasterClient::CalcCacheStats() {
|
|||
MasterMetricManager::CacheHitStatDict>();
|
||||
}
|
||||
|
||||
tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
MasterClient::BatchQueryIp(const std::vector<UUID>& client_ids) {
|
||||
ScopedVLogTimer timer(1, "MasterClient::BatchQueryIp");
|
||||
timer.LogRequest("client_ids_count=", client_ids.size());
|
||||
|
||||
auto result = invoke_rpc<
|
||||
&WrappedMasterService::BatchQueryIp,
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>>(
|
||||
client_ids);
|
||||
|
||||
timer.LogResponseExpected(result);
|
||||
return result;
|
||||
}
|
||||
|
||||
tl::expected<std::unordered_map<std::string, std::vector<Replica::Descriptor>>,
|
||||
ErrorCode>
|
||||
MasterClient::GetReplicaListByRegex(const std::string& str) {
|
||||
|
|
|
|||
|
|
@ -130,6 +130,20 @@ MasterMetricManager::MasterMetricManager()
|
|||
batch_exist_key_failed_items_(
|
||||
"master_batch_exist_key_failed_items_total",
|
||||
"Total number of failed items in BatchExistKey requests"),
|
||||
batch_query_ip_requests_(
|
||||
"master_batch_query_ip_requests_total",
|
||||
"Total number of BatchQueryIp requests received"),
|
||||
batch_query_ip_failures_("master_batch_query_ip_failures_total",
|
||||
"Total number of failed BatchQueryIp requests"),
|
||||
batch_query_ip_partial_successes_(
|
||||
"master_batch_query_ip_partial_successes_total",
|
||||
"Total number of partially successful BatchQueryIp requests"),
|
||||
batch_query_ip_items_(
|
||||
"master_batch_query_ip_items_total",
|
||||
"Total number of items processed in BatchQueryIp requests"),
|
||||
batch_query_ip_failed_items_(
|
||||
"master_batch_query_ip_failed_items_total",
|
||||
"Total number of failed items in BatchQueryIp requests"),
|
||||
batch_get_replica_list_requests_(
|
||||
"master_batch_get_replica_list_requests_total",
|
||||
"Total number of BatchGetReplicaList requests received"),
|
||||
|
|
@ -274,6 +288,11 @@ void MasterMetricManager::update_metrics_for_zero_output() {
|
|||
batch_exist_key_partial_successes_.inc(0);
|
||||
batch_exist_key_items_.inc(0);
|
||||
batch_exist_key_failed_items_.inc(0);
|
||||
batch_query_ip_requests_.inc(0);
|
||||
batch_query_ip_failures_.inc(0);
|
||||
batch_query_ip_partial_successes_.inc(0);
|
||||
batch_query_ip_items_.inc(0);
|
||||
batch_query_ip_failed_items_.inc(0);
|
||||
batch_get_replica_list_requests_.inc(0);
|
||||
batch_get_replica_list_failures_.inc(0);
|
||||
batch_get_replica_list_partial_successes_.inc(0);
|
||||
|
|
@ -576,6 +595,19 @@ void MasterMetricManager::inc_batch_exist_key_partial_success(
|
|||
batch_exist_key_partial_successes_.inc(1);
|
||||
batch_exist_key_failed_items_.inc(failed_items);
|
||||
}
|
||||
void MasterMetricManager::inc_batch_query_ip_requests(int64_t items) {
|
||||
batch_query_ip_requests_.inc(1);
|
||||
batch_query_ip_items_.inc(items);
|
||||
}
|
||||
void MasterMetricManager::inc_batch_query_ip_failures(int64_t failed_items) {
|
||||
batch_query_ip_failures_.inc(1);
|
||||
batch_query_ip_failed_items_.inc(failed_items);
|
||||
}
|
||||
void MasterMetricManager::inc_batch_query_ip_partial_success(
|
||||
int64_t failed_items) {
|
||||
batch_query_ip_partial_successes_.inc(1);
|
||||
batch_query_ip_failed_items_.inc(failed_items);
|
||||
}
|
||||
void MasterMetricManager::inc_batch_get_replica_list_requests(int64_t items) {
|
||||
batch_get_replica_list_requests_.inc(1);
|
||||
batch_get_replica_list_items_.inc(items);
|
||||
|
|
@ -767,6 +799,26 @@ int64_t MasterMetricManager::get_batch_exist_key_failed_items() {
|
|||
return batch_exist_key_failed_items_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_query_ip_requests() {
|
||||
return batch_query_ip_requests_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_query_ip_failures() {
|
||||
return batch_query_ip_failures_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_query_ip_partial_successes() {
|
||||
return batch_query_ip_partial_successes_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_query_ip_items() {
|
||||
return batch_query_ip_items_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_query_ip_failed_items() {
|
||||
return batch_query_ip_failed_items_.value();
|
||||
}
|
||||
|
||||
int64_t MasterMetricManager::get_batch_get_replica_list_requests() {
|
||||
return batch_get_replica_list_requests_.value();
|
||||
}
|
||||
|
|
@ -946,6 +998,8 @@ std::string MasterMetricManager::serialize_metrics() {
|
|||
// Serialize Batch Request Counters
|
||||
serialize_metric(batch_exist_key_requests_);
|
||||
serialize_metric(batch_exist_key_failures_);
|
||||
serialize_metric(batch_query_ip_requests_);
|
||||
serialize_metric(batch_query_ip_failures_);
|
||||
serialize_metric(batch_get_replica_list_requests_);
|
||||
serialize_metric(batch_get_replica_list_failures_);
|
||||
serialize_metric(batch_put_start_requests_);
|
||||
|
|
@ -1099,6 +1153,12 @@ std::string MasterMetricManager::get_summary_string() {
|
|||
int64_t batch_exist_key_items = batch_exist_key_items_.value();
|
||||
int64_t batch_exist_key_failed_items =
|
||||
batch_exist_key_failed_items_.value();
|
||||
int64_t batch_query_ip_requests = batch_query_ip_requests_.value();
|
||||
int64_t batch_query_ip_fails = batch_query_ip_failures_.value();
|
||||
int64_t batch_query_ip_partial_successes =
|
||||
batch_query_ip_partial_successes_.value();
|
||||
int64_t batch_query_ip_items = batch_query_ip_items_.value();
|
||||
int64_t batch_query_ip_failed_items = batch_query_ip_failed_items_.value();
|
||||
|
||||
// Eviction counters
|
||||
int64_t eviction_success = eviction_success_.value();
|
||||
|
|
@ -1181,6 +1241,13 @@ std::string MasterMetricManager::get_summary_string() {
|
|||
<< batch_exist_key_requests
|
||||
<< ", Item=" << batch_exist_key_items - batch_exist_key_failed_items
|
||||
<< "/" << batch_exist_key_items << "), ";
|
||||
ss << "QueryIp:(Req="
|
||||
<< batch_query_ip_requests - batch_query_ip_fails -
|
||||
batch_query_ip_partial_successes
|
||||
<< "/" << batch_query_ip_partial_successes << "/"
|
||||
<< batch_query_ip_requests
|
||||
<< ", Item=" << batch_query_ip_items - batch_query_ip_failed_items << "/"
|
||||
<< batch_query_ip_items << "), ";
|
||||
|
||||
// Eviction summary
|
||||
ss << " | Eviction: " << "Success/Attempts=" << eviction_success << "/"
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
#include <cstdint>
|
||||
#include <shared_mutex>
|
||||
#include <regex>
|
||||
#include <unordered_set>
|
||||
#include <ylt/util/tl/expected.hpp>
|
||||
|
||||
#include "master_metric_manager.h"
|
||||
|
|
@ -280,6 +281,63 @@ auto MasterService::QuerySegments(const std::string& segment)
|
|||
return std::make_pair(used, capacity);
|
||||
}
|
||||
|
||||
auto MasterService::QueryIp(const UUID& client_id)
|
||||
-> tl::expected<std::vector<std::string>, ErrorCode> {
|
||||
ScopedSegmentAccess segment_access = segment_manager_.getSegmentAccess();
|
||||
std::vector<Segment> segments;
|
||||
ErrorCode err = segment_access.GetClientSegments(client_id, segments);
|
||||
if (err != ErrorCode::OK) {
|
||||
if (err == ErrorCode::SEGMENT_NOT_FOUND) {
|
||||
VLOG(1) << "QueryIp: client_id=" << client_id
|
||||
<< " not found or has no segments";
|
||||
return tl::make_unexpected(ErrorCode::CLIENT_NOT_FOUND);
|
||||
}
|
||||
|
||||
LOG(ERROR) << "QueryIp: failed to get segments for client_id="
|
||||
<< client_id << ", error=" << toString(err);
|
||||
|
||||
return tl::make_unexpected(err);
|
||||
}
|
||||
|
||||
std::unordered_set<std::string> unique_ips;
|
||||
unique_ips.reserve(segments.size());
|
||||
for (const auto& segment : segments) {
|
||||
if (!segment.te_endpoint.empty()) {
|
||||
size_t colon_pos = segment.te_endpoint.find(':');
|
||||
if (colon_pos != std::string::npos) {
|
||||
std::string ip = segment.te_endpoint.substr(0, colon_pos);
|
||||
unique_ips.emplace(ip);
|
||||
} else {
|
||||
unique_ips.emplace(segment.te_endpoint);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (unique_ips.empty()) {
|
||||
LOG(WARNING) << "QueryIp: client_id=" << client_id
|
||||
<< " has no valid IP addresses";
|
||||
return {};
|
||||
}
|
||||
std::vector<std::string> result(unique_ips.begin(), unique_ips.end());
|
||||
return result;
|
||||
}
|
||||
|
||||
auto MasterService::BatchQueryIp(const std::vector<UUID>& client_ids)
|
||||
-> tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode> {
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>
|
||||
results;
|
||||
results.reserve(client_ids.size());
|
||||
for (const auto& client_id : client_ids) {
|
||||
auto ip_result = QueryIp(client_id);
|
||||
if (ip_result.has_value()) {
|
||||
results.emplace(client_id, std::move(ip_result.value()));
|
||||
}
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
auto MasterService::GetReplicaListByRegex(const std::string& regex_pattern)
|
||||
-> tl::expected<
|
||||
std::unordered_map<std::string, std::vector<Replica::Descriptor>>,
|
||||
|
|
|
|||
|
|
@ -218,6 +218,46 @@ std::vector<tl::expected<bool, ErrorCode>> WrappedMasterService::BatchExistKey(
|
|||
return result;
|
||||
}
|
||||
|
||||
tl::expected<
|
||||
std::unordered_map<UUID, std::vector<std::string>, boost::hash<UUID>>,
|
||||
ErrorCode>
|
||||
WrappedMasterService::BatchQueryIp(const std::vector<UUID>& client_ids) {
|
||||
ScopedVLogTimer timer(1, "BatchQueryIp");
|
||||
const size_t total_client_ids = client_ids.size();
|
||||
timer.LogRequest("client_ids_count=", total_client_ids);
|
||||
MasterMetricManager::instance().inc_batch_query_ip_requests(
|
||||
total_client_ids);
|
||||
|
||||
auto result = master_service_.BatchQueryIp(client_ids);
|
||||
|
||||
size_t failure_count = 0;
|
||||
if (!result.has_value()) {
|
||||
failure_count = total_client_ids;
|
||||
} else {
|
||||
for (size_t i = 0; i < client_ids.size(); ++i) {
|
||||
const auto& client_id = client_ids[i];
|
||||
if (result.value().find(client_id) == result.value().end()) {
|
||||
failure_count++;
|
||||
VLOG(1) << "BatchQueryIp failed for client_id[" << i << "] '"
|
||||
<< client_id << "': not found in results";
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (failure_count == total_client_ids) {
|
||||
MasterMetricManager::instance().inc_batch_query_ip_failures(
|
||||
failure_count);
|
||||
} else if (failure_count != 0) {
|
||||
MasterMetricManager::instance().inc_batch_query_ip_partial_success(
|
||||
failure_count);
|
||||
}
|
||||
|
||||
timer.LogResponse("total=", total_client_ids,
|
||||
", success=", total_client_ids - failure_count,
|
||||
", failures=", failure_count);
|
||||
return result;
|
||||
}
|
||||
|
||||
tl::expected<std::unordered_map<std::string, std::vector<Replica::Descriptor>>,
|
||||
ErrorCode>
|
||||
WrappedMasterService::GetReplicaListByRegex(const std::string& str) {
|
||||
|
|
@ -604,6 +644,8 @@ void RegisterRpcService(
|
|||
mooncake::WrappedMasterService& wrapped_master_service) {
|
||||
server.register_handler<&mooncake::WrappedMasterService::ExistKey>(
|
||||
&wrapped_master_service);
|
||||
server.register_handler<&mooncake::WrappedMasterService::BatchQueryIp>(
|
||||
&wrapped_master_service);
|
||||
server.register_handler<
|
||||
&mooncake::WrappedMasterService::GetReplicaListByRegex>(
|
||||
&wrapped_master_service);
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ const std::string& toString(ErrorCode errorCode) noexcept {
|
|||
{ErrorCode::SHARD_INDEX_OUT_OF_RANGE, "SHARD_INDEX_OUT_OF_RANGE"},
|
||||
{ErrorCode::SEGMENT_NOT_FOUND, "SEGMENT_NOT_FOUND"},
|
||||
{ErrorCode::SEGMENT_ALREADY_EXISTS, "SEGMENT_ALREADY_EXISTS"},
|
||||
{ErrorCode::CLIENT_NOT_FOUND, "CLIENT_NOT_FOUND"},
|
||||
{ErrorCode::NO_AVAILABLE_HANDLE, "NO_AVAILABLE_HANDLE"},
|
||||
{ErrorCode::INVALID_VERSION, "INVALID_VERSION"},
|
||||
{ErrorCode::INVALID_KEY, "INVALID_KEY"},
|
||||
|
|
|
|||
|
|
@ -6,6 +6,11 @@
|
|||
#include <memory>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
#include <regex>
|
||||
#include <unordered_set>
|
||||
#include <unordered_map>
|
||||
#include <thread>
|
||||
#include <chrono>
|
||||
|
||||
#include "allocator.h"
|
||||
#include "client_service.h"
|
||||
|
|
@ -22,6 +27,58 @@ DEFINE_uint64(default_kv_lease_ttl, mooncake::DEFAULT_DEFAULT_KV_LEASE_TTL,
|
|||
namespace mooncake {
|
||||
namespace testing {
|
||||
|
||||
// Helper functions for client_id parsing
|
||||
std::string FormatClientId(const UUID& client_id) {
|
||||
return std::to_string(client_id.first) + "-" +
|
||||
std::to_string(client_id.second);
|
||||
}
|
||||
|
||||
UUID ParseClientId(const std::string& client_id_str) {
|
||||
UUID client_id{0, 0};
|
||||
size_t dash_pos = client_id_str.find('-');
|
||||
if (dash_pos != std::string::npos) {
|
||||
try {
|
||||
client_id.first = std::stoull(client_id_str.substr(0, dash_pos));
|
||||
client_id.second = std::stoull(client_id_str.substr(dash_pos + 1));
|
||||
} catch (const std::exception& e) {
|
||||
LOG(ERROR) << "Failed to parse client_id: " << e.what();
|
||||
}
|
||||
} else {
|
||||
LOG(ERROR) << "Invalid client_id format. Expected format: first-second";
|
||||
}
|
||||
return client_id;
|
||||
}
|
||||
|
||||
class ClientIdCaptureSink : public google::LogSink {
|
||||
public:
|
||||
std::string captured_client_id;
|
||||
|
||||
void send(google::LogSeverity severity, const char* full_filename,
|
||||
const char* base_filename, int line, const struct ::tm* tm_time,
|
||||
const char* message, size_t message_len) override {
|
||||
(void)severity;
|
||||
(void)full_filename;
|
||||
(void)base_filename;
|
||||
(void)line;
|
||||
(void)tm_time;
|
||||
|
||||
std::string msg(message, message_len);
|
||||
|
||||
size_t pos = msg.find("client_id=");
|
||||
if (pos != std::string::npos) {
|
||||
std::string client_id_str = msg.substr(pos + 10);
|
||||
client_id_str.erase(0, client_id_str.find_first_not_of(" \t\n\r"));
|
||||
client_id_str.erase(client_id_str.find_last_not_of(" \t\n\r") + 1);
|
||||
|
||||
std::regex uuid_pattern(R"((\d+)-(\d+))");
|
||||
std::smatch match;
|
||||
if (std::regex_search(client_id_str, match, uuid_pattern)) {
|
||||
captured_client_id = match[0].str();
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
class ClientIntegrationTest : public ::testing::Test {
|
||||
protected:
|
||||
static std::shared_ptr<Client> CreateClient(const std::string& host_name) {
|
||||
|
|
@ -94,13 +151,51 @@ class ClientIntegrationTest : public ::testing::Test {
|
|||
|
||||
static void InitializeClients() {
|
||||
// This client is used for testing purposes.
|
||||
// Capture test_client_ client_id from logs
|
||||
ClientIdCaptureSink* test_client_sink = new ClientIdCaptureSink();
|
||||
google::AddLogSink(test_client_sink);
|
||||
|
||||
test_client_ = CreateClient("localhost:17813");
|
||||
ASSERT_TRUE(test_client_ != nullptr);
|
||||
|
||||
// Wait for logs to flush
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(200));
|
||||
google::RemoveLogSink(test_client_sink);
|
||||
|
||||
if (!test_client_sink->captured_client_id.empty()) {
|
||||
UUID extracted_id =
|
||||
ParseClientId(test_client_sink->captured_client_id);
|
||||
if (extracted_id.first != 0 || extracted_id.second != 0) {
|
||||
test_client_id_ = extracted_id;
|
||||
LOG(INFO) << "Captured test_client_id: "
|
||||
<< FormatClientId(test_client_id_);
|
||||
}
|
||||
}
|
||||
delete test_client_sink;
|
||||
|
||||
// This client is used to provide segments.
|
||||
// Capture segment_provider_client_ client_id from logs
|
||||
ClientIdCaptureSink* provider_client_sink = new ClientIdCaptureSink();
|
||||
google::AddLogSink(provider_client_sink);
|
||||
|
||||
segment_provider_client_ = CreateClient("localhost:17812");
|
||||
ASSERT_TRUE(segment_provider_client_ != nullptr);
|
||||
|
||||
// Wait for logs to flush
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(200));
|
||||
google::RemoveLogSink(provider_client_sink);
|
||||
|
||||
if (!provider_client_sink->captured_client_id.empty()) {
|
||||
UUID extracted_id =
|
||||
ParseClientId(provider_client_sink->captured_client_id);
|
||||
if (extracted_id.first != 0 || extracted_id.second != 0) {
|
||||
segment_provider_client_id_ = extracted_id;
|
||||
LOG(INFO) << "Captured segment_provider_client_id: "
|
||||
<< FormatClientId(segment_provider_client_id_);
|
||||
}
|
||||
}
|
||||
delete provider_client_sink;
|
||||
|
||||
client_buffer_allocator_ =
|
||||
std::make_unique<SimpleAllocator>(128 * 1024 * 1024);
|
||||
auto register_result = test_client_->RegisterLocalMemory(
|
||||
|
|
@ -174,6 +269,8 @@ class ClientIntegrationTest : public ::testing::Test {
|
|||
static InProcMaster master_;
|
||||
static std::string master_address_;
|
||||
static std::string metadata_url_;
|
||||
static UUID test_client_id_;
|
||||
static UUID segment_provider_client_id_;
|
||||
};
|
||||
|
||||
// Static members initialization
|
||||
|
|
@ -190,6 +287,8 @@ uint64_t ClientIntegrationTest::default_kv_lease_ttl_ = 0;
|
|||
InProcMaster ClientIntegrationTest::master_;
|
||||
std::string ClientIntegrationTest::master_address_;
|
||||
std::string ClientIntegrationTest::metadata_url_;
|
||||
UUID ClientIntegrationTest::test_client_id_{0, 0};
|
||||
UUID ClientIntegrationTest::segment_provider_client_id_{0, 0};
|
||||
|
||||
// Test basic Put/Get operations through the client
|
||||
TEST_F(ClientIntegrationTest, BasicPutGetOperations) {
|
||||
|
|
@ -629,6 +728,100 @@ TEST_F(ClientIntegrationTest, BatchIsExistOperations) {
|
|||
}
|
||||
}
|
||||
|
||||
// Test batch QueryIp operations through the client
|
||||
TEST_F(ClientIntegrationTest, BatchQueryIpOperations) {
|
||||
// Skip test if we couldn't capture client_ids
|
||||
if ((test_client_id_.first == 0 && test_client_id_.second == 0) ||
|
||||
(segment_provider_client_id_.first == 0 &&
|
||||
segment_provider_client_id_.second == 0)) {
|
||||
GTEST_SKIP()
|
||||
<< "Could not capture client_ids, skipping BatchQueryIp test";
|
||||
}
|
||||
|
||||
// Test 1: Query IP for test_client_
|
||||
std::vector<UUID> client_ids = {test_client_id_};
|
||||
auto result = test_client_->BatchQueryIp(client_ids);
|
||||
|
||||
ASSERT_TRUE(result.has_value())
|
||||
<< "BatchQueryIp failed: " << toString(result.error());
|
||||
|
||||
const auto& results = result.value();
|
||||
ASSERT_FALSE(results.empty()) << "BatchQueryIp returned empty results";
|
||||
|
||||
auto it = results.find(test_client_id_);
|
||||
ASSERT_NE(it, results.end()) << "test_client_id not found in results";
|
||||
|
||||
const auto& ip_addresses = it->second;
|
||||
ASSERT_FALSE(ip_addresses.empty())
|
||||
<< "test_client_ should have at least one IP address";
|
||||
|
||||
LOG(INFO) << "test_client_ IP addresses (" << ip_addresses.size() << "):";
|
||||
for (size_t i = 0; i < ip_addresses.size(); ++i) {
|
||||
LOG(INFO) << " [" << (i + 1) << "] " << ip_addresses[i];
|
||||
}
|
||||
|
||||
// Verify IP addresses are valid (should contain "127.0.0.1" or "localhost")
|
||||
bool has_valid_ip = false;
|
||||
for (const auto& ip : ip_addresses) {
|
||||
if (ip == "127.0.0.1" || ip.find("127.0.0.1") != std::string::npos ||
|
||||
ip == "localhost" || ip.find("localhost") != std::string::npos) {
|
||||
has_valid_ip = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
EXPECT_TRUE(has_valid_ip) << "Expected at least one valid IP address";
|
||||
|
||||
// Test 2: Query IP for multiple client_ids
|
||||
std::vector<UUID> multiple_ids = {test_client_id_,
|
||||
segment_provider_client_id_};
|
||||
auto multi_result = test_client_->BatchQueryIp(multiple_ids);
|
||||
|
||||
ASSERT_TRUE(multi_result.has_value())
|
||||
<< "BatchQueryIp failed for multiple client_ids: "
|
||||
<< toString(multi_result.error());
|
||||
|
||||
const auto& multi_results = multi_result.value();
|
||||
|
||||
// Verify test_client_id_ is in results
|
||||
auto test_it = multi_results.find(test_client_id_);
|
||||
if (test_it != multi_results.end()) {
|
||||
EXPECT_FALSE(test_it->second.empty())
|
||||
<< "test_client_ should have IP addresses";
|
||||
}
|
||||
|
||||
// Verify segment_provider_client_id_ is in results
|
||||
auto provider_it = multi_results.find(segment_provider_client_id_);
|
||||
if (provider_it != multi_results.end()) {
|
||||
EXPECT_FALSE(provider_it->second.empty())
|
||||
<< "segment_provider_client_ should have IP addresses";
|
||||
LOG(INFO) << "segment_provider_client_ IP addresses ("
|
||||
<< provider_it->second.size() << "):";
|
||||
for (size_t i = 0; i < provider_it->second.size(); ++i) {
|
||||
LOG(INFO) << " [" << (i + 1) << "] " << provider_it->second[i];
|
||||
}
|
||||
}
|
||||
|
||||
// Test 3: Query with empty client_ids list
|
||||
std::vector<UUID> empty_client_ids;
|
||||
auto empty_result = test_client_->BatchQueryIp(empty_client_ids);
|
||||
|
||||
ASSERT_TRUE(empty_result.has_value());
|
||||
EXPECT_TRUE(empty_result.value().empty())
|
||||
<< "Empty client_ids should return empty results";
|
||||
|
||||
// Test 4: Query with non-existent client_id (should be silently skipped)
|
||||
UUID non_existent_client_id = generate_uuid();
|
||||
std::vector<UUID> non_existent_ids = {non_existent_client_id};
|
||||
auto non_existent_result = test_client_->BatchQueryIp(non_existent_ids);
|
||||
|
||||
ASSERT_TRUE(non_existent_result.has_value());
|
||||
// Non-existent client_id should not be in results (silently skipped)
|
||||
EXPECT_TRUE(non_existent_result.value().empty() ||
|
||||
non_existent_result.value().find(non_existent_client_id) ==
|
||||
non_existent_result.value().end())
|
||||
<< "Non-existent client_id should not be in results";
|
||||
}
|
||||
|
||||
// Test batch put with duplicate keys
|
||||
TEST_F(ClientIntegrationTest, BatchPutDuplicateKeys) {
|
||||
const std::string test_data = "test_data_duplicate";
|
||||
|
|
|
|||
|
|
@ -2087,6 +2087,148 @@ TEST_F(MasterServiceTest, BatchExistKeyTest) {
|
|||
ASSERT_FALSE(exist_resp[test_object_num].value());
|
||||
}
|
||||
|
||||
TEST_F(MasterServiceTest, BatchQueryIpTest) {
|
||||
std::unique_ptr<MasterService> service_(new MasterService());
|
||||
const UUID client_id = generate_uuid();
|
||||
|
||||
// Mount a segment with a specific te_endpoint (IP:Port format)
|
||||
constexpr size_t buffer = 0x300000000;
|
||||
constexpr size_t size = 1024 * 1024 * 16;
|
||||
Segment segment = MakeSegment("test_segment", buffer, size);
|
||||
segment.te_endpoint = "127.0.0.1:12345"; // Set IP:Port format for testing
|
||||
auto mount_result = service_->MountSegment(segment, client_id);
|
||||
ASSERT_TRUE(mount_result.has_value());
|
||||
|
||||
// Test BatchQueryIp with a single client_id
|
||||
std::vector<UUID> client_ids = {client_id};
|
||||
auto query_result = service_->BatchQueryIp(client_ids);
|
||||
|
||||
ASSERT_TRUE(query_result.has_value())
|
||||
<< "BatchQueryIp failed: " << toString(query_result.error());
|
||||
|
||||
const auto& results = query_result.value();
|
||||
ASSERT_FALSE(results.empty()) << "BatchQueryIp returned empty results";
|
||||
|
||||
auto it = results.find(client_id);
|
||||
ASSERT_NE(it, results.end()) << "Client ID not found in results";
|
||||
|
||||
const auto& ip_addresses = it->second;
|
||||
ASSERT_FALSE(ip_addresses.empty()) << "No IP addresses found for client";
|
||||
ASSERT_EQ(1u, ip_addresses.size()) << "Expected exactly 1 IP address";
|
||||
EXPECT_EQ("127.0.0.1", ip_addresses[0]) << "IP address mismatch";
|
||||
|
||||
// Test BatchQueryIp with multiple client_ids (one valid, one invalid)
|
||||
UUID non_existent_client_id = generate_uuid();
|
||||
std::vector<UUID> mixed_client_ids = {client_id, non_existent_client_id};
|
||||
auto mixed_query_result = service_->BatchQueryIp(mixed_client_ids);
|
||||
|
||||
ASSERT_TRUE(mixed_query_result.has_value());
|
||||
const auto& mixed_results = mixed_query_result.value();
|
||||
|
||||
// Valid client_id should be in results
|
||||
ASSERT_NE(mixed_results.find(client_id), mixed_results.end())
|
||||
<< "Valid client_id should be in results";
|
||||
|
||||
// Invalid client_id should not be in results (silently skipped)
|
||||
EXPECT_EQ(mixed_results.find(non_existent_client_id), mixed_results.end())
|
||||
<< "Invalid client_id should not be in results";
|
||||
}
|
||||
|
||||
TEST_F(MasterServiceTest, BatchQueryIpMultipleSegmentsTest) {
|
||||
std::unique_ptr<MasterService> service_(new MasterService());
|
||||
const UUID client_id = generate_uuid();
|
||||
|
||||
// Mount multiple segments with different IPs for the same client
|
||||
constexpr size_t buffer1 = 0x300000000;
|
||||
constexpr size_t buffer2 = 0x400000000;
|
||||
constexpr size_t size = 1024 * 1024 * 16;
|
||||
|
||||
Segment segment1 = MakeSegment("segment1", buffer1, size);
|
||||
segment1.te_endpoint = "127.0.0.1:12345";
|
||||
auto mount_result1 = service_->MountSegment(segment1, client_id);
|
||||
ASSERT_TRUE(mount_result1.has_value());
|
||||
|
||||
Segment segment2 = MakeSegment("segment2", buffer2, size);
|
||||
segment2.te_endpoint = "127.0.0.1:12346"; // Same IP, different port
|
||||
auto mount_result2 = service_->MountSegment(segment2, client_id);
|
||||
ASSERT_TRUE(mount_result2.has_value());
|
||||
|
||||
Segment segment3 = MakeSegment("segment3", 0x500000000, size);
|
||||
segment3.te_endpoint = "192.168.1.1:12345"; // Different IP
|
||||
auto mount_result3 = service_->MountSegment(segment3, client_id);
|
||||
ASSERT_TRUE(mount_result3.has_value());
|
||||
|
||||
// Test BatchQueryIp - should return unique IPs
|
||||
std::vector<UUID> client_ids = {client_id};
|
||||
auto query_result = service_->BatchQueryIp(client_ids);
|
||||
|
||||
ASSERT_TRUE(query_result.has_value());
|
||||
const auto& results = query_result.value();
|
||||
auto it = results.find(client_id);
|
||||
ASSERT_NE(it, results.end());
|
||||
|
||||
const auto& ip_addresses = it->second;
|
||||
// Should have 2 unique IPs: 127.0.0.1 and 192.168.1.1
|
||||
ASSERT_EQ(2u, ip_addresses.size()) << "Expected 2 unique IP addresses";
|
||||
|
||||
// Verify both IPs are present
|
||||
std::unordered_set<std::string> ip_set(ip_addresses.begin(),
|
||||
ip_addresses.end());
|
||||
EXPECT_NE(ip_set.find("127.0.0.1"), ip_set.end());
|
||||
EXPECT_NE(ip_set.find("192.168.1.1"), ip_set.end());
|
||||
}
|
||||
|
||||
TEST_F(MasterServiceTest, BatchQueryIpEmptyClientIdTest) {
|
||||
std::unique_ptr<MasterService> service_(new MasterService());
|
||||
|
||||
// Test with empty client_ids list
|
||||
std::vector<UUID> empty_client_ids;
|
||||
auto query_result = service_->BatchQueryIp(empty_client_ids);
|
||||
|
||||
ASSERT_TRUE(query_result.has_value());
|
||||
const auto& results = query_result.value();
|
||||
EXPECT_TRUE(results.empty())
|
||||
<< "Empty client_ids should return empty results";
|
||||
}
|
||||
|
||||
TEST_F(MasterServiceTest, BatchQueryIpMultipleSegmentsEmptyTeEndpointTest) {
|
||||
std::unique_ptr<MasterService> service_(new MasterService());
|
||||
const UUID client_id = generate_uuid();
|
||||
|
||||
// Mount multiple segments, all with empty te_endpoint
|
||||
constexpr size_t buffer1 = 0x300000000;
|
||||
constexpr size_t buffer2 = 0x400000000;
|
||||
constexpr size_t size = 1024 * 1024 * 16;
|
||||
|
||||
Segment segment1 = MakeSegment("segment1", buffer1, size);
|
||||
segment1.te_endpoint = ""; // Empty te_endpoint
|
||||
auto mount_result1 = service_->MountSegment(segment1, client_id);
|
||||
ASSERT_TRUE(mount_result1.has_value());
|
||||
|
||||
Segment segment2 = MakeSegment("segment2", buffer2, size);
|
||||
segment2.te_endpoint = ""; // Empty te_endpoint
|
||||
auto mount_result2 = service_->MountSegment(segment2, client_id);
|
||||
ASSERT_TRUE(mount_result2.has_value());
|
||||
|
||||
// Test BatchQueryIp - should return client with empty IP vector
|
||||
std::vector<UUID> client_ids = {client_id};
|
||||
auto query_result = service_->BatchQueryIp(client_ids);
|
||||
|
||||
ASSERT_TRUE(query_result.has_value());
|
||||
const auto& results = query_result.value();
|
||||
ASSERT_FALSE(results.empty())
|
||||
<< "BatchQueryIp should include client in results even with empty IPs";
|
||||
|
||||
auto it = results.find(client_id);
|
||||
ASSERT_NE(it, results.end()) << "Client ID should be found in results even "
|
||||
"with all empty te_endpoints";
|
||||
|
||||
// Verify the IP vector is empty
|
||||
const auto& ip_addresses = it->second;
|
||||
EXPECT_TRUE(ip_addresses.empty())
|
||||
<< "Client with all empty te_endpoints should have empty IP vector";
|
||||
}
|
||||
|
||||
TEST_F(MasterServiceTest, PutStartExpiringTest) {
|
||||
// Reset storage space metrics.
|
||||
MasterMetricManager::instance().reset_allocated_mem_size();
|
||||
|
|
|
|||
Loading…
Reference in New Issue