Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

[csds] Make grpc_dump_xds_configs return ClientStatusResponse (#35612)

Browse Source 
Closes #35612

PiperOrigin-RevId: 602443727
pull/35731/head
Eugene Ostroukhov 10 months ago committed by Copybara-Service
parent fb14685fd6
commit 8ea3b417cc
11 changed files with 252 additions and 213 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 213
      src/core/ext/xds/xds_api.cc
  2. 9
      src/core/ext/xds/xds_api.h
  3. 89
      src/core/ext/xds/xds_client.cc
  4. 25
      src/core/ext/xds/xds_client.h
  5. 56
      src/core/ext/xds/xds_client_grpc.cc
  6. 3
      src/core/ext/xds/xds_client_grpc.h
  7. 31
      src/cpp/server/csds/csds.cc
  8. 5
      src/python/grpcio_csds/grpc_csds/__init__.py
  9. 2
      src/python/grpcio_tests/tests/admin/admin_test.py
  10. 10
      src/python/grpcio_tests/tests/csds/csds_test.py
  11. 22
      test/core/xds/xds_client_fuzzer.cc

213
src/core/ext/xds/xds_api.cc
Unescape View File

@ -75,32 +75,29 @@ struct XdsApiContext {
upb_Arena* arena;
};
void PopulateMetadataValue(const XdsApiContext& context,
google_protobuf_Value* value_pb, const Json& value);
void PopulateMetadataValue(google_protobuf_Value* value_pb, const Json& value,
upb_Arena* arena);
void PopulateListValue(const XdsApiContext& context,
google_protobuf_ListValue* list_value,
const Json::Array& values) {
void PopulateListValue(google_protobuf_ListValue* list_value,
const Json::Array& values, upb_Arena* arena) {
for (const auto& value : values) {
auto* value_pb =
google_protobuf_ListValue_add_values(list_value, context.arena);
PopulateMetadataValue(context, value_pb, value);
auto* value_pb = google_protobuf_ListValue_add_values(list_value, arena);
PopulateMetadataValue(value_pb, value, arena);
}
}
void PopulateMetadata(const XdsApiContext& context,
google_protobuf_Struct* metadata_pb,
const Json::Object& metadata) {
void PopulateMetadata(google_protobuf_Struct* metadata_pb,
const Json::Object& metadata, upb_Arena* arena) {
for (const auto& p : metadata) {
google_protobuf_Value* value = google_protobuf_Value_new(context.arena);
PopulateMetadataValue(context, value, p.second);
google_protobuf_Value* value = google_protobuf_Value_new(arena);
PopulateMetadataValue(value, p.second, arena);
google_protobuf_Struct_fields_set(
metadata_pb, StdStringToUpbString(p.first), value, context.arena);
metadata_pb, StdStringToUpbString(p.first), value, arena);
}
}
void PopulateMetadataValue(const XdsApiContext& context,
google_protobuf_Value* value_pb, const Json& value) {
void PopulateMetadataValue(google_protobuf_Value* value_pb, const Json& value,
upb_Arena* arena) {
switch (value.type()) {
case Json::Type::kNull:
google_protobuf_Value_set_null_value(value_pb, 0);
@ -118,65 +115,19 @@ void PopulateMetadataValue(const XdsApiContext& context,
break;
case Json::Type::kObject: {
google_protobuf_Struct* struct_value =
google_protobuf_Value_mutable_struct_value(value_pb, context.arena);
PopulateMetadata(context, struct_value, value.object());
google_protobuf_Value_mutable_struct_value(value_pb, arena);
PopulateMetadata(struct_value, value.object(), arena);
break;
}
case Json::Type::kArray: {
google_protobuf_ListValue* list_value =
google_protobuf_Value_mutable_list_value(value_pb, context.arena);
PopulateListValue(context, list_value, value.array());
google_protobuf_Value_mutable_list_value(value_pb, arena);
PopulateListValue(list_value, value.array(), arena);
break;
}
}
}
void PopulateNode(const XdsApiContext& context, const XdsBootstrap::Node* node,
const std::string& user_agent_name,
const std::string& user_agent_version,
envoy_config_core_v3_Node* node_msg) {
if (node != nullptr) {
if (!node->id().empty()) {
envoy_config_core_v3_Node_set_id(node_msg,
StdStringToUpbString(node->id()));
}
if (!node->cluster().empty()) {
envoy_config_core_v3_Node_set_cluster(
node_msg, StdStringToUpbString(node->cluster()));
}
if (!node->metadata().empty()) {
google_protobuf_Struct* metadata =
envoy_config_core_v3_Node_mutable_metadata(node_msg, context.arena);
PopulateMetadata(context, metadata, node->metadata());
}
if (!node->locality_region().empty() || !node->locality_zone().empty() ||
!node->locality_sub_zone().empty()) {
envoy_config_core_v3_Locality* locality =
envoy_config_core_v3_Node_mutable_locality(node_msg, context.arena);
if (!node->locality_region().empty()) {
envoy_config_core_v3_Locality_set_region(
locality, StdStringToUpbString(node->locality_region()));
}
if (!node->locality_zone().empty()) {
envoy_config_core_v3_Locality_set_zone(
locality, StdStringToUpbString(node->locality_zone()));
}
if (!node->locality_sub_zone().empty()) {
envoy_config_core_v3_Locality_set_sub_zone(
locality, StdStringToUpbString(node->locality_sub_zone()));
}
}
}
envoy_config_core_v3_Node_set_user_agent_name(
node_msg, StdStringToUpbString(user_agent_name));
envoy_config_core_v3_Node_set_user_agent_version(
node_msg, StdStringToUpbString(user_agent_version));
envoy_config_core_v3_Node_add_client_features(
node_msg,
upb_StringView_FromString("envoy.lb.does_not_support_overprovisioning"),
context.arena);
}
void MaybeLogDiscoveryRequest(
const XdsApiContext& context,
const envoy_service_discovery_v3_DiscoveryRequest* request) {
@ -203,6 +154,50 @@ std::string SerializeDiscoveryRequest(
} // namespace
void XdsApi::PopulateNode(envoy_config_core_v3_Node* node_msg,
upb_Arena* arena) {
if (node_ != nullptr) {
if (!node_->id().empty()) {
envoy_config_core_v3_Node_set_id(node_msg,
StdStringToUpbString(node_->id()));
}
if (!node_->cluster().empty()) {
envoy_config_core_v3_Node_set_cluster(
node_msg, StdStringToUpbString(node_->cluster()));
}
if (!node_->metadata().empty()) {
google_protobuf_Struct* metadata =
envoy_config_core_v3_Node_mutable_metadata(node_msg, arena);
PopulateMetadata(metadata, node_->metadata(), arena);
}
if (!node_->locality_region().empty() || !node_->locality_zone().empty() ||
!node_->locality_sub_zone().empty()) {
envoy_config_core_v3_Locality* locality =
envoy_config_core_v3_Node_mutable_locality(node_msg, arena);
if (!node_->locality_region().empty()) {
envoy_config_core_v3_Locality_set_region(
locality, StdStringToUpbString(node_->locality_region()));
}
if (!node_->locality_zone().empty()) {
envoy_config_core_v3_Locality_set_zone(
locality, StdStringToUpbString(node_->locality_zone()));
}
if (!node_->locality_sub_zone().empty()) {
envoy_config_core_v3_Locality_set_sub_zone(
locality, StdStringToUpbString(node_->locality_sub_zone()));
}
}
}
envoy_config_core_v3_Node_set_user_agent_name(
node_msg, StdStringToUpbString(user_agent_name_));
envoy_config_core_v3_Node_set_user_agent_version(
node_msg, StdStringToUpbString(user_agent_version_));
envoy_config_core_v3_Node_add_client_features(
node_msg,
upb_StringView_FromString("envoy.lb.does_not_support_overprovisioning"),
arena);
}
std::string XdsApi::CreateAdsRequest(
absl::string_view type_url, absl::string_view version,
absl::string_view nonce, const std::vector<std::string>& resource_names,
@ -249,8 +244,7 @@ std::string XdsApi::CreateAdsRequest(
envoy_config_core_v3_Node* node_msg =
envoy_service_discovery_v3_DiscoveryRequest_mutable_node(request,
arena.ptr());
PopulateNode(context, node_, user_agent_name_, user_agent_version_,
node_msg);
PopulateNode(node_msg, arena.ptr());
envoy_config_core_v3_Node_add_client_features(
node_msg, upb_StringView_FromString("xds.config.resource-in-sotw"),
context.arena);
@ -393,7 +387,7 @@ std::string XdsApi::CreateLrsInitialRequest() {
envoy_config_core_v3_Node* node_msg =
envoy_service_load_stats_v3_LoadStatsRequest_mutable_node(request,
arena.ptr());
PopulateNode(context, node_, user_agent_name_, user_agent_version_, node_msg);
PopulateNode(node_msg, arena.ptr());
envoy_config_core_v3_Node_add_client_features(
node_msg,
upb_StringView_FromString("envoy.lrs.supports_send_all_clusters"),
@ -575,85 +569,4 @@ absl::Status XdsApi::ParseLrsResponse(absl::string_view encoded_response,
return absl::OkStatus();
}
namespace {
google_protobuf_Timestamp* EncodeTimestamp(const XdsApiContext& context,
Timestamp value) {
google_protobuf_Timestamp* timestamp =
google_protobuf_Timestamp_new(context.arena);
gpr_timespec timespec = value.as_timespec(GPR_CLOCK_REALTIME);
google_protobuf_Timestamp_set_seconds(timestamp, timespec.tv_sec);
google_protobuf_Timestamp_set_nanos(timestamp, timespec.tv_nsec);
return timestamp;
}
} // namespace
std::string XdsApi::AssembleClientConfig(
const ResourceTypeMetadataMap& resource_type_metadata_map) {
upb::Arena arena;
// Create the ClientConfig for resource metadata from XdsClient
auto* client_config = envoy_service_status_v3_ClientConfig_new(arena.ptr());
// Fill-in the node information
auto* node = envoy_service_status_v3_ClientConfig_mutable_node(client_config,
arena.ptr());
const XdsApiContext context = {client_, tracer_, def_pool_->ptr(),
arena.ptr()};
PopulateNode(context, node_, user_agent_name_, user_agent_version_, node);
// Dump each resource.
std::vector<std::string> type_url_storage;
for (const auto& p : resource_type_metadata_map) {
absl::string_view type_url = p.first;
const ResourceMetadataMap& resource_metadata_map = p.second;
type_url_storage.emplace_back(
absl::StrCat("type.googleapis.com/", type_url));
for (const auto& q : resource_metadata_map) {
absl::string_view resource_name = q.first;
const ResourceMetadata& metadata = *q.second;
auto* entry =
envoy_service_status_v3_ClientConfig_add_generic_xds_configs(
client_config, context.arena);
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_type_url(
entry, StdStringToUpbString(type_url_storage.back()));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_name(
entry, StdStringToUpbString(resource_name));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_client_status(
entry, metadata.client_status);
if (!metadata.serialized_proto.empty()) {
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_version_info(
entry, StdStringToUpbString(metadata.version));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_last_updated(
entry, EncodeTimestamp(context, metadata.update_time));
auto* any_field =
envoy_service_status_v3_ClientConfig_GenericXdsConfig_mutable_xds_config(
entry, context.arena);
google_protobuf_Any_set_type_url(
any_field, StdStringToUpbString(type_url_storage.back()));
google_protobuf_Any_set_value(
any_field, StdStringToUpbString(metadata.serialized_proto));
}
if (metadata.client_status == XdsApi::ResourceMetadata::NACKED) {
auto* update_failure_state =
envoy_admin_v3_UpdateFailureState_new(context.arena);
envoy_admin_v3_UpdateFailureState_set_details(
update_failure_state,
StdStringToUpbString(metadata.failed_details));
envoy_admin_v3_UpdateFailureState_set_version_info(
update_failure_state,
StdStringToUpbString(metadata.failed_version));
envoy_admin_v3_UpdateFailureState_set_last_update_attempt(
update_failure_state,
EncodeTimestamp(context, metadata.failed_update_time));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_error_state(
entry, update_failure_state);
}
}
}
// Serialize the upb message to bytes
size_t output_length;
char* output = envoy_service_status_v3_ClientConfig_serialize(
client_config, arena.ptr(), &output_length);
return std::string(output, output_length);
}
} // namespace grpc_core

9
src/core/ext/xds/xds_api.h
Unescape View File

@ -30,6 +30,7 @@
#include "absl/status/status.h"
#include "absl/strings/string_view.h"
#include "envoy/admin/v3/config_dump_shared.upb.h"
#include "envoy/service/status/v3/csds.upb.h"
#include "upb/mem/arena.h"
#include "upb/reflection/def.hpp"
@ -126,10 +127,6 @@ class XdsApi {
// Timestamp of the last failed update attempt.
Timestamp failed_update_time;
};
using ResourceMetadataMap =
std::map<std::string /*resource_name*/, const ResourceMetadata*>;
using ResourceTypeMetadataMap =
std::map<absl::string_view /*type_url*/, ResourceMetadataMap>;
static_assert(static_cast<ResourceMetadata::ClientResourceStatus>(
envoy_admin_v3_REQUESTED) ==
ResourceMetadata::ClientResourceStatus::REQUESTED,
@ -176,9 +173,7 @@ class XdsApi {
std::set<std::string>* cluster_names,
Duration* load_reporting_interval);
// Assemble the client config proto message and return the serialized result.
std::string AssembleClientConfig(
const ResourceTypeMetadataMap& resource_type_metadata_map);
void PopulateNode(envoy_config_core_v3_Node* node_msg, upb_Arena* arena);
private:
XdsClient* client_;

89
src/core/ext/xds/xds_client.cc
Unescape View File

@ -34,11 +34,17 @@
#include "absl/strings/string_view.h"
#include "absl/strings/strip.h"
#include "absl/types/optional.h"
#include "envoy/config/core/v3/base.upb.h"
#include "envoy/service/status/v3/csds.upb.h"
#include "google/protobuf/any.upb.h"
#include "google/protobuf/timestamp.upb.h"
#include "upb/base/string_view.h"
#include "upb/mem/arena.h"
#include <grpc/event_engine/event_engine.h>
#include <grpc/support/log.h>
#include "src/core/ext/xds/upb_utils.h"
#include "src/core/ext/xds/xds_api.h"
#include "src/core/ext/xds/xds_bootstrap.h"
#include "src/core/ext/xds/xds_client_stats.h"
@ -2032,25 +2038,86 @@ XdsApi::ClusterLoadReportMap XdsClient::BuildLoadReportSnapshotLocked(
return snapshot_map;
}
std::string XdsClient::DumpClientConfigBinary() {
MutexLock lock(&mu_);
XdsApi::ResourceTypeMetadataMap resource_type_metadata_map;
namespace {
google_protobuf_Timestamp* EncodeTimestamp(Timestamp value, upb_Arena* arena) {
google_protobuf_Timestamp* timestamp = google_protobuf_Timestamp_new(arena);
gpr_timespec timespec = value.as_timespec(GPR_CLOCK_REALTIME);
google_protobuf_Timestamp_set_seconds(timestamp, timespec.tv_sec);
google_protobuf_Timestamp_set_nanos(timestamp, timespec.tv_nsec);
return timestamp;
}
void FillGenericXdsConfig(
const XdsApi::ResourceMetadata& metadata, upb_StringView type_url,
upb_StringView resource_name, upb_Arena* arena,
envoy_service_status_v3_ClientConfig_GenericXdsConfig* entry) {
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_type_url(entry,
type_url);
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_name(entry,
resource_name);
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_client_status(
entry, metadata.client_status);
if (!metadata.serialized_proto.empty()) {
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_version_info(
entry, StdStringToUpbString(metadata.version));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_last_updated(
entry, EncodeTimestamp(metadata.update_time, arena));
auto* any_field =
envoy_service_status_v3_ClientConfig_GenericXdsConfig_mutable_xds_config(
entry, arena);
google_protobuf_Any_set_type_url(any_field, type_url);
google_protobuf_Any_set_value(
any_field, StdStringToUpbString(metadata.serialized_proto));
}
if (metadata.client_status == XdsApi::ResourceMetadata::NACKED) {
auto* update_failure_state = envoy_admin_v3_UpdateFailureState_new(arena);
envoy_admin_v3_UpdateFailureState_set_details(
update_failure_state, StdStringToUpbString(metadata.failed_details));
envoy_admin_v3_UpdateFailureState_set_version_info(
update_failure_state, StdStringToUpbString(metadata.failed_version));
envoy_admin_v3_UpdateFailureState_set_last_update_attempt(
update_failure_state,
EncodeTimestamp(metadata.failed_update_time, arena));
envoy_service_status_v3_ClientConfig_GenericXdsConfig_set_error_state(
entry, update_failure_state);
}
}
} // namespace
void XdsClient::DumpClientConfig(
std::set<std::string>* string_pool, upb_Arena* arena,
envoy_service_status_v3_ClientConfig* client_config) {
// Assemble config dump messages
// Fill-in the node information
auto* node =
envoy_service_status_v3_ClientConfig_mutable_node(client_config, arena);
api_.PopulateNode(node, arena);
// Dump each resource.
for (const auto& a : authority_state_map_) { // authority
const std::string& authority = a.first;
for (const auto& t : a.second.resource_map) { // type
const XdsResourceType* type = t.first;
auto& resource_metadata_map =
resource_type_metadata_map[type->type_url()];
auto it =
string_pool
->emplace(absl::StrCat("type.googleapis.com/", type->type_url()))
.first;
upb_StringView type_url = StdStringToUpbString(*it);
for (const auto& r : t.second) { // resource id
const XdsResourceKey& resource_key = r.first;
const ResourceState& resource_state = r.second;
resource_metadata_map[ConstructFullXdsResourceName(
authority, type->type_url(), resource_key)] = &resource_state.meta;
auto it2 = string_pool
->emplace(ConstructFullXdsResourceName(
authority, type->type_url(), r.first))
.first;
upb_StringView resource_name = StdStringToUpbString(*it2);
envoy_service_status_v3_ClientConfig_GenericXdsConfig* entry =
envoy_service_status_v3_ClientConfig_add_generic_xds_configs(
client_config, arena);
FillGenericXdsConfig(r.second.meta, type_url, resource_name, arena,
entry);
}
}
}
// Assemble config dump messages
return api_.AssembleClientConfig(resource_type_metadata_map);
}
} // namespace grpc_core

25
src/core/ext/xds/xds_client.h
Unescape View File

@ -51,6 +51,10 @@
namespace grpc_core {
namespace testing {
class XdsClientTestPeer;
}
extern TraceFlag grpc_xds_client_trace;
extern TraceFlag grpc_xds_client_refcount_trace;
@ -147,20 +151,23 @@ class XdsClient : public DualRefCounted<XdsClient> {
// Resets connection backoff state.
void ResetBackoff();
// Dumps the active xDS config in JSON format.
// Individual xDS resource is encoded as envoy.admin.v3.*ConfigDump. Returns
// envoy.service.status.v3.ClientConfig which also includes the config
// status (e.g., CLIENT_REQUESTED, CLIENT_ACKED, CLIENT_NACKED).
//
// Expected to be invoked by wrapper languages in their CSDS service
// implementation.
std::string DumpClientConfigBinary();
grpc_event_engine::experimental::EventEngine* engine() {
return engine_.get();
}
protected:
// Dumps the active xDS config to the provided
// envoy.service.status.v3.ClientConfig message including the config status
// (e.g., CLIENT_REQUESTED, CLIENT_ACKED, CLIENT_NACKED).
void DumpClientConfig(std::set<std::string>* string_pool, upb_Arena* arena,
envoy_service_status_v3_ClientConfig* client_config)
ABSL_EXCLUSIVE_LOCKS_REQUIRED(&mu_);
Mutex* mu() ABSL_LOCK_RETURNED(&mu_) { return &mu_; }
private:
friend testing::XdsClientTestPeer;
struct XdsResourceKey {
std::string id;
std::vector<URI::QueryParam> query_params;

56
src/core/ext/xds/xds_client_grpc.cc
Unescape View File

@ -19,15 +19,19 @@
#include "src/core/ext/xds/xds_client_grpc.h"
#include <algorithm>
#include <cstddef>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "absl/base/thread_annotations.h"
#include "absl/status/status.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "envoy/service/status/v3/csds.upb.h"
#include "upb/base/string_view.h"
#include <grpc/grpc.h>
#include <grpc/impl/channel_arg_names.h>
@ -36,9 +40,12 @@
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include "src/core/ext/xds/upb_utils.h"
#include "src/core/ext/xds/xds_api.h"
#include "src/core/ext/xds/xds_bootstrap.h"
#include "src/core/ext/xds/xds_bootstrap_grpc.h"
#include "src/core/ext/xds/xds_channel_args.h"
#include "src/core/ext/xds/xds_client.h"
#include "src/core/ext/xds/xds_transport.h"
#include "src/core/ext/xds/xds_transport_grpc.h"
#include "src/core/lib/channel/channel_args.h"
@ -133,6 +140,19 @@ absl::StatusOr<std::string> GetBootstrapContents(const char* fallback_config) {
"not defined");
}
std::vector<RefCountedPtr<GrpcXdsClient>> GetAllXdsClients() {
MutexLock lock(g_mu);
std::vector<RefCountedPtr<GrpcXdsClient>> xds_clients;
if (g_xds_client != nullptr) {
auto xds_client =
g_xds_client->RefIfNonZero(DEBUG_LOCATION, "DumpAllClientConfigs");
if (xds_client != nullptr) {
xds_clients.emplace_back(xds_client.TakeAsSubclass<GrpcXdsClient>());
}
}
return xds_clients;
}
} // namespace
absl::StatusOr<RefCountedPtr<GrpcXdsClient>> GrpcXdsClient::GetOrCreate(
@ -178,6 +198,34 @@ absl::StatusOr<RefCountedPtr<GrpcXdsClient>> GrpcXdsClient::GetOrCreate(
return xds_client;
}
// ABSL_NO_THREAD_SAFETY_ANALYSIS because we have to manually manage locks for
// individual XdsClients and compiler struggles with checking the validity
grpc_slice GrpcXdsClient::DumpAllClientConfigs()
ABSL_NO_THREAD_SAFETY_ANALYSIS {
auto xds_clients = GetAllXdsClients();
upb::Arena arena;
// Contains strings that should survive till serialization
std::set<std::string> string_pool;
auto response = envoy_service_status_v3_ClientStatusResponse_new(arena.ptr());
// We lock each XdsClient mutex till we are done with the serialization to
// ensure that all data referenced from the UPB proto message stays alive.
for (const auto& xds_client : xds_clients) {
auto client_config =
envoy_service_status_v3_ClientStatusResponse_add_config(response,
arena.ptr());
xds_client->mu()->Lock();
xds_client->DumpClientConfig(&string_pool, arena.ptr(), client_config);
}
// Serialize the upb message to bytes
size_t output_length;
char* output = envoy_service_status_v3_ClientStatusResponse_serialize(
response, arena.ptr(), &output_length);
for (const auto& xds_client : xds_clients) {
xds_client->mu()->Unlock();
}
return grpc_slice_from_cpp_string(std::string(output, output_length));
}
GrpcXdsClient::GrpcXdsClient(
std::unique_ptr<GrpcXdsBootstrap> bootstrap, const ChannelArgs& args,
OrphanablePtr<XdsTransportFactory> transport_factory)
@ -233,11 +281,5 @@ void SetXdsFallbackBootstrapConfig(const char* config) {
grpc_slice grpc_dump_xds_configs(void) {
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::ExecCtx exec_ctx;
auto xds_client = grpc_core::GrpcXdsClient::GetOrCreate(
grpc_core::ChannelArgs(), "grpc_dump_xds_configs()");
if (!xds_client.ok()) {
// If we aren't using xDS, just return an empty string.
return grpc_empty_slice();
}
return grpc_slice_from_cpp_string((*xds_client)->DumpClientConfigBinary());
return grpc_core::GrpcXdsClient::DumpAllClientConfigs();
}

3
src/core/ext/xds/xds_client_grpc.h
Unescape View File

@ -45,6 +45,9 @@ class GrpcXdsClient : public XdsClient {
static absl::StatusOr<RefCountedPtr<GrpcXdsClient>> GetOrCreate(
const ChannelArgs& args, const char* reason);
// Builds ClientStatusResponse containing all resources from all XdsClients
static grpc_slice DumpAllClientConfigs();
// Do not instantiate directly -- use GetOrCreate() instead.
// TODO(roth): The transport factory is injectable here to support
// tests that want to use a fake transport factory with code that

31
src/cpp/server/csds/csds.cc
Unescape View File

@ -34,21 +34,20 @@ namespace grpc {
namespace xds {
namespace experimental {
using envoy::service::status::v3::ClientConfig;
using envoy::service::status::v3::ClientStatusRequest;
using envoy::service::status::v3::ClientStatusResponse;
namespace {
absl::StatusOr<ClientConfig> DumpClientConfig() {
ClientConfig client_config;
absl::StatusOr<ClientStatusResponse> DumpClientStatusResponse() {
ClientStatusResponse response;
grpc_slice serialized_client_config = grpc_dump_xds_configs();
std::string bytes = StringFromCopiedSlice(serialized_client_config);
grpc_slice_unref(serialized_client_config);
if (!client_config.ParseFromString(bytes)) {
return absl::InternalError("Failed to parse ClientConfig.");
if (!response.ParseFromString(bytes)) {
return absl::InternalError("Failed to parse ClientStatusResponse.");
}
return client_config;
return response;
}
} // namespace
@ -58,19 +57,17 @@ Status ClientStatusDiscoveryService::StreamClientStatus(
ServerReaderWriter<ClientStatusResponse, ClientStatusRequest>* stream) {
ClientStatusRequest request;
while (stream->Read(&request)) {
ClientStatusResponse response;
absl::StatusOr<ClientConfig> s = DumpClientConfig();
if (!s.ok()) {
if (s.status().code() == absl::StatusCode::kUnavailable) {
absl::StatusOr<ClientStatusResponse> response = DumpClientStatusResponse();
if (!response.ok()) {
if (response.status().code() == absl::StatusCode::kUnavailable) {
// If the xDS client is not initialized, return empty response
stream->Write(response);
stream->Write(ClientStatusResponse());
continue;
}
return Status(static_cast<StatusCode>(s.status().raw_code()),
s.status().ToString());
return Status(static_cast<StatusCode>(response.status().raw_code()),
response.status().ToString());
}
*response.add_config() = std::move(s.value());
stream->Write(response);
stream->Write(*response);
}
return Status::OK;
}
@ -78,7 +75,7 @@ Status ClientStatusDiscoveryService::StreamClientStatus(
Status ClientStatusDiscoveryService::FetchClientStatus(
ServerContext* /*context*/, const ClientStatusRequest* /*request*/,
ClientStatusResponse* response) {
absl::StatusOr<ClientConfig> s = DumpClientConfig();
absl::StatusOr<ClientStatusResponse> s = DumpClientStatusResponse();
if (!s.ok()) {
if (s.status().code() == absl::StatusCode::kUnavailable) {
// If the xDS client is not initialized, return empty response
@ -87,7 +84,7 @@ Status ClientStatusDiscoveryService::FetchClientStatus(
return Status(static_cast<StatusCode>(s.status().raw_code()),
s.status().ToString());
}
*response->add_config() = std::move(s.value());
*response = std::move(*s);
return Status::OK;
}

5
src/python/grpcio_csds/grpc_csds/__init__.py
Unescape View File

@ -26,12 +26,9 @@ class ClientStatusDiscoveryServiceServicer(
@staticmethod
def FetchClientStatus(request, unused_context):
client_config = csds_pb2.ClientConfig.FromString(
return csds_pb2.ClientStatusResponse.FromString(
cygrpc.dump_xds_configs()
)
response = csds_pb2.ClientStatusResponse()
response.config.append(client_config)
return response
@staticmethod
def StreamClientStatus(request_iterator, context):

2
src/python/grpcio_tests/tests/admin/admin_test.py
Unescape View File

@ -46,7 +46,7 @@ class TestAdmin(unittest.TestCase):
stub = csds_pb2_grpc.ClientStatusDiscoveryServiceStub(self._channel)
resp = stub.FetchClientStatus(csds_pb2.ClientStatusRequest())
# No exception raised and the response is valid
self.assertGreater(len(resp.config), 0)
self.assertEqual(len(resp.config), 0)
def test_has_channelz(self):
stub = channelz_pb2_grpc.ChannelzStub(self._channel)

10
src/python/grpcio_tests/tests/csds/csds_test.py
Unescape View File

@ -81,12 +81,6 @@ class TestCsds(unittest.TestCase):
def get_xds_config_dump(self):
return self._stub.FetchClientStatus(csds_pb2.ClientStatusRequest())
def test_has_node(self):
resp = self.get_xds_config_dump()
self.assertEqual(1, len(resp.config))
self.assertEqual("python_test_csds", resp.config[0].node.id)
self.assertEqual("test", resp.config[0].node.cluster)
def test_no_lds_found(self):
dummy_channel = grpc.insecure_channel(_DUMMY_XDS_ADDRESS)
@ -100,6 +94,10 @@ class TestCsds(unittest.TestCase):
# The resource request will fail with DOES_NOT_EXIST (after 15s)
while True:
resp = self.get_xds_config_dump()
# Check node is setup in the CSDS response
self.assertEqual(1, len(resp.config))
self.assertEqual("python_test_csds", resp.config[0].node.id)
self.assertEqual("test", resp.config[0].node.cluster)
config = json_format.MessageToDict(resp)
ok = False
try:

22
test/core/xds/xds_client_fuzzer.cc
Unescape View File

@ -47,6 +47,26 @@
namespace grpc_core {
namespace testing {
class XdsClientTestPeer {
public:
explicit XdsClientTestPeer(XdsClient* xds_client) : xds_client_(xds_client) {}
void TestDumpClientConfig() {
upb::Arena arena;
auto client_config = envoy_service_status_v3_ClientConfig_new(arena.ptr());
std::set<std::string> string_pool;
MutexLock lock(xds_client_->mu());
xds_client_->DumpClientConfig(&string_pool, arena.ptr(), client_config);
}
private:
XdsClient* xds_client_;
};
} // namespace testing
class Fuzzer {
public:
explicit Fuzzer(absl::string_view bootstrap_json) {
@ -113,7 +133,7 @@ class Fuzzer {
}
break;
case xds_client_fuzzer::Action::kDumpCsdsData:
xds_client_->DumpClientConfigBinary();
testing::XdsClientTestPeer(xds_client_.get()).TestDumpClientConfig();
break;
case xds_client_fuzzer::Action::kTriggerConnectionFailure:
TriggerConnectionFailure(

Write Preview
Loading…
Cancel
Save