mirror of https://github.com/grpc/grpc.git
[XDS Interop] Move XdsStatsWatcher to a separate file. (#34000)
This will help with introducing test coverage as the logic becomes more complex.pull/33996/head^2
Eugene Ostroukhov
2 years ago
committed by
GitHub
parent
bc41f18beb
commit
18be986e3b
8 changed files with 475 additions and 159 deletions
@ -0,0 +1,118 @@ |
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "test/cpp/interop/xds_stats_watcher.h" |
||||
|
||||
#include <map> |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
XdsStatsWatcher::XdsStatsWatcher(int start_id, int end_id) |
||||
: start_id_(start_id), end_id_(end_id), rpcs_needed_(end_id - start_id) {} |
||||
|
||||
void XdsStatsWatcher::RpcCompleted(const AsyncClientCallResult& call, |
||||
const std::string& peer) { |
||||
// We count RPCs for global watcher or if the request_id falls into the
|
||||
// watcher's interested range of request ids.
|
||||
if ((start_id_ == 0 && end_id_ == 0) || |
||||
(start_id_ <= call.saved_request_id && call.saved_request_id < end_id_)) { |
||||
{ |
||||
std::lock_guard<std::mutex> lock(m_); |
||||
if (peer.empty()) { |
||||
no_remote_peer_++; |
||||
++no_remote_peer_by_type_[call.rpc_type]; |
||||
} else { |
||||
// RPC is counted into both per-peer bin and per-method-per-peer bin.
|
||||
rpcs_by_peer_[peer]++; |
||||
rpcs_by_type_[call.rpc_type][peer]++; |
||||
} |
||||
rpcs_needed_--; |
||||
// Report accumulated stats.
|
||||
auto& stats_per_method = *accumulated_stats_.mutable_stats_per_method(); |
||||
auto& method_stat = |
||||
stats_per_method[ClientConfigureRequest_RpcType_Name(call.rpc_type)]; |
||||
auto& result = *method_stat.mutable_result(); |
||||
grpc_status_code code = |
||||
static_cast<grpc_status_code>(call.status.error_code()); |
||||
auto& num_rpcs = result[code]; |
||||
++num_rpcs; |
||||
auto rpcs_started = method_stat.rpcs_started(); |
||||
method_stat.set_rpcs_started(++rpcs_started); |
||||
} |
||||
cv_.notify_one(); |
||||
} |
||||
} |
||||
|
||||
void XdsStatsWatcher::WaitForRpcStatsResponse( |
||||
LoadBalancerStatsResponse* response, int timeout_sec) { |
||||
std::unique_lock<std::mutex> lock(m_); |
||||
cv_.wait_for(lock, std::chrono::seconds(timeout_sec), |
||||
[this] { return rpcs_needed_ == 0; }); |
||||
response->mutable_rpcs_by_peer()->insert(rpcs_by_peer_.begin(), |
||||
rpcs_by_peer_.end()); |
||||
auto& response_rpcs_by_method = *response->mutable_rpcs_by_method(); |
||||
for (const auto& rpc_by_type : rpcs_by_type_) { |
||||
std::string method_name; |
||||
if (rpc_by_type.first == ClientConfigureRequest::EMPTY_CALL) { |
||||
method_name = "EmptyCall"; |
||||
} else if (rpc_by_type.first == ClientConfigureRequest::UNARY_CALL) { |
||||
method_name = "UnaryCall"; |
||||
} else { |
||||
GPR_ASSERT(0); |
||||
} |
||||
// TODO(@donnadionne): When the test runner changes to accept EMPTY_CALL
|
||||
// and UNARY_CALL we will just use the name of the enum instead of the
|
||||
// method_name variable.
|
||||
auto& response_rpc_by_method = response_rpcs_by_method[method_name]; |
||||
auto& response_rpcs_by_peer = |
||||
*response_rpc_by_method.mutable_rpcs_by_peer(); |
||||
for (const auto& rpc_by_peer : rpc_by_type.second) { |
||||
auto& response_rpc_by_peer = response_rpcs_by_peer[rpc_by_peer.first]; |
||||
response_rpc_by_peer = rpc_by_peer.second; |
||||
} |
||||
} |
||||
response->set_num_failures(no_remote_peer_ + rpcs_needed_); |
||||
} |
||||
|
||||
void XdsStatsWatcher::GetCurrentRpcStats( |
||||
LoadBalancerAccumulatedStatsResponse* response, |
||||
StatsWatchers* stats_watchers) { |
||||
std::unique_lock<std::mutex> lock(m_); |
||||
response->CopyFrom(accumulated_stats_); |
||||
// TODO(@donnadionne): delete deprecated stats below when the test is no
|
||||
// longer using them.
|
||||
auto& response_rpcs_started_by_method = |
||||
*response->mutable_num_rpcs_started_by_method(); |
||||
auto& response_rpcs_succeeded_by_method = |
||||
*response->mutable_num_rpcs_succeeded_by_method(); |
||||
auto& response_rpcs_failed_by_method = |
||||
*response->mutable_num_rpcs_failed_by_method(); |
||||
for (const auto& rpc_by_type : rpcs_by_type_) { |
||||
auto total_succeeded = 0; |
||||
for (const auto& rpc_by_peer : rpc_by_type.second) { |
||||
total_succeeded += rpc_by_peer.second; |
||||
} |
||||
response_rpcs_succeeded_by_method[ClientConfigureRequest_RpcType_Name( |
||||
rpc_by_type.first)] = total_succeeded; |
||||
response_rpcs_started_by_method[ClientConfigureRequest_RpcType_Name( |
||||
rpc_by_type.first)] = |
||||
stats_watchers->global_request_id_by_type[rpc_by_type.first]; |
||||
response_rpcs_failed_by_method[ClientConfigureRequest_RpcType_Name( |
||||
rpc_by_type.first)] = no_remote_peer_by_type_[rpc_by_type.first]; |
||||
} |
||||
} |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,104 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
//
|
||||
|
||||
#ifndef GRPC_TEST_CPP_INTEROP_XDS_STATS_WATCHER_H |
||||
#define GRPC_TEST_CPP_INTEROP_XDS_STATS_WATCHER_H |
||||
|
||||
#include <atomic> |
||||
#include <chrono> |
||||
#include <condition_variable> |
||||
#include <deque> |
||||
#include <map> |
||||
#include <mutex> |
||||
#include <set> |
||||
#include <sstream> |
||||
#include <string> |
||||
#include <thread> |
||||
#include <vector> |
||||
|
||||
#include "absl/status/status.h" |
||||
|
||||
#include <grpcpp/grpcpp.h> |
||||
|
||||
#include "src/proto/grpc/testing/empty.pb.h" |
||||
#include "src/proto/grpc/testing/messages.pb.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
|
||||
class XdsStatsWatcher; |
||||
|
||||
struct AsyncClientCallResult { |
||||
Empty empty_response; |
||||
SimpleResponse simple_response; |
||||
Status status; |
||||
int saved_request_id; |
||||
ClientConfigureRequest::RpcType rpc_type; |
||||
}; |
||||
|
||||
struct StatsWatchers { |
||||
// Unique ID for each outgoing RPC
|
||||
int global_request_id = 0; |
||||
// Unique ID for each outgoing RPC by RPC method type
|
||||
std::map<int, int> global_request_id_by_type; |
||||
// Stores a set of watchers that should be notified upon outgoing RPC
|
||||
// completion
|
||||
std::set<XdsStatsWatcher*> watchers; |
||||
// Global watcher for accumululated stats.
|
||||
XdsStatsWatcher* global_watcher; |
||||
// Mutex for global_request_id and watchers
|
||||
std::mutex mu; |
||||
}; |
||||
|
||||
/// Records the remote peer distribution for a given range of RPCs.
|
||||
class XdsStatsWatcher { |
||||
public: |
||||
XdsStatsWatcher(int start_id, int end_id); |
||||
|
||||
// Upon the completion of an RPC, we will look at the request_id, the
|
||||
// rpc_type, and the peer the RPC was sent to in order to count
|
||||
// this RPC into the right stats bin.
|
||||
void RpcCompleted(const AsyncClientCallResult& call, const std::string& peer); |
||||
|
||||
void WaitForRpcStatsResponse(LoadBalancerStatsResponse* response, |
||||
int timeout_sec); |
||||
|
||||
void GetCurrentRpcStats(LoadBalancerAccumulatedStatsResponse* response, |
||||
StatsWatchers* stats_watchers); |
||||
|
||||
private: |
||||
int start_id_; |
||||
int end_id_; |
||||
int rpcs_needed_; |
||||
int no_remote_peer_ = 0; |
||||
std::map<int, int> no_remote_peer_by_type_; |
||||
// A map of stats keyed by peer name.
|
||||
std::map<std::string, int> rpcs_by_peer_; |
||||
// A two-level map of stats keyed at top level by RPC method and second level
|
||||
// by peer name.
|
||||
std::map<int, std::map<std::string, int>> rpcs_by_type_; |
||||
// Storing accumulated stats in the response proto format.
|
||||
LoadBalancerAccumulatedStatsResponse accumulated_stats_; |
||||
std::mutex m_; |
||||
std::condition_variable cv_; |
||||
}; |
||||
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_TEST_CPP_INTEROP_XDS_STATS_WATCHER_H
|
||||
@ -0,0 +1,71 @@ |
||||
//
|
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
|
||||
#include "test/cpp/interop/xds_stats_watcher.h" |
||||
|
||||
#include <map> |
||||
#include <memory> |
||||
|
||||
#include <gmock/gmock.h> |
||||
#include <gtest/gtest.h> |
||||
|
||||
#include <grpc/grpc.h> |
||||
|
||||
#include "test/core/util/test_config.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
AsyncClientCallResult BuildCallResult(int saved_request_id) { |
||||
AsyncClientCallResult result; |
||||
result.saved_request_id = saved_request_id; |
||||
result.rpc_type = ClientConfigureRequest::UNARY_CALL; |
||||
return result; |
||||
} |
||||
|
||||
TEST(XdsStatsWatcherTest, CollectsMetadata) { |
||||
XdsStatsWatcher watcher(0, 3); |
||||
watcher.RpcCompleted(BuildCallResult(0), "peer1"); |
||||
watcher.RpcCompleted(BuildCallResult(1), "peer1"); |
||||
watcher.RpcCompleted(BuildCallResult(2), "peer2"); |
||||
LoadBalancerStatsResponse lb_response; |
||||
watcher.WaitForRpcStatsResponse(&lb_response, 1); |
||||
EXPECT_EQ( |
||||
(std::multimap<std::string, int32_t>(lb_response.rpcs_by_peer().begin(), |
||||
lb_response.rpcs_by_peer().end())), |
||||
(std::multimap<std::string, int32_t>({{"peer1", 2}, {"peer2", 1}}))); |
||||
EXPECT_EQ(lb_response.rpcs_by_method_size(), 1); |
||||
auto rpcs = lb_response.rpcs_by_method().find("UnaryCall"); |
||||
EXPECT_NE(rpcs, lb_response.rpcs_by_method().end()); |
||||
std::multimap<std::string, int32_t> by_peer( |
||||
rpcs->second.rpcs_by_peer().begin(), rpcs->second.rpcs_by_peer().end()); |
||||
EXPECT_EQ( |
||||
by_peer, |
||||
(std::multimap<std::string, int32_t>({{"peer1", 2}, {"peer2", 1}}))); |
||||
} |
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc::testing::TestEnvironment env(&argc, argv); |
||||
grpc_init(); |
||||
auto result = RUN_ALL_TESTS(); |
||||
grpc_shutdown(); |
||||
return result; |
||||
} |
||||
Loading…
Reference in new issue