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

[grpc][Gpr_To_Absl_Logging] Migrating from gpr to absl logging - gpr_log (#36942)

Browse Source 
[grpc][Gpr_To_Absl_Logging] Migrating from gpr to absl logging - gpr_log
In this CL we are migrating from gRPCs own gpr logging mechanism to absl logging mechanism. The intention is to deprecate gpr_log in the future.

We have the following mapping

1. gpr_log(GPR_INFO,...) -> LOG(INFO)
2. gpr_log(GPR_ERROR,...) -> LOG(ERROR)
3. gpr_log(GPR_DEBUG,...) -> VLOG(2)

Reviewers need to check :

1. If the above mapping is correct.
2. The content of the log is as before.
gpr_log format strings did not use string_view or std::string . absl LOG accepts these. So there will be some elimination of string_view and std::string related conversions. This is expected.

Closes #36942

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/36942 from tanvi-jagtap:src_core_lb 524b45fef1
PiperOrigin-RevId: 646311061
pull/36971/merge
Tanvi Jagtap 5 months ago committed by Copybara-Service
parent 1913e16526
commit 68c7d72f5c
9 changed files with 407 additions and 484 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. 110
      src/core/load_balancing/grpclb/grpclb.cc
  2. 291
      src/core/load_balancing/pick_first/pick_first.cc
  3. 128
      src/core/load_balancing/priority/priority.cc
  4. 16
      src/core/load_balancing/ring_hash/ring_hash.cc
  5. 195
      src/core/load_balancing/rls/rls.cc
  6. 117
      src/core/load_balancing/weighted_round_robin/weighted_round_robin.cc
  7. 19
      src/core/load_balancing/xds/cds.cc
  8. 7
      src/core/load_balancing/xds/xds_cluster_manager.cc
  9. 8
      tools/run_tests/sanity/banned_functions.py

110
src/core/load_balancing/grpclb/grpclb.cc
Unescape View File

@ -93,7 +93,6 @@
#include <grpc/slice.h>
#include <grpc/status.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include "src/core/channelz/channelz.h"
#include "src/core/client_channel/client_channel_filter.h"
@ -497,10 +496,9 @@ class GrpcLb final : public LoadBalancingPolicy {
new_state == GRPC_CHANNEL_TRANSIENT_FAILURE) {
// In TRANSIENT_FAILURE. Cancel the fallback timer and go into
// fallback mode immediately.
gpr_log(GPR_INFO,
"[grpclb %p] balancer channel in state:TRANSIENT_FAILURE (%s); "
"entering fallback mode",
parent_.get(), status.ToString().c_str());
LOG(INFO) << "[grpclb " << parent_.get()
<< "] balancer channel in state:TRANSIENT_FAILURE ("
<< status.ToString() << "); entering fallback mode";
parent_->fallback_at_startup_checks_pending_ = false;
parent_->channel_control_helper()->GetEventEngine()->Cancel(
*parent_->lb_fallback_timer_handle_);
@ -852,12 +850,11 @@ void GrpcLb::Helper::UpdateState(grpc_connectivity_state state,
client_stats = parent()->lb_calld_->client_stats()->Ref();
}
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p helper %p] state=%s (%s) wrapping child "
"picker %p (serverlist=%p, client_stats=%p)",
parent(), this, ConnectivityStateName(state),
status.ToString().c_str(), picker.get(), serverlist.get(),
client_stats.get());
LOG(INFO) << "[grpclb " << parent() << " helper " << this
<< "] state=" << ConnectivityStateName(state) << " ("
<< status.ToString() << ") wrapping child picker " << picker.get()
<< " (serverlist=" << serverlist.get()
<< ", client_stats=" << client_stats.get() << ")";
}
parent()->channel_control_helper()->UpdateState(
state, status,
@ -949,8 +946,8 @@ void GrpcLb::BalancerCallState::Orphan() {
void GrpcLb::BalancerCallState::StartQuery() {
CHECK_NE(lb_call_, nullptr);
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO, "[grpclb %p] lb_calld=%p: Starting LB call %p",
grpclb_policy_.get(), this, lb_call_);
LOG(INFO) << "[grpclb " << grpclb_policy_.get() << "] lb_calld=" << this
<< ": Starting LB call " << lb_call_;
}
// Create the ops.
grpc_call_error call_error;
@ -1176,18 +1173,16 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
client_stats_report_interval_ = std::max(
Duration::Seconds(1), response.client_stats_report_interval);
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] lb_calld=%p: Received initial LB response "
"message; client load reporting interval = %" PRId64
" milliseconds",
grpclb_policy(), this,
client_stats_report_interval_.millis());
LOG(INFO) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
<< ": Received initial LB response message; client load "
"reporting interval = "
<< client_stats_report_interval_.millis()
<< " milliseconds";
}
} else if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] lb_calld=%p: Received initial LB response "
"message; client load reporting NOT enabled",
grpclb_policy(), this);
LOG(INFO) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
<< ": Received initial LB response message; client load "
"reporting NOT enabled";
}
seen_initial_response_ = true;
break;
@ -1197,12 +1192,11 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
auto serverlist_wrapper =
MakeRefCounted<Serverlist>(std::move(response.serverlist));
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] lb_calld=%p: Serverlist with %" PRIuPTR
" servers received:\n%s",
grpclb_policy(), this,
serverlist_wrapper->serverlist().size(),
serverlist_wrapper->AsText().c_str());
LOG(INFO) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
<< ": Serverlist with "
<< serverlist_wrapper->serverlist().size()
<< " servers received:\n"
<< serverlist_wrapper->AsText();
}
seen_serverlist_ = true;
// Start sending client load report only after we start using the
@ -1218,10 +1212,9 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
if (grpclb_policy()->serverlist_ != nullptr &&
*grpclb_policy()->serverlist_ == *serverlist_wrapper) {
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] lb_calld=%p: Incoming server list identical "
"to current, ignoring.",
grpclb_policy(), this);
LOG(INFO) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
<< ": Incoming server list identical to current, "
"ignoring.";
}
} else { // New serverlist.
// Dispose of the fallback.
@ -1244,10 +1237,9 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
// it in favor of the xds policy. We will implement this the
// right way in the xds policy instead.
if (grpclb_policy()->fallback_mode_) {
gpr_log(GPR_INFO,
"[grpclb %p] Received response from balancer; exiting "
"fallback mode",
grpclb_policy());
LOG(INFO) << "[grpclb " << grpclb_policy()
<< "] Received response from balancer; exiting fallback "
"mode";
grpclb_policy()->fallback_mode_ = false;
}
if (grpclb_policy()->fallback_at_startup_checks_pending_) {
@ -1266,9 +1258,8 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked() {
}
case response.FALLBACK: {
if (!grpclb_policy()->fallback_mode_) {
gpr_log(GPR_INFO,
"[grpclb %p] Entering fallback mode as requested by balancer",
grpclb_policy());
LOG(INFO) << "[grpclb " << grpclb_policy()
<< "] Entering fallback mode as requested by balancer";
if (grpclb_policy()->fallback_at_startup_checks_pending_) {
grpclb_policy()->fallback_at_startup_checks_pending_ = false;
grpclb_policy()->channel_control_helper()->GetEventEngine()->Cancel(
@ -1317,11 +1308,11 @@ void GrpcLb::BalancerCallState::OnBalancerStatusReceivedLocked(
CHECK_NE(lb_call_, nullptr);
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
char* status_details = grpc_slice_to_c_string(lb_call_status_details_);
gpr_log(GPR_INFO,
"[grpclb %p] lb_calld=%p: Status from LB server received. "
"Status = %d, details = '%s', (lb_call: %p), error '%s'",
grpclb_policy(), this, lb_call_status_, status_details, lb_call_,
StatusToString(error).c_str());
LOG(INFO) << "[grpclb " << grpclb_policy() << "] lb_calld=" << this
<< ": Status from LB server received. Status = "
<< lb_call_status_ << ", details = '" << status_details
<< "', (lb_call: " << lb_call_ << "), error '"
<< StatusToString(error) << "'";
gpr_free(status_details);
}
// If this lb_calld is still in use, this call ended because of a failure so
@ -1334,10 +1325,9 @@ void GrpcLb::BalancerCallState::OnBalancerStatusReceivedLocked(
grpclb_policy()->lb_calld_.reset();
if (grpclb_policy()->fallback_at_startup_checks_pending_) {
CHECK(!seen_serverlist_);
gpr_log(GPR_INFO,
"[grpclb %p] Balancer call finished without receiving "
"serverlist; entering fallback mode",
grpclb_policy());
LOG(INFO) << "[grpclb " << grpclb_policy()
<< "] Balancer call finished without receiving serverlist; "
"entering fallback mode";
grpclb_policy()->fallback_at_startup_checks_pending_ = false;
grpclb_policy()->channel_control_helper()->GetEventEngine()->Cancel(
*grpclb_policy()->lb_fallback_timer_handle_);
@ -1666,9 +1656,9 @@ void GrpcLb::StartBalancerCallLocked() {
CHECK(lb_calld_ == nullptr);
lb_calld_ = MakeOrphanable<BalancerCallState>(Ref());
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] Query for backends (lb_channel: %p, lb_calld: %p)",
this, lb_channel_.get(), lb_calld_.get());
LOG(INFO) << "[grpclb " << this
<< "] Query for backends (lb_channel: " << lb_channel_.get()
<< ", lb_calld: " << lb_calld_.get() << ")";
}
lb_calld_->StartQuery();
}
@ -1678,8 +1668,8 @@ void GrpcLb::StartBalancerCallRetryTimerLocked() {
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
LOG(INFO) << "[grpclb " << this << "] Connection to LB server lost...";
if (timeout > Duration::Zero()) {
gpr_log(GPR_INFO, "[grpclb %p] ... retry_timer_active in %" PRId64 "ms.",
this, timeout.millis());
LOG(INFO) << "[grpclb " << this << "] ... retry_timer_active in "
<< timeout.millis() << "ms.";
} else {
LOG(INFO) << "[grpclb " << this
<< "] ... retry_timer_active immediately.";
@ -1724,10 +1714,9 @@ void GrpcLb::MaybeEnterFallbackModeAfterStartup() {
if (!fallback_mode_ && !fallback_at_startup_checks_pending_ &&
(lb_calld_ == nullptr || !lb_calld_->seen_serverlist()) &&
!child_policy_ready_) {
gpr_log(GPR_INFO,
"[grpclb %p] lost contact with balancer and backends from "
"most recent serverlist; entering fallback mode",
this);
LOG(INFO) << "[grpclb " << this
<< "] lost contact with balancer and backends from most recent "
"serverlist; entering fallback mode";
fallback_mode_ = true;
CreateOrUpdateChildPolicyLocked();
}
@ -1873,9 +1862,8 @@ void GrpcLb::OnSubchannelCacheTimerLocked() {
auto it = cached_subchannels_.begin();
if (it != cached_subchannels_.end()) {
if (GRPC_TRACE_FLAG_ENABLED(glb)) {
gpr_log(GPR_INFO,
"[grpclb %p] removing %" PRIuPTR " subchannels from cache",
this, it->second.size());
LOG(INFO) << "[grpclb " << this << "] removing " << it->second.size()
<< " subchannels from cache";
}
cached_subchannels_.erase(it);
}

291
src/core/load_balancing/pick_first/pick_first.cc
Unescape View File

@ -39,7 +39,6 @@
#include <grpc/event_engine/event_engine.h>
#include <grpc/impl/channel_arg_names.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/address_utils/sockaddr_utils.h"
@ -633,9 +632,8 @@ void PickFirst::HealthWatcher::OnConnectivityStateChange(
grpc_connectivity_state new_state, absl::Status status) {
if (policy_->health_watcher_ != this) return;
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] health watch state update: %s (%s)",
policy_.get(), ConnectivityStateName(new_state),
status.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] health watch state update: "
<< ConnectivityStateName(new_state) << " (" << status << ")";
}
switch (new_state) {
case GRPC_CHANNEL_READY:
@ -675,9 +673,8 @@ PickFirst::SubchannelList::SubchannelData::SubchannelState::SubchannelState(
pick_first_(subchannel_data_->subchannel_list_->policy_),
subchannel_(std::move(subchannel)) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel state %p (subchannel %p): starting watch",
pick_first_.get(), this, subchannel_.get());
LOG(INFO) << "[PF " << pick_first_.get() << "] subchannel state " << this
<< " (subchannel " << subchannel_.get() << "): starting watch";
}
auto watcher = std::make_unique<Watcher>(Ref(DEBUG_LOCATION, "Watcher"));
watcher_ = watcher.get();
@ -686,10 +683,9 @@ PickFirst::SubchannelList::SubchannelData::SubchannelState::SubchannelState(
void PickFirst::SubchannelList::SubchannelData::SubchannelState::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel state %p (subchannel %p): "
"cancelling watch and unreffing subchannel",
pick_first_.get(), this, subchannel_.get());
LOG(INFO) << "[PF " << pick_first_.get() << "] subchannel state " << this
<< " (subchannel " << subchannel_.get()
<< "): cancelling watch and unreffing subchannel";
}
subchannel_data_ = nullptr;
subchannel_->CancelConnectivityStateWatch(watcher_);
@ -701,8 +697,8 @@ void PickFirst::SubchannelList::SubchannelData::SubchannelState::Orphan() {
void PickFirst::SubchannelList::SubchannelData::SubchannelState::Select() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "Pick First %p selected subchannel %p", pick_first_.get(),
subchannel_.get());
LOG(INFO) << "Pick First " << pick_first_.get() << " selected subchannel "
<< subchannel_.get();
}
CHECK_NE(subchannel_data_, nullptr);
pick_first_->UnsetSelectedSubchannel(); // Cancel health watch, if any.
@ -713,7 +709,7 @@ void PickFirst::SubchannelList::SubchannelData::SubchannelState::Select() {
// If health checking is NOT enabled, report READY.
if (pick_first_->enable_health_watch_) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] starting health watch", pick_first_.get());
LOG(INFO) << "[PF " << pick_first_.get() << "] starting health watch";
}
auto watcher = std::make_unique<HealthWatcher>(
pick_first_.Ref(DEBUG_LOCATION, "HealthWatcher"));
@ -751,13 +747,13 @@ void PickFirst::SubchannelList::SubchannelData::SubchannelState::
absl::Status status) {
if (watcher_ == nullptr) return;
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel state %p (subchannel %p): connectivity "
"changed: new_state=%s, status=%s, watcher=%p, "
"subchannel_data_=%p, pick_first_->selected_=%p",
pick_first_.get(), this, subchannel_.get(),
ConnectivityStateName(new_state), status.ToString().c_str(),
watcher_, subchannel_data_, pick_first_->selected_.get());
LOG(INFO) << "[PF " << pick_first_.get() << "] subchannel state " << this
<< " (subchannel " << subchannel_.get()
<< "): connectivity changed: new_state="
<< ConnectivityStateName(new_state) << ", status=" << status
<< ", watcher=" << watcher_
<< ", subchannel_data_=" << subchannel_data_
<< ", pick_first_->selected_=" << pick_first_->selected_.get();
}
// If we're still part of a subchannel list trying to connect, check
// if we're connected.
@ -774,9 +770,9 @@ void PickFirst::SubchannelList::SubchannelData::SubchannelState::
// We aren't trying to connect, so we must be the selected subchannel.
CHECK(pick_first_->selected_.get() == this);
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p selected subchannel connectivity changed to %s",
pick_first_.get(), ConnectivityStateName(new_state));
LOG(INFO) << "Pick First " << pick_first_.get()
<< " selected subchannel connectivity changed to "
<< ConnectivityStateName(new_state);
}
// Any state change is considered to be a failure of the existing
// connection. Report the failure.
@ -798,10 +794,9 @@ PickFirst::SubchannelList::SubchannelData::SubchannelData(
RefCountedPtr<SubchannelInterface> subchannel)
: subchannel_list_(subchannel_list), index_(index) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR
": creating subchannel data",
subchannel_list_->policy_.get(), subchannel_list_, index_);
LOG(INFO) << "[PF " << subchannel_list_->policy_.get()
<< "] subchannel list " << subchannel_list_ << " index " << index_
<< ": creating subchannel data";
}
subchannel_state_ =
MakeOrphanable<SubchannelState>(this, std::move(subchannel));
@ -811,20 +806,20 @@ void PickFirst::SubchannelList::SubchannelData::OnConnectivityStateChange(
grpc_connectivity_state new_state, absl::Status status) {
PickFirst* p = subchannel_list_->policy_.get();
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(
GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR
" (subchannel_state %p): connectivity changed: old_state=%s, "
"new_state=%s, status=%s, seen_transient_failure=%d, p->selected_=%p, "
"p->subchannel_list_=%p, p->subchannel_list_->shutting_down_=%d",
p, subchannel_list_, index_, subchannel_list_->size(),
subchannel_state_.get(),
(connectivity_state_.has_value()
? ConnectivityStateName(*connectivity_state_)
: "N/A"),
ConnectivityStateName(new_state), status.ToString().c_str(),
seen_transient_failure_, p->selected_.get(), p->subchannel_list_.get(),
p->subchannel_list_->shutting_down_);
LOG(INFO) << "[PF " << p << "] subchannel list " << subchannel_list_
<< " index " << index_ << " of " << subchannel_list_->size()
<< " (subchannel_state " << subchannel_state_.get()
<< "): connectivity changed: old_state="
<< (connectivity_state_.has_value()
? ConnectivityStateName(*connectivity_state_)
: "N/A")
<< ", new_state=" << ConnectivityStateName(new_state)
<< ", status=" << status
<< ", seen_transient_failure=" << seen_transient_failure_
<< ", p->selected_=" << p->selected_.get()
<< ", p->subchannel_list_=" << p->subchannel_list_.get()
<< ", p->subchannel_list_->shutting_down_="
<< p->subchannel_list_->shutting_down_;
}
if (subchannel_list_->shutting_down_) return;
// The notification must be for a subchannel in the current list.
@ -864,10 +859,9 @@ void PickFirst::SubchannelList::SubchannelData::OnConnectivityStateChange(
// connection and report IDLE.
if (p->selected_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p: new update has no subchannels in "
"state READY; dropping existing connection and going IDLE",
p, subchannel_list_);
LOG(INFO) << "[PF " << p << "] subchannel list " << subchannel_list_
<< ": new update has no subchannels in "
<< "state READY; dropping existing connection and going IDLE";
}
p->GoIdle();
} else {
@ -965,11 +959,10 @@ void PickFirst::SubchannelList::SubchannelData::RequestConnectionWithTimer() {
if (index_ != subchannel_list_->size() - 1) {
PickFirst* p = subchannel_list_->policy_.get();
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p: starting Connection "
"Attempt Delay timer for %" PRId64 "ms for index %" PRIuPTR,
p, subchannel_list_, p->connection_attempt_delay_.millis(),
index_);
LOG(INFO) << "Pick First " << p << " subchannel list " << subchannel_list_
<< ": starting Connection Attempt Delay timer for "
<< p->connection_attempt_delay_.millis() << "ms for index "
<< index_;
}
subchannel_list_->timer_handle_ =
p->channel_control_helper()->GetEventEngine()->RunAfter(
@ -982,14 +975,13 @@ void PickFirst::SubchannelList::SubchannelData::RequestConnectionWithTimer() {
sl->policy_->work_serializer()->Run(
[subchannel_list = std::move(subchannel_list)]() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p: Connection "
"Attempt Delay timer fired (shutting_down=%d, "
"selected=%p)",
subchannel_list->policy_.get(),
subchannel_list.get(),
subchannel_list->shutting_down_,
subchannel_list->policy_->selected_.get());
LOG(INFO)
<< "Pick First " << subchannel_list->policy_.get()
<< " subchannel list " << subchannel_list.get()
<< ": Connection Attempt Delay timer fired "
"(shutting_down="
<< subchannel_list->shutting_down_ << ", selected="
<< subchannel_list->policy_->selected_.get() << ")";
}
if (subchannel_list->shutting_down_) return;
if (subchannel_list->policy_->selected_ != nullptr) return;
@ -1015,8 +1007,8 @@ PickFirst::SubchannelList::SubchannelList(RefCountedPtr<PickFirst> policy,
.Remove(
GRPC_ARG_INTERNAL_PICK_FIRST_OMIT_STATUS_MESSAGE_PREFIX)) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Creating subchannel list %p - channel args: %s",
policy_.get(), this, args_.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] Creating subchannel list "
<< this << " - channel args: " << args_.ToString();
}
if (addresses == nullptr) return;
// Create a subchannel for each address.
@ -1028,18 +1020,16 @@ PickFirst::SubchannelList::SubchannelList(RefCountedPtr<PickFirst> policy,
if (subchannel == nullptr) {
// Subchannel could not be created.
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] could not create subchannel for address %s, ignoring",
policy_.get(), address.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get()
<< "] could not create subchannel for address "
<< address.ToString() << ", ignoring";
}
return;
}
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR
": Created subchannel %p for address %s",
policy_.get(), this, subchannels_.size(), subchannel.get(),
address.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] subchannel list " << this
<< " index " << subchannels_.size() << ": Created subchannel "
<< subchannel.get() << " for address " << address.ToString();
}
subchannels_.emplace_back(std::make_unique<SubchannelData>(
this, subchannels_.size(), std::move(subchannel)));
@ -1048,15 +1038,15 @@ PickFirst::SubchannelList::SubchannelList(RefCountedPtr<PickFirst> policy,
PickFirst::SubchannelList::~SubchannelList() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Destroying subchannel_list %p", policy_.get(),
this);
LOG(INFO) << "[PF " << policy_.get() << "] Destroying subchannel_list "
<< this;
}
}
void PickFirst::SubchannelList::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Shutting down subchannel_list %p", policy_.get(),
this);
LOG(INFO) << "[PF " << policy_.get() << "] Shutting down subchannel_list "
<< this;
}
CHECK(!shutting_down_);
shutting_down_ = true;
@ -1100,10 +1090,8 @@ void PickFirst::SubchannelList::MaybeFinishHappyEyeballsPass() {
// so report TRANSIENT_FAILURE and switch to a mode in which we try to
// connect to all addresses in parallel.
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p failed to connect to "
"all subchannels",
policy_.get(), this);
LOG(INFO) << "Pick First " << policy_.get() << " subchannel list " << this
<< " failed to connect to all subchannels";
}
// Re-resolve and report TRANSIENT_FAILURE.
policy_->channel_control_helper()->RequestReresolution();
@ -1577,9 +1565,8 @@ void OldPickFirst::HealthWatcher::OnConnectivityStateChange(
grpc_connectivity_state new_state, absl::Status status) {
if (policy_->health_watcher_ != this) return;
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] health watch state update: %s (%s)",
policy_.get(), ConnectivityStateName(new_state),
status.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] health watch state update: "
<< ConnectivityStateName(new_state) << " (" << status << ")";
}
switch (new_state) {
case GRPC_CHANNEL_READY:
@ -1619,11 +1606,9 @@ OldPickFirst::SubchannelList::SubchannelData::SubchannelData(
index_(index),
subchannel_(std::move(subchannel)) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR
" (subchannel %p): starting watch",
subchannel_list_->policy_.get(), subchannel_list_, index_,
subchannel_.get());
LOG(INFO) << "[PF " << subchannel_list_->policy_.get()
<< "] subchannel list " << subchannel_list_ << " index " << index_
<< " (subchannel " << subchannel_.get() << "): starting watch";
}
auto watcher = std::make_unique<Watcher>(
subchannel_list_->Ref(DEBUG_LOCATION, "Watcher"), index_);
@ -1634,11 +1619,11 @@ OldPickFirst::SubchannelList::SubchannelData::SubchannelData(
void OldPickFirst::SubchannelList::SubchannelData::ShutdownLocked() {
if (subchannel_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR
" (subchannel %p): cancelling watch and unreffing subchannel",
subchannel_list_->policy_.get(), subchannel_list_, index_,
subchannel_list_->size(), subchannel_.get());
LOG(INFO) << "[PF " << subchannel_list_->policy_.get()
<< "] subchannel list " << subchannel_list_ << " index "
<< index_ << " of " << subchannel_list_->size()
<< " (subchannel " << subchannel_.get()
<< "): cancelling watch and unreffing subchannel";
}
subchannel_->CancelConnectivityStateWatch(pending_watcher_);
pending_watcher_ = nullptr;
@ -1650,22 +1635,22 @@ void OldPickFirst::SubchannelList::SubchannelData::OnConnectivityStateChange(
grpc_connectivity_state new_state, absl::Status status) {
OldPickFirst* p = subchannel_list_->policy_.get();
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(
GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR " of %" PRIuPTR
" (subchannel %p): connectivity changed: old_state=%s, new_state=%s, "
"status=%s, shutting_down=%d, pending_watcher=%p, "
"seen_transient_failure=%d, p->selected_=%p, "
"p->subchannel_list_=%p, p->latest_pending_subchannel_list_=%p",
p, subchannel_list_, index_, subchannel_list_->size(),
subchannel_.get(),
(connectivity_state_.has_value()
? ConnectivityStateName(*connectivity_state_)
: "N/A"),
ConnectivityStateName(new_state), status.ToString().c_str(),
subchannel_list_->shutting_down_, pending_watcher_,
seen_transient_failure_, p->selected_, p->subchannel_list_.get(),
p->latest_pending_subchannel_list_.get());
LOG(INFO) << "[PF " << p << "] subchannel list " << subchannel_list_
<< " index " << index_ << " of " << subchannel_list_->size()
<< " (subchannel " << subchannel_.get()
<< "): connectivity changed: old_state="
<< (connectivity_state_.has_value()
? ConnectivityStateName(*connectivity_state_)
: "N/A")
<< ", new_state=" << ConnectivityStateName(new_state)
<< ", status=" << status
<< ", shutting_down=" << subchannel_list_->shutting_down_
<< ", pending_watcher=" << pending_watcher_
<< ", seen_transient_failure=" << seen_transient_failure_
<< ", p->selected_=" << p->selected_
<< ", p->subchannel_list_=" << p->subchannel_list_.get()
<< ", p->latest_pending_subchannel_list_="
<< p->latest_pending_subchannel_list_.get();
}
if (subchannel_list_->shutting_down_ || pending_watcher_ == nullptr) return;
auto& stats_plugins = subchannel_list_->policy_->channel_control_helper()
@ -1682,9 +1667,9 @@ void OldPickFirst::SubchannelList::SubchannelData::OnConnectivityStateChange(
if (p->selected_ == this) {
CHECK(subchannel_list_ == p->subchannel_list_.get());
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p selected subchannel connectivity changed to %s", p,
ConnectivityStateName(new_state));
LOG(INFO) << "Pick First " << p
<< " selected subchannel connectivity changed to "
<< ConnectivityStateName(new_state);
}
// Any state change is considered to be a failure of the existing
// connection.
@ -1701,11 +1686,9 @@ void OldPickFirst::SubchannelList::SubchannelData::OnConnectivityStateChange(
// If there is a pending update, switch to the pending update.
if (p->latest_pending_subchannel_list_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p promoting pending subchannel list %p to "
"replace %p",
p, p->latest_pending_subchannel_list_.get(),
p->subchannel_list_.get());
LOG(INFO) << "Pick First " << p << " promoting pending subchannel "
<< "list " << p->latest_pending_subchannel_list_.get()
<< " to replace " << p->subchannel_list_.get();
}
p->UnsetSelectedSubchannel();
p->subchannel_list_ = std::move(p->latest_pending_subchannel_list_);
@ -1865,11 +1848,10 @@ void OldPickFirst::SubchannelList::SubchannelData::
if (index_ != subchannel_list_->size() - 1) {
OldPickFirst* p = subchannel_list_->policy_.get();
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p: starting Connection "
"Attempt Delay timer for %" PRId64 "ms for index %" PRIuPTR,
p, subchannel_list_, p->connection_attempt_delay_.millis(),
index_);
LOG(INFO) << "Pick First " << p << " subchannel list " << subchannel_list_
<< ": starting Connection Attempt Delay timer for "
<< p->connection_attempt_delay_.millis() << "ms for index "
<< index_;
}
subchannel_list_->timer_handle_ =
p->channel_control_helper()->GetEventEngine()->RunAfter(
@ -1882,14 +1864,13 @@ void OldPickFirst::SubchannelList::SubchannelData::
sl->policy_->work_serializer()->Run(
[subchannel_list = std::move(subchannel_list)]() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p: Connection "
"Attempt Delay timer fired (shutting_down=%d, "
"selected=%p)",
subchannel_list->policy_.get(),
subchannel_list.get(),
subchannel_list->shutting_down_,
subchannel_list->policy_->selected_);
LOG(INFO)
<< "Pick First " << subchannel_list->policy_.get()
<< " subchannel list " << subchannel_list.get()
<< ": Connection Attempt Delay timer fired "
<< "(shutting_down="
<< subchannel_list->shutting_down_ << ", selected="
<< subchannel_list->policy_->selected_ << ")";
}
if (subchannel_list->shutting_down_) return;
if (subchannel_list->policy_->selected_ != nullptr) return;
@ -1923,19 +1904,17 @@ void OldPickFirst::SubchannelList::SubchannelData::
// Case 2. Promote p->latest_pending_subchannel_list_ to p->subchannel_list_.
if (subchannel_list_ == p->latest_pending_subchannel_list_.get()) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p promoting pending subchannel list %p to "
"replace %p",
p, p->latest_pending_subchannel_list_.get(),
p->subchannel_list_.get());
LOG(INFO) << "Pick First " << p << " promoting pending subchannel list "
<< p->latest_pending_subchannel_list_.get() << " to replace "
<< p->subchannel_list_.get();
}
p->UnsetSelectedSubchannel();
p->subchannel_list_ = std::move(p->latest_pending_subchannel_list_);
}
// Cases 1 and 2.
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "Pick First %p selected subchannel %p", p,
subchannel_.get());
LOG(INFO) << "Pick First " << p << " selected subchannel "
<< subchannel_.get();
}
p->selected_ = this;
// If health checking is enabled, start the health watch, but don't
@ -1944,7 +1923,7 @@ void OldPickFirst::SubchannelList::SubchannelData::
// If health checking is NOT enabled, report READY.
if (p->enable_health_watch_) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] starting health watch", p);
LOG(INFO) << "[PF " << p << "] starting health watch";
}
auto watcher = std::make_unique<HealthWatcher>(
p->RefAsSubclass<OldPickFirst>(DEBUG_LOCATION, "HealthWatcher"));
@ -1979,8 +1958,8 @@ OldPickFirst::SubchannelList::SubchannelList(
.Remove(
GRPC_ARG_INTERNAL_PICK_FIRST_OMIT_STATUS_MESSAGE_PREFIX)) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Creating subchannel list %p - channel args: %s",
policy_.get(), this, args_.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] Creating subchannel list "
<< this << " - channel args: " << args_.ToString();
}
if (addresses == nullptr) return;
// Create a subchannel for each address.
@ -1992,18 +1971,16 @@ OldPickFirst::SubchannelList::SubchannelList(
if (subchannel == nullptr) {
// Subchannel could not be created.
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] could not create subchannel for address %s, ignoring",
policy_.get(), address.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get()
<< "] could not create subchannel for address "
<< address.ToString() << ", ignoring";
}
return;
}
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"[PF %p] subchannel list %p index %" PRIuPTR
": Created subchannel %p for address %s",
policy_.get(), this, subchannels_.size(), subchannel.get(),
address.ToString().c_str());
LOG(INFO) << "[PF " << policy_.get() << "] subchannel list " << this
<< " index " << subchannels_.size() << ": Created subchannel "
<< subchannel.get() << " for address " << address.ToString();
}
subchannels_.emplace_back(this, subchannels_.size(), std::move(subchannel));
});
@ -2011,15 +1988,15 @@ OldPickFirst::SubchannelList::SubchannelList(
OldPickFirst::SubchannelList::~SubchannelList() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Destroying subchannel_list %p", policy_.get(),
this);
LOG(INFO) << "[PF " << policy_.get() << "] Destroying subchannel_list "
<< this;
}
}
void OldPickFirst::SubchannelList::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO, "[PF %p] Shutting down subchannel_list %p", policy_.get(),
this);
LOG(INFO) << "[PF " << policy_.get() << "] Shutting down subchannel_list "
<< this;
}
CHECK(!shutting_down_);
shutting_down_ = true;
@ -2065,21 +2042,18 @@ void OldPickFirst::SubchannelList::MaybeFinishHappyEyeballsPass() {
// so report TRANSIENT_FAILURE and switch to a mode in which we try to
// connect to all addresses in parallel.
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p subchannel list %p failed to connect to "
"all subchannels",
policy_.get(), this);
LOG(INFO) << "Pick First " << policy_.get() << " subchannel list " << this
<< " failed to connect to all subchannels";
}
// In case 2, swap to the new subchannel list. This means reporting
// TRANSIENT_FAILURE and dropping the existing (working) connection,
// but we can't ignore what the control plane has told us.
if (policy_->latest_pending_subchannel_list_.get() == this) {
if (GRPC_TRACE_FLAG_ENABLED(pick_first)) {
gpr_log(GPR_INFO,
"Pick First %p promoting pending subchannel list %p to "
"replace %p",
policy_.get(), policy_->latest_pending_subchannel_list_.get(),
this);
LOG(INFO) << "Pick First " << policy_.get()
<< " promoting pending subchannel list "
<< policy_->latest_pending_subchannel_list_.get()
<< " to replace " << this;
}
policy_->UnsetSelectedSubchannel();
policy_->subchannel_list_ =
@ -2108,7 +2082,6 @@ void OldPickFirst::SubchannelList::MaybeFinishHappyEyeballsPass() {
}
}
}
//
// factory
//

128
src/core/load_balancing/priority/priority.cc
Unescape View File

@ -37,7 +37,6 @@
#include <grpc/event_engine/event_engine.h>
#include <grpc/impl/channel_arg_names.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/channel/channel_args.h"
@ -318,9 +317,9 @@ void PriorityLb::ExitIdleLocked() {
if (current_priority_ != UINT32_MAX) {
const std::string& child_name = config_->priorities()[current_priority_];
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] exiting IDLE for current priority %d child %s",
this, current_priority_, child_name.c_str());
LOG(INFO) << "[priority_lb " << this
<< "] exiting IDLE for current priority " << current_priority_
<< " child " << child_name;
}
children_[child_name]->ExitIdleLocked();
}
@ -404,8 +403,8 @@ void PriorityLb::ChoosePriorityLocked() {
// If the child for the priority does not exist yet, create it.
const std::string& child_name = config_->priorities()[priority];
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] trying priority %u, child %s", this,
priority, child_name.c_str());
LOG(INFO) << "[priority_lb " << this << "] trying priority " << priority
<< ", child " << child_name;
}
auto& child = children_[child_name];
// Create child if needed.
@ -445,28 +444,26 @@ void PriorityLb::ChoosePriorityLocked() {
}
// Child has been failing for a while. Move on to the next priority.
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] skipping priority %u, child %s: state=%s, "
"failover timer not pending",
this, priority, child_name.c_str(),
ConnectivityStateName(child->connectivity_state()));
LOG(INFO) << "[priority_lb " << this << "] skipping priority " << priority
<< ", child " << child_name << ": state="
<< ConnectivityStateName(child->connectivity_state())
<< ", failover timer not pending";
}
}
// If we didn't find any priority to try, pick the first one in state
// CONNECTING.
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] no priority reachable, checking for CONNECTING "
"priority to delegate to",
this);
LOG(INFO) << "[priority_lb " << this
<< "] no priority reachable, checking for CONNECTING priority to "
"delegate to";
}
for (uint32_t priority = 0; priority < config_->priorities().size();
++priority) {
// If the child for the priority does not exist yet, create it.
const std::string& child_name = config_->priorities()[priority];
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] trying priority %u, child %s", this,
priority, child_name.c_str());
LOG(INFO) << "[priority_lb " << this << "] trying priority " << priority
<< ", child " << child_name;
}
auto& child = children_[child_name];
CHECK(child != nullptr);
@ -486,11 +483,10 @@ void PriorityLb::SetCurrentPriorityLocked(int32_t priority,
bool deactivate_lower_priorities,
const char* reason) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] selecting priority %u, child %s (%s, "
"deactivate_lower_priorities=%d)",
this, priority, config_->priorities()[priority].c_str(), reason,
deactivate_lower_priorities);
LOG(INFO) << "[priority_lb " << this << "] selecting priority " << priority
<< ", child " << config_->priorities()[priority] << " (" << reason
<< ", deactivate_lower_priorities=" << deactivate_lower_priorities
<< ")";
}
current_priority_ = priority;
if (deactivate_lower_priorities) {
@ -515,12 +511,10 @@ PriorityLb::ChildPriority::DeactivationTimer::DeactivationTimer(
RefCountedPtr<PriorityLb::ChildPriority> child_priority)
: child_priority_(std::move(child_priority)) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): deactivating -- will remove in "
"%" PRId64 "ms",
child_priority_->priority_policy_.get(),
child_priority_->name_.c_str(), child_priority_.get(),
kChildRetentionInterval.millis());
LOG(INFO) << "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " ("
<< child_priority_.get() << "): deactivating -- will remove in "
<< kChildRetentionInterval.millis() << "ms";
}
timer_handle_ =
child_priority_->priority_policy_->channel_control_helper()
@ -539,9 +533,9 @@ PriorityLb::ChildPriority::DeactivationTimer::DeactivationTimer(
void PriorityLb::ChildPriority::DeactivationTimer::Orphan() {
if (timer_handle_.has_value()) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] child %s (%p): reactivating",
child_priority_->priority_policy_.get(),
child_priority_->name_.c_str(), child_priority_.get());
LOG(INFO) << "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " ("
<< child_priority_.get() << "): reactivating";
}
child_priority_->priority_policy_->channel_control_helper()
->GetEventEngine()
@ -555,11 +549,10 @@ void PriorityLb::ChildPriority::DeactivationTimer::OnTimerLocked() {
if (timer_handle_.has_value()) {
timer_handle_.reset();
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): deactivation timer fired, "
"deleting child",
child_priority_->priority_policy_.get(),
child_priority_->name_.c_str(), child_priority_.get());
LOG(INFO) << "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " ("
<< child_priority_.get()
<< "): deactivation timer fired, deleting child";
}
child_priority_->priority_policy_->DeleteChild(child_priority_.get());
}
@ -573,13 +566,12 @@ PriorityLb::ChildPriority::FailoverTimer::FailoverTimer(
RefCountedPtr<PriorityLb::ChildPriority> child_priority)
: child_priority_(std::move(child_priority)) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(
GPR_INFO,
"[priority_lb %p] child %s (%p): starting failover timer for %" PRId64
"ms",
child_priority_->priority_policy_.get(), child_priority_->name_.c_str(),
child_priority_.get(),
child_priority_->priority_policy_->child_failover_timeout_.millis());
LOG(INFO)
<< "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " (" << child_priority_.get()
<< "): starting failover timer for "
<< child_priority_->priority_policy_->child_failover_timeout_.millis()
<< "ms";
}
timer_handle_ =
child_priority_->priority_policy_->channel_control_helper()
@ -599,10 +591,9 @@ PriorityLb::ChildPriority::FailoverTimer::FailoverTimer(
void PriorityLb::ChildPriority::FailoverTimer::Orphan() {
if (timer_handle_.has_value()) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): cancelling failover timer",
child_priority_->priority_policy_.get(),
child_priority_->name_.c_str(), child_priority_.get());
LOG(INFO) << "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " ("
<< child_priority_.get() << "): cancelling failover timer";
}
child_priority_->priority_policy_->channel_control_helper()
->GetEventEngine()
@ -616,11 +607,10 @@ void PriorityLb::ChildPriority::FailoverTimer::OnTimerLocked() {
if (timer_handle_.has_value()) {
timer_handle_.reset();
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): failover timer fired, "
"reporting TRANSIENT_FAILURE",
child_priority_->priority_policy_.get(),
child_priority_->name_.c_str(), child_priority_.get());
LOG(INFO) << "[priority_lb " << child_priority_->priority_policy_.get()
<< "] child " << child_priority_->name_ << " ("
<< child_priority_.get()
<< "): failover timer fired, reporting TRANSIENT_FAILURE";
}
child_priority_->OnConnectivityStateUpdateLocked(
GRPC_CHANNEL_TRANSIENT_FAILURE,
@ -637,8 +627,8 @@ PriorityLb::ChildPriority::ChildPriority(
RefCountedPtr<PriorityLb> priority_policy, std::string name)
: priority_policy_(std::move(priority_policy)), name_(std::move(name)) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] creating child %s (%p)",
priority_policy_.get(), name_.c_str(), this);
LOG(INFO) << "[priority_lb " << priority_policy_.get()
<< "] creating child " << name_ << " (" << this << ")";
}
// Start the failover timer.
failover_timer_ = MakeOrphanable<FailoverTimer>(Ref());
@ -646,8 +636,8 @@ PriorityLb::ChildPriority::ChildPriority(
void PriorityLb::ChildPriority::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] child %s (%p): orphaned",
priority_policy_.get(), name_.c_str(), this);
LOG(INFO) << "[priority_lb " << priority_policy_.get() << "] child "
<< name_ << " (" << this << "): orphaned";
}
failover_timer_.reset();
deactivation_timer_.reset();
@ -676,8 +666,8 @@ absl::Status PriorityLb::ChildPriority::UpdateLocked(
bool ignore_reresolution_requests) {
if (priority_policy_->shutting_down_) return absl::OkStatus();
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO, "[priority_lb %p] child %s (%p): start update",
priority_policy_.get(), name_.c_str(), this);
LOG(INFO) << "[priority_lb " << priority_policy_.get() << "] child "
<< name_ << " (" << this << "): start update";
}
ignore_reresolution_requests_ = ignore_reresolution_requests;
// Create policy if needed.
@ -702,9 +692,9 @@ absl::Status PriorityLb::ChildPriority::UpdateLocked(
update_args.args = priority_policy_->args_;
// Update the policy.
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): updating child policy handler %p",
priority_policy_.get(), name_.c_str(), this, child_policy_.get());
LOG(INFO) << "[priority_lb " << priority_policy_.get() << "] child "
<< name_ << " (" << this << "): updating child policy handler "
<< child_policy_.get();
}
return child_policy_->UpdateLocked(std::move(update_args));
}
@ -720,10 +710,9 @@ PriorityLb::ChildPriority::CreateChildPolicyLocked(const ChannelArgs& args) {
MakeOrphanable<ChildPolicyHandler>(std::move(lb_policy_args),
&priority_lb_trace);
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): created new child policy "
"handler %p",
priority_policy_.get(), name_.c_str(), this, lb_policy.get());
LOG(INFO) << "[priority_lb " << priority_policy_.get() << "] child "
<< name_ << " (" << this << "): created new child policy handler "
<< lb_policy.get();
}
// Add the parent's interested_parties pollset_set to that of the newly
// created child policy. This will make the child policy progress upon
@ -745,11 +734,10 @@ void PriorityLb::ChildPriority::OnConnectivityStateUpdateLocked(
grpc_connectivity_state state, const absl::Status& status,
RefCountedPtr<SubchannelPicker> picker) {
if (GRPC_TRACE_FLAG_ENABLED(priority_lb)) {
gpr_log(GPR_INFO,
"[priority_lb %p] child %s (%p): state update: %s (%s) picker %p",
priority_policy_.get(), name_.c_str(), this,
ConnectivityStateName(state), status.ToString().c_str(),
picker.get());
LOG(INFO) << "[priority_lb " << priority_policy_.get() << "] child "
<< name_ << " (" << this
<< "): state update: " << ConnectivityStateName(state) << " ("
<< status << ") picker " << picker.get();
}
// Store the state and picker.
connectivity_state_ = state;

16
src/core/load_balancing/ring_hash/ring_hash.cc
Unescape View File

@ -40,7 +40,6 @@
#include <grpc/impl/channel_arg_names.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/json.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/client_channel/client_channel_internal.h"
@ -573,14 +572,13 @@ void RingHash::RingHashEndpoint::OnStateUpdate(
grpc_connectivity_state new_state, const absl::Status& status,
RefCountedPtr<SubchannelPicker> picker) {
if (GRPC_TRACE_FLAG_ENABLED(ring_hash_lb)) {
gpr_log(
GPR_INFO,
"[RH %p] connectivity changed for endpoint %p (%s, child_policy=%p): "
"prev_state=%s new_state=%s (%s)",
ring_hash_.get(), this,
ring_hash_->endpoints_[index_].ToString().c_str(), child_policy_.get(),
ConnectivityStateName(connectivity_state_),
ConnectivityStateName(new_state), status.ToString().c_str());
LOG(INFO) << "[RH " << ring_hash_.get()
<< "] connectivity changed for endpoint " << this << " ("
<< ring_hash_->endpoints_[index_].ToString()
<< ", child_policy=" << child_policy_.get()
<< "): prev_state=" << ConnectivityStateName(connectivity_state_)
<< " new_state=" << ConnectivityStateName(new_state) << " ("
<< status << ")";
}
if (child_policy_ == nullptr) return; // Already orphaned.
// Update state.

195
src/core/load_balancing/rls/rls.cc
Unescape View File

@ -64,7 +64,6 @@
#include <grpc/slice.h>
#include <grpc/status.h>
#include <grpc/support/json.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/channelz/channelz.h"
@ -807,8 +806,9 @@ RlsLb::ChildPolicyWrapper::ChildPolicyWrapper(RefCountedPtr<RlsLb> lb_policy,
void RlsLb::ChildPolicyWrapper::Orphaned() {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] ChildPolicyWrapper=%p [%s]: shutdown",
lb_policy_.get(), this, target_.c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get()
<< "] ChildPolicyWrapper=" << this << " [" << target_
<< "]: shutdown";
}
is_shutdown_ = true;
lb_policy_->child_policy_map_.erase(target_);
@ -866,11 +866,10 @@ void RlsLb::ChildPolicyWrapper::StartUpdate() {
lb_policy_->config_->child_policy_config(), &errors);
CHECK(child_policy_config.has_value());
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(
GPR_INFO,
"[rlslb %p] ChildPolicyWrapper=%p [%s]: validating update, config: %s",
lb_policy_.get(), this, target_.c_str(),
JsonDump(*child_policy_config).c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get()
<< "] ChildPolicyWrapper=" << this << " [" << target_
<< "]: validating update, config: "
<< JsonDump(*child_policy_config);
}
auto config =
CoreConfiguration::Get().lb_policy_registry().ParseLoadBalancingConfig(
@ -878,11 +877,9 @@ void RlsLb::ChildPolicyWrapper::StartUpdate() {
// Returned RLS target fails the validation.
if (!config.ok()) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] ChildPolicyWrapper=%p [%s]: config failed to parse: "
"%s",
lb_policy_.get(), this, target_.c_str(),
config.status().ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get()
<< "] ChildPolicyWrapper=" << this << " [" << target_
<< "]: config failed to parse: " << config.status();
}
pending_config_.reset();
picker_ = MakeRefCounted<TransientFailurePicker>(
@ -907,20 +904,19 @@ absl::Status RlsLb::ChildPolicyWrapper::MaybeFinishUpdate() {
child_policy_ = MakeOrphanable<ChildPolicyHandler>(std::move(create_args),
&rls_lb_trace);
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] ChildPolicyWrapper=%p [%s], created new child policy "
"handler %p",
lb_policy_.get(), this, target_.c_str(), child_policy_.get());
LOG(INFO) << "[rlslb " << lb_policy_.get()
<< "] ChildPolicyWrapper=" << this << " [" << target_
<< "], created new child policy handler "
<< child_policy_.get();
}
grpc_pollset_set_add_pollset_set(child_policy_->interested_parties(),
lb_policy_->interested_parties());
}
// Send the child the updated config.
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] ChildPolicyWrapper=%p [%s], updating child policy "
"handler %p",
lb_policy_.get(), this, target_.c_str(), child_policy_.get());
LOG(INFO) << "[rlslb " << lb_policy_.get()
<< "] ChildPolicyWrapper=" << this << " [" << target_
<< "], updating child policy handler " << child_policy_.get();
}
UpdateArgs update_args;
update_args.config = std::move(pending_config_);
@ -937,12 +933,11 @@ void RlsLb::ChildPolicyWrapper::ChildPolicyHelper::UpdateState(
grpc_connectivity_state state, const absl::Status& status,
RefCountedPtr<SubchannelPicker> picker) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] ChildPolicyWrapper=%p [%s] ChildPolicyHelper=%p: "
"UpdateState(state=%s, status=%s, picker=%p)",
wrapper_->lb_policy_.get(), wrapper_.get(),
wrapper_->target_.c_str(), this, ConnectivityStateName(state),
status.ToString().c_str(), picker.get());
LOG(INFO) << "[rlslb " << wrapper_->lb_policy_.get()
<< "] ChildPolicyWrapper=" << wrapper_.get() << " ["
<< wrapper_->target_ << "] ChildPolicyHelper=" << this
<< ": UpdateState(state=" << ConnectivityStateName(state)
<< ", status=" << status << ", picker=" << picker.get() << ")";
}
{
MutexLock lock(&wrapper_->lb_policy_->mu_);
@ -1045,8 +1040,8 @@ LoadBalancingPolicy::PickResult RlsLb::Picker::Pick(PickArgs args) {
lb_policy_->channel_control_helper()->GetAuthority(),
args.initial_metadata)};
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] picker=%p: request keys: %s",
lb_policy_.get(), this, key.ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] picker=" << this
<< ": request keys: " << key.ToString();
}
Timestamp now = Timestamp::Now();
MutexLock lock(&lb_policy_->mu_);
@ -1083,8 +1078,8 @@ LoadBalancingPolicy::PickResult RlsLb::Picker::Pick(PickArgs args) {
// If the entry has non-expired data, use it.
if (entry->data_expiration_time() >= now) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] picker=%p: using cache entry %p",
lb_policy_.get(), this, entry);
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] picker=" << this
<< ": using cache entry " << entry;
}
return entry->Pick(args);
}
@ -1099,8 +1094,8 @@ LoadBalancingPolicy::PickResult RlsLb::Picker::Pick(PickArgs args) {
}
// RLS call pending. Queue the pick.
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] picker=%p: RLS request pending; queuing pick",
lb_policy_.get(), this);
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] picker=" << this
<< ": RLS request pending; queuing pick";
}
return PickResult::Queue();
}
@ -1109,8 +1104,8 @@ LoadBalancingPolicy::PickResult RlsLb::Picker::PickFromDefaultTargetOrFail(
const char* reason, PickArgs args, absl::Status status) {
if (default_child_policy_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] picker=%p: %s; using default target",
lb_policy_.get(), this, reason);
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] picker=" << this << ": "
<< reason << "; using default target";
}
auto pick_result = default_child_policy_->Pick(args);
lb_policy_->MaybeExportPickCount(kMetricDefaultTargetPicks,
@ -1118,8 +1113,8 @@ LoadBalancingPolicy::PickResult RlsLb::Picker::PickFromDefaultTargetOrFail(
return pick_result;
}
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] picker=%p: %s; failing pick",
lb_policy_.get(), this, reason);
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] picker=" << this << ": "
<< reason << "; failing pick";
}
auto& stats_plugins =
lb_policy_->channel_control_helper()->GetStatsPluginGroup();
@ -1155,10 +1150,11 @@ void RlsLb::Cache::Entry::BackoffTimer::Orphan() {
entry_->lb_policy_->channel_control_helper()->GetEventEngine()->Cancel(
*backoff_timer_task_handle_)) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] cache entry=%p %s, backoff timer canceled",
entry_->lb_policy_.get(), entry_.get(),
entry_->is_shutdown_ ? "(shut down)"
: entry_->lru_iterator_->ToString().c_str());
LOG(INFO) << "[rlslb " << entry_->lb_policy_.get()
<< "] cache entry=" << entry_.get() << " "
<< (entry_->is_shutdown_ ? "(shut down)"
: entry_->lru_iterator_->ToString())
<< ", backoff timer canceled";
}
}
backoff_timer_task_handle_.reset();
@ -1169,10 +1165,11 @@ void RlsLb::Cache::Entry::BackoffTimer::OnBackoffTimerLocked() {
{
MutexLock lock(&entry_->lb_policy_->mu_);
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] cache entry=%p %s, backoff timer fired",
entry_->lb_policy_.get(), entry_.get(),
entry_->is_shutdown_ ? "(shut down)"
: entry_->lru_iterator_->ToString().c_str());
LOG(INFO) << "[rlslb " << entry_->lb_policy_.get()
<< "] cache entry=" << entry_.get() << " "
<< (entry_->is_shutdown_ ? "(shut down)"
: entry_->lru_iterator_->ToString())
<< ", backoff timer fired";
}
// Skip the update if Orphaned
if (!backoff_timer_task_handle_.has_value()) return;
@ -1208,8 +1205,8 @@ RlsLb::Cache::Entry::Entry(RefCountedPtr<RlsLb> lb_policy,
void RlsLb::Cache::Entry::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] cache entry=%p %s: cache entry evicted",
lb_policy_.get(), this, lru_iterator_->ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] cache entry=" << this
<< " " << lru_iterator_->ToString() << ": cache entry evicted";
}
is_shutdown_ = true;
lb_policy_->cache_.lru_list_.erase(lru_iterator_);
@ -1239,12 +1236,11 @@ LoadBalancingPolicy::PickResult RlsLb::Cache::Entry::Pick(PickArgs args) {
GRPC_CHANNEL_TRANSIENT_FAILURE &&
i < child_policy_wrappers_.size() - 1) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] cache entry=%p %s: target %s (%" PRIuPTR
" of %" PRIuPTR ") in state TRANSIENT_FAILURE; skipping",
lb_policy_.get(), this, lru_iterator_->ToString().c_str(),
child_policy_wrapper->target().c_str(), i,
child_policy_wrappers_.size());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] cache entry=" << this
<< " " << lru_iterator_->ToString() << ": target "
<< child_policy_wrapper->target() << " (" << i << " of "
<< child_policy_wrappers_.size()
<< ") in state TRANSIENT_FAILURE; skipping";
}
continue;
}
@ -1253,13 +1249,13 @@ LoadBalancingPolicy::PickResult RlsLb::Cache::Entry::Pick(PickArgs args) {
// Child policy not in TRANSIENT_FAILURE or is the last target in
// the list, so delegate.
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] cache entry=%p %s: target %s (%" PRIuPTR " of %" PRIuPTR
") in state %s; delegating",
lb_policy_.get(), this, lru_iterator_->ToString().c_str(),
child_policy_wrapper->target().c_str(), i,
child_policy_wrappers_.size(),
ConnectivityStateName(child_policy_wrapper->connectivity_state()));
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] cache entry=" << this
<< " " << lru_iterator_->ToString() << ": target "
<< child_policy_wrapper->target() << " (" << i << " of "
<< child_policy_wrappers_.size() << ") in state "
<< ConnectivityStateName(
child_policy_wrapper->connectivity_state())
<< "; delegating";
}
auto pick_result = child_policy_wrapper->Pick(args);
lb_policy_->MaybeExportPickCount(kMetricTargetPicks,
@ -1408,15 +1404,15 @@ RlsLb::Cache::Entry* RlsLb::Cache::FindOrInsert(const RequestKey& key) {
map_.emplace(key, OrphanablePtr<Entry>(entry));
size_ += entry_size;
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] key=%s: cache entry added, entry=%p",
lb_policy_, key.ToString().c_str(), entry);
LOG(INFO) << "[rlslb " << lb_policy_ << "] key=" << key.ToString()
<< ": cache entry added, entry=" << entry;
}
return entry;
}
// Entry found, so use it.
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] key=%s: found cache entry %p", lb_policy_,
key.ToString().c_str(), it->second.get());
LOG(INFO) << "[rlslb " << lb_policy_ << "] key=" << key.ToString()
<< ": found cache entry " << it->second.get();
}
it->second->MarkUsed();
return it->second.get();
@ -1424,8 +1420,8 @@ RlsLb::Cache::Entry* RlsLb::Cache::FindOrInsert(const RequestKey& key) {
void RlsLb::Cache::Resize(size_t bytes) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] resizing cache to %" PRIuPTR " bytes",
lb_policy_, bytes);
LOG(INFO) << "[rlslb " << lb_policy_ << "] resizing cache to " << bytes
<< " bytes";
}
size_limit_ = bytes;
MaybeShrinkSize(size_limit_);
@ -1445,7 +1441,7 @@ void RlsLb::Cache::Shutdown() {
lb_policy_->channel_control_helper()->GetEventEngine()->Cancel(
*cleanup_timer_handle_)) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] cache cleanup timer canceled", lb_policy_);
LOG(INFO) << "[rlslb " << lb_policy_ << "] cache cleanup timer canceled";
}
}
cleanup_timer_handle_.reset();
@ -1483,7 +1479,7 @@ void RlsLb::Cache::StartCleanupTimer() {
void RlsLb::Cache::OnCleanupTimer() {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] cache cleanup timer fired", lb_policy_);
LOG(INFO) << "[rlslb " << lb_policy_ << "] cache cleanup timer fired";
}
MutexLock lock(&lb_policy_->mu_);
if (!cleanup_timer_handle_.has_value()) return;
@ -1512,17 +1508,16 @@ void RlsLb::Cache::MaybeShrinkSize(size_t bytes) {
CHECK(map_it != map_.end());
if (!map_it->second->CanEvict()) break;
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] LRU eviction: removing entry %p %s",
lb_policy_, map_it->second.get(), lru_it->ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_ << "] LRU eviction: removing entry "
<< map_it->second.get() << " " << lru_it->ToString();
}
size_ -= map_it->second->Size();
map_.erase(map_it);
}
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] LRU pass complete: desired size=%" PRIuPTR
" size=%" PRIuPTR,
lb_policy_, bytes, size_);
LOG(INFO) << "[rlslb " << lb_policy_
<< "] LRU pass complete: desired size=" << bytes
<< " size=" << size_;
}
}
@ -1534,11 +1529,9 @@ void RlsLb::RlsChannel::StateWatcher::OnConnectivityStateChange(
grpc_connectivity_state new_state, const absl::Status& status) {
auto* lb_policy = rls_channel_->lb_policy_.get();
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] RlsChannel=%p StateWatcher=%p: "
"state changed to %s (%s)",
lb_policy, rls_channel_.get(), this,
ConnectivityStateName(new_state), status.ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy << "] RlsChannel=" << rls_channel_.get()
<< " StateWatcher=" << this << ": state changed to "
<< ConnectivityStateName(new_state) << " (" << status << ")";
}
if (rls_channel_->is_shutdown_) return;
MutexLock lock(&lb_policy->mu_);
@ -1633,9 +1626,9 @@ RlsLb::RlsChannel::RlsChannel(RefCountedPtr<RlsLb> lb_policy)
grpc_channel_create(lb_policy_->config_->lookup_service().c_str(),
creds.get(), args.ToC().get())));
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] RlsChannel=%p: created channel %p for %s",
lb_policy_.get(), this, channel_.get(),
lb_policy_->config_->lookup_service().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] RlsChannel=" << this
<< ": created channel " << channel_.get() << " for "
<< lb_policy_->config_->lookup_service();
}
if (channel_ != nullptr) {
// Set up channelz linkage.
@ -1656,8 +1649,8 @@ RlsLb::RlsChannel::RlsChannel(RefCountedPtr<RlsLb> lb_policy)
void RlsLb::RlsChannel::Orphan() {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] RlsChannel=%p, channel=%p: shutdown",
lb_policy_.get(), this, channel_.get());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] RlsChannel=" << this
<< ", channel=" << channel_.get() << ": shutdown";
}
is_shutdown_ = true;
if (channel_ != nullptr) {
@ -1723,9 +1716,8 @@ RlsLb::RlsRequest::RlsRequest(
reason_(reason),
stale_header_data_(std::move(stale_header_data)) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] rls_request=%p: RLS request created for key %s",
lb_policy_.get(), this, key_.ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] rls_request=" << this
<< ": RLS request created for key " << key_.ToString();
}
GRPC_CLOSURE_INIT(&call_complete_cb_, OnRlsCallComplete, this, nullptr);
ExecCtx::Run(
@ -1740,8 +1732,8 @@ RlsLb::RlsRequest::~RlsRequest() { CHECK_EQ(call_, nullptr); }
void RlsLb::RlsRequest::Orphan() {
if (call_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] rls_request=%p %s: cancelling RLS call",
lb_policy_.get(), this, key_.ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] rls_request=" << this
<< " " << key_.ToString() << ": cancelling RLS call";
}
grpc_call_cancel_internal(call_);
}
@ -1814,12 +1806,10 @@ void RlsLb::RlsRequest::OnRlsCallComplete(void* arg, grpc_error_handle error) {
void RlsLb::RlsRequest::OnRlsCallCompleteLocked(grpc_error_handle error) {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
std::string status_message(StringViewFromSlice(status_details_recv_));
gpr_log(GPR_INFO,
"[rlslb %p] rls_request=%p %s, error=%s, status={%d, %s} RLS call "
"response received",
lb_policy_.get(), this, key_.ToString().c_str(),
StatusToString(error).c_str(), status_recv_,
status_message.c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] rls_request=" << this
<< " " << key_.ToString() << ", error=" << StatusToString(error)
<< ", status={" << status_recv_ << ", " << status_message << "}"
<< " RLS call response received";
}
// Parse response.
ResponseInfo response;
@ -1846,9 +1836,9 @@ void RlsLb::RlsRequest::OnRlsCallCompleteLocked(grpc_error_handle error) {
call_ = nullptr;
// Return result to cache.
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] rls_request=%p %s: response info: %s",
lb_policy_.get(), this, key_.ToString().c_str(),
response.ToString().c_str());
LOG(INFO) << "[rlslb " << lb_policy_.get() << "] rls_request=" << this
<< " " << key_.ToString()
<< ": response info: " << response.ToString();
}
std::vector<ChildPolicyWrapper*> child_policies_to_finish_update;
{
@ -2042,8 +2032,8 @@ absl::Status RlsLb::UpdateLocked(UpdateArgs args) {
created_default_child = true;
} else {
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO,
"[rlslb %p] using existing child for default target", this);
LOG(INFO) << "[rlslb " << this
<< "] using existing child for default target";
}
default_child_policy_ =
it->second->Ref(DEBUG_LOCATION, "DefaultChildPolicy");
@ -2197,9 +2187,8 @@ void RlsLb::UpdatePickerLocked() {
for (auto& p : child_policy_map_) {
grpc_connectivity_state child_state = p.second->connectivity_state();
if (GRPC_TRACE_FLAG_ENABLED(rls_lb)) {
gpr_log(GPR_INFO, "[rlslb %p] target %s in state %s", this,
p.second->target().c_str(),
ConnectivityStateName(child_state));
LOG(INFO) << "[rlslb " << this << "] target " << p.second->target()
<< " in state " << ConnectivityStateName(child_state);
}
if (child_state == GRPC_CHANNEL_READY) {
state = GRPC_CHANNEL_READY;

117
src/core/load_balancing/weighted_round_robin/weighted_round_robin.cc
Unescape View File

@ -42,7 +42,6 @@
#include <grpc/event_engine/event_engine.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/channel/channel_args.h"
@ -449,11 +448,11 @@ void WeightedRoundRobin::EndpointWeight::MaybeUpdateWeight(
}
if (weight == 0) {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] subchannel %s: qps=%f, eps=%f, utilization=%f: "
"error_util_penalty=%f, weight=%f (not updating)",
wrr_.get(), key_.ToString().c_str(), qps, eps, utilization,
error_utilization_penalty, weight);
LOG(INFO) << "[WRR " << wrr_.get() << "] subchannel " << key_.ToString()
<< ": qps=" << qps << ", eps=" << eps
<< ", utilization=" << utilization
<< ": error_util_penalty=" << error_utilization_penalty
<< ", weight=" << weight << " (not updating)";
}
return;
}
@ -461,14 +460,14 @@ void WeightedRoundRobin::EndpointWeight::MaybeUpdateWeight(
// Grab the lock and update the data.
MutexLock lock(&mu_);
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] subchannel %s: qps=%f, eps=%f, utilization=%f "
"error_util_penalty=%f : setting weight=%f weight_=%f now=%s "
"last_update_time_=%s non_empty_since_=%s",
wrr_.get(), key_.ToString().c_str(), qps, eps, utilization,
error_utilization_penalty, weight, weight_, now.ToString().c_str(),
last_update_time_.ToString().c_str(),
non_empty_since_.ToString().c_str());
LOG(INFO) << "[WRR " << wrr_.get() << "] subchannel " << key_.ToString()
<< ": qps=" << qps << ", eps=" << eps
<< ", utilization=" << utilization
<< " error_util_penalty=" << error_utilization_penalty
<< " : setting weight=" << weight << " weight_=" << weight_
<< " now=" << now.ToString()
<< " last_update_time_=" << last_update_time_.ToString()
<< " non_empty_since_=" << non_empty_since_.ToString();
}
if (non_empty_since_ == Timestamp::InfFuture()) non_empty_since_ = now;
weight_ = weight;
@ -480,15 +479,14 @@ float WeightedRoundRobin::EndpointWeight::GetWeight(
uint64_t* num_not_yet_usable, uint64_t* num_stale) {
MutexLock lock(&mu_);
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] subchannel %s: getting weight: now=%s "
"weight_expiration_period=%s blackout_period=%s "
"last_update_time_=%s non_empty_since_=%s weight_=%f",
wrr_.get(), key_.ToString().c_str(), now.ToString().c_str(),
weight_expiration_period.ToString().c_str(),
blackout_period.ToString().c_str(),
last_update_time_.ToString().c_str(),
non_empty_since_.ToString().c_str(), weight_);
LOG(INFO) << "[WRR " << wrr_.get() << "] subchannel " << key_.ToString()
<< ": getting weight: now=" << now.ToString()
<< " weight_expiration_period="
<< weight_expiration_period.ToString()
<< " blackout_period=" << blackout_period.ToString()
<< " last_update_time_=" << last_update_time_.ToString()
<< " non_empty_since_=" << non_empty_since_.ToString()
<< " weight_=" << weight_;
}
// If the most recent update was longer ago than the expiration
// period, reset non_empty_since_ so that we apply the blackout period
@ -558,24 +556,25 @@ WeightedRoundRobin::Picker::Picker(RefCountedPtr<WeightedRoundRobin> wrr,
global_stats().IncrementWrrSubchannelListSize(endpoint_list->size());
global_stats().IncrementWrrSubchannelReadySize(endpoints_.size());
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p picker %p] created picker from endpoint_list=%p "
"with %" PRIuPTR " subchannels",
wrr_.get(), this, endpoint_list, endpoints_.size());
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] created picker from endpoint_list=" << endpoint_list
<< " with " << endpoints_.size() << " subchannels";
}
BuildSchedulerAndStartTimerLocked();
}
WeightedRoundRobin::Picker::~Picker() {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] destroying picker", wrr_.get(), this);
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] destroying picker";
}
}
void WeightedRoundRobin::Picker::Orphaned() {
MutexLock lock(&timer_mu_);
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] cancelling timer", wrr_.get(), this);
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] cancelling timer";
}
wrr_->channel_control_helper()->GetEventEngine()->Cancel(*timer_handle_);
timer_handle_.reset();
@ -587,9 +586,9 @@ WeightedRoundRobin::PickResult WeightedRoundRobin::Picker::Pick(PickArgs args) {
CHECK(index < endpoints_.size());
auto& endpoint_info = endpoints_[index];
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p picker %p] returning index %" PRIuPTR ", picker=%p",
wrr_.get(), this, index, endpoint_info.picker.get());
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] returning index " << index
<< ", picker=" << endpoint_info.picker.get();
}
auto result = endpoint_info.picker->Pick(args);
// Collect per-call utilization data if needed.
@ -643,8 +642,8 @@ void WeightedRoundRobin::Picker::BuildSchedulerAndStartTimerLocked() {
{wrr_->channel_control_helper()->GetTarget()},
{wrr_->locality_name_});
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] new weights: %s", wrr_.get(), this,
absl::StrJoin(weights, " ").c_str());
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] new weights: " << absl::StrJoin(weights, " ");
}
auto scheduler_or = StaticStrideScheduler::Make(
weights, [this]() { return wrr_->scheduler_state_.fetch_add(1); });
@ -653,13 +652,13 @@ void WeightedRoundRobin::Picker::BuildSchedulerAndStartTimerLocked() {
scheduler =
std::make_shared<StaticStrideScheduler>(std::move(*scheduler_or));
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] new scheduler: %p", wrr_.get(),
this, scheduler.get());
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] new scheduler: " << scheduler.get();
}
} else {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] no scheduler, falling back to RR",
wrr_.get(), this);
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] no scheduler, falling back to RR";
}
stats_plugins.AddCounter(kMetricRrFallback, 1,
{wrr_->channel_control_helper()->GetTarget()},
@ -671,8 +670,9 @@ void WeightedRoundRobin::Picker::BuildSchedulerAndStartTimerLocked() {
}
// Start timer.
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] scheduling timer for %s", wrr_.get(),
this, config_->weight_update_period().ToString().c_str());
LOG(INFO) << "[WRR " << wrr_.get() << " picker " << this
<< "] scheduling timer for "
<< config_->weight_update_period().ToString();
}
timer_handle_ = wrr_->channel_control_helper()->GetEventEngine()->RunAfter(
config_->weight_update_period(),
@ -684,8 +684,8 @@ void WeightedRoundRobin::Picker::BuildSchedulerAndStartTimerLocked() {
MutexLock lock(&self->timer_mu_);
if (self->timer_handle_.has_value()) {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO, "[WRR %p picker %p] timer fired",
self->wrr_.get(), self.get());
LOG(INFO) << "[WRR " << self->wrr_.get() << " picker "
<< self.get() << "] timer fired";
}
self->BuildSchedulerAndStartTimerLocked();
}
@ -876,19 +876,19 @@ void WeightedRoundRobin::WrrEndpointList::WrrEndpoint::OnStateUpdate(
auto* wrr_endpoint_list = endpoint_list<WrrEndpointList>();
auto* wrr = policy<WeightedRoundRobin>();
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] connectivity changed for child %p, endpoint_list %p "
"(index %" PRIuPTR " of %" PRIuPTR
"): prev_state=%s new_state=%s (%s)",
wrr, this, wrr_endpoint_list, Index(), wrr_endpoint_list->size(),
(old_state.has_value() ? ConnectivityStateName(*old_state) : "N/A"),
ConnectivityStateName(new_state), status.ToString().c_str());
LOG(INFO) << "[WRR " << wrr << "] connectivity changed for child " << this
<< ", endpoint_list " << wrr_endpoint_list << " (index "
<< Index() << " of " << wrr_endpoint_list->size()
<< "): prev_state="
<< (old_state.has_value() ? ConnectivityStateName(*old_state)
: "N/A")
<< " new_state=" << ConnectivityStateName(new_state) << " ("
<< status << ")";
}
if (new_state == GRPC_CHANNEL_IDLE) {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] child %p reported IDLE; requesting connection", wrr,
this);
LOG(INFO) << "[WRR " << wrr << "] child " << this
<< " reported IDLE; requesting connection";
}
ExitIdleLocked();
} else if (new_state == GRPC_CHANNEL_READY) {
@ -970,10 +970,9 @@ void WeightedRoundRobin::WrrEndpointList::
const std::string old_counters_string =
wrr->endpoint_list_ != nullptr ? wrr->endpoint_list_->CountersString()
: "";
gpr_log(GPR_INFO,
"[WRR %p] swapping out endpoint list %p (%s) in favor of %p (%s)",
wrr, wrr->endpoint_list_.get(), old_counters_string.c_str(), this,
CountersString().c_str());
LOG(INFO) << "[WRR " << wrr << "] swapping out endpoint list "
<< wrr->endpoint_list_.get() << " (" << old_counters_string
<< ") in favor of " << this << " (" << CountersString() << ")";
}
wrr->endpoint_list_ = std::move(wrr->latest_pending_endpoint_list_);
}
@ -1001,9 +1000,9 @@ void WeightedRoundRobin::WrrEndpointList::
MakeRefCounted<QueuePicker>(nullptr));
} else if (num_transient_failure_ == size()) {
if (GRPC_TRACE_FLAG_ENABLED(weighted_round_robin_lb)) {
gpr_log(GPR_INFO,
"[WRR %p] reporting TRANSIENT_FAILURE with endpoint list %p: %s",
wrr, this, status_for_tf.ToString().c_str());
LOG(INFO) << "[WRR " << wrr
<< "] reporting TRANSIENT_FAILURE with endpoint list " << this
<< ": " << status_for_tf;
}
if (!status_for_tf.ok()) {
last_failure_ = absl::UnavailableError(

19
src/core/load_balancing/xds/cds.cc
Unescape View File

@ -35,7 +35,6 @@
#include <grpc/grpc_security.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/json.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/lib/channel/channel_args.h"
@ -282,8 +281,9 @@ absl::Status CdsLb::UpdateLocked(UpdateArgs args) {
// Get new config.
auto new_config = args.config.TakeAsSubclass<CdsLbConfig>();
if (GRPC_TRACE_FLAG_ENABLED(cds_lb)) {
gpr_log(GPR_INFO, "[cdslb %p] received update: cluster=%s is_dynamic=%d",
this, new_config->cluster().c_str(), new_config->is_dynamic());
LOG(INFO) << "[cdslb " << this
<< "] received update: cluster=" << new_config->cluster()
<< " is_dynamic=" << new_config->is_dynamic();
}
CHECK(new_config != nullptr);
// Cluster name should never change, because we should use a different
@ -327,10 +327,9 @@ absl::Status CdsLb::UpdateLocked(UpdateArgs args) {
// recently subscribed but another update came through before we
// got the new cluster, in which case it will still be missing.
if (GRPC_TRACE_FLAG_ENABLED(cds_lb)) {
gpr_log(GPR_INFO,
"[cdslb %p] xDS config has no entry for dynamic cluster %s, "
"waiting for subsequent update",
this, cluster_name_.c_str());
LOG(INFO) << "[cdslb " << this
<< "] xDS config has no entry for dynamic cluster "
<< cluster_name_ << ", waiting for subsequent update";
}
// Stay in CONNECTING until we get an update that has the cluster.
return absl::OkStatus();
@ -454,9 +453,9 @@ absl::Status CdsLb::UpdateLocked(UpdateArgs args) {
grpc_pollset_set_add_pollset_set(child_policy_->interested_parties(),
interested_parties());
if (GRPC_TRACE_FLAG_ENABLED(cds_lb)) {
gpr_log(GPR_INFO, "[cdslb %p] created child policy %s (%p)", this,
std::string((*child_config)->name()).c_str(),
child_policy_.get());
LOG(INFO) << "[cdslb " << this << "] created child policy "
<< (*child_config)->name() << " (" << child_policy_.get()
<< ")";
}
}
// Update child policy.

7
src/core/load_balancing/xds/xds_cluster_manager.cc
Unescape View File

@ -35,7 +35,6 @@
#include <grpc/event_engine/event_engine.h>
#include <grpc/impl/connectivity_state.h>
#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
#include "src/core/client_channel/client_channel_internal.h"
@ -251,10 +250,8 @@ XdsClusterManagerLb::XdsClusterManagerLb(Args args)
XdsClusterManagerLb::~XdsClusterManagerLb() {
if (GRPC_TRACE_FLAG_ENABLED(xds_cluster_manager_lb)) {
gpr_log(
GPR_INFO,
"[xds_cluster_manager_lb %p] destroying xds_cluster_manager LB policy",
this);
LOG(INFO) << "[xds_cluster_manager_lb " << this
<< "] destroying xds_cluster_manager LB policy";
}
}

8
tools/run_tests/sanity/banned_functions.py
Unescape View File

@ -158,14 +158,6 @@ DEPRECATED_FUNCTION_TEMP_ALLOW_LIST = {
"./src/core/lib/transport/call_filters.cc",
"./src/core/lib/transport/connectivity_state.cc",
"./src/core/lib/transport/transport.h",
"./src/core/load_balancing/grpclb/grpclb.cc",
"./src/core/load_balancing/pick_first/pick_first.cc",
"./src/core/load_balancing/priority/priority.cc",
"./src/core/load_balancing/ring_hash/ring_hash.cc",
"./src/core/load_balancing/rls/rls.cc",
"./src/core/load_balancing/weighted_round_robin/weighted_round_robin.cc",
"./src/core/load_balancing/xds/cds.cc",
"./src/core/load_balancing/xds/xds_cluster_manager.cc",
"./src/core/resolver/dns/c_ares/grpc_ares_wrapper.cc",
"./src/core/resolver/dns/c_ares/grpc_ares_wrapper.h",
"./src/core/resolver/dns/event_engine/event_engine_client_channel_resolver.cc",

Write Preview
Loading…
Cancel
Save