|  | @@ -125,10 +125,35 @@ class XdsClient::ChannelState::AdsCallState
 | 
	
		
			
				|  |  |    XdsClient* xds_client() const { return chand()->xds_client(); }
 | 
	
		
			
				|  |  |    bool seen_response() const { return seen_response_; }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | +  // If \a type_url is an unsupported type, \a nonce_for_unsupported_type and
 | 
	
		
			
				|  |  | +  // \a error_for_unsupported_type will be used in the request; otherwise, the
 | 
	
		
			
				|  |  | +  // nonce and error stored in each ADS call state will be used. Takes ownership
 | 
	
		
			
				|  |  | +  // of \a error_for_unsupported_type.
 | 
	
		
			
				|  |  | +  void SendMessageLocked(const std::string& type_url,
 | 
	
		
			
				|  |  | +                         const std::string& nonce_for_unsupported_type,
 | 
	
		
			
				|  |  | +                         grpc_error* error_for_unsupported_type,
 | 
	
		
			
				|  |  | +                         bool is_first_message);
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |   private:
 | 
	
		
			
				|  |  | +  struct BufferedRequest {
 | 
	
		
			
				|  |  | +    std::string nonce;
 | 
	
		
			
				|  |  | +    grpc_error* error;
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    // Takes ownership of \a error.
 | 
	
		
			
				|  |  | +    BufferedRequest(std::string nonce, grpc_error* error)
 | 
	
		
			
				|  |  | +        : nonce(std::move(nonce)), error(error) {}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    ~BufferedRequest() { GRPC_ERROR_UNREF(error); }
 | 
	
		
			
				|  |  | +  };
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  void AcceptCdsUpdate(CdsUpdateMap cds_update_map, std::string new_version);
 | 
	
		
			
				|  |  | +  void AcceptEdsUpdate(EdsUpdateMap eds_update_map, std::string new_version);
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  static void OnRequestSent(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  | +  static void OnRequestSentLocked(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  |    static void OnResponseReceived(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  | -  static void OnStatusReceived(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  |    static void OnResponseReceivedLocked(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  | +  static void OnStatusReceived(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  |    static void OnStatusReceivedLocked(void* arg, grpc_error* error);
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    bool IsCurrentCallOnChannel() const;
 | 
	
	
		
			
				|  | @@ -145,6 +170,7 @@ class XdsClient::ChannelState::AdsCallState
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    // send_message
 | 
	
		
			
				|  |  |    grpc_byte_buffer* send_message_payload_ = nullptr;
 | 
	
		
			
				|  |  | +  grpc_closure on_request_sent_;
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    // recv_message
 | 
	
		
			
				|  |  |    grpc_byte_buffer* recv_message_payload_ = nullptr;
 | 
	
	
		
			
				|  | @@ -155,6 +181,14 @@ class XdsClient::ChannelState::AdsCallState
 | 
	
		
			
				|  |  |    grpc_status_code status_code_;
 | 
	
		
			
				|  |  |    grpc_slice status_details_;
 | 
	
		
			
				|  |  |    grpc_closure on_status_received_;
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  // Version state.
 | 
	
		
			
				|  |  | +  VersionState cds_version_;
 | 
	
		
			
				|  |  | +  VersionState eds_version_;
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  // Buffered requests.
 | 
	
		
			
				|  |  | +  std::map<std::string /*type_url*/, std::unique_ptr<BufferedRequest>>
 | 
	
		
			
				|  |  | +      buffered_request_map_;
 | 
	
		
			
				|  |  |  };
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  // Contains an LRS call to the xds server.
 | 
	
	
		
			
				|  | @@ -168,6 +202,7 @@ class XdsClient::ChannelState::LrsCallState
 | 
	
		
			
				|  |  |    void Orphan() override;
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    void MaybeStartReportingLocked();
 | 
	
		
			
				|  |  | +  bool ShouldSendLoadReports(const StringView& cluster_name) const;
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    RetryableCall<LrsCallState>* parent() { return parent_.get(); }
 | 
	
		
			
				|  |  |    ChannelState* chand() const { return parent_->chand(); }
 | 
	
	
		
			
				|  | @@ -244,7 +279,7 @@ class XdsClient::ChannelState::LrsCallState
 | 
	
		
			
				|  |  |    grpc_closure on_status_received_;
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |    // Load reporting state.
 | 
	
		
			
				|  |  | -  grpc_core::UniquePtr<char> cluster_name_;
 | 
	
		
			
				|  |  | +  std::set<std::string> cluster_names_;  // Asked for by the LRS server.
 | 
	
		
			
				|  |  |    grpc_millis load_reporting_interval_ = 0;
 | 
	
		
			
				|  |  |    OrphanablePtr<Reporter> reporter_;
 | 
	
		
			
				|  |  |  };
 | 
	
	
		
			
				|  | @@ -376,14 +411,6 @@ bool XdsClient::ChannelState::HasActiveAdsCall() const {
 | 
	
		
			
				|  |  |    return ads_calld_->calld() != nullptr;
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -void XdsClient::ChannelState::MaybeStartAdsCall() {
 | 
	
		
			
				|  |  | -  if (ads_calld_ != nullptr) return;
 | 
	
		
			
				|  |  | -  ads_calld_.reset(
 | 
	
		
			
				|  |  | -      new RetryableCall<AdsCallState>(Ref(DEBUG_LOCATION, "ChannelState+ads")));
 | 
	
		
			
				|  |  | -}
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -void XdsClient::ChannelState::StopAdsCall() { ads_calld_.reset(); }
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  |  void XdsClient::ChannelState::MaybeStartLrsCall() {
 | 
	
		
			
				|  |  |    if (lrs_calld_ != nullptr) return;
 | 
	
		
			
				|  |  |    lrs_calld_.reset(
 | 
	
	
		
			
				|  | @@ -409,6 +436,33 @@ void XdsClient::ChannelState::CancelConnectivityWatchLocked() {
 | 
	
		
			
				|  |  |    grpc_client_channel_stop_connectivity_watch(client_channel_elem, watcher_);
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::OnResourceNamesChanged(
 | 
	
		
			
				|  |  | +    const std::string& type_url) {
 | 
	
		
			
				|  |  | +  if (ads_calld_ == nullptr) {
 | 
	
		
			
				|  |  | +    // Start the ADS call if this is the first request.
 | 
	
		
			
				|  |  | +    ads_calld_.reset(new RetryableCall<AdsCallState>(
 | 
	
		
			
				|  |  | +        Ref(DEBUG_LOCATION, "ChannelState+ads")));
 | 
	
		
			
				|  |  | +    // Note: AdsCallState's ctor will automatically send necessary messages, so
 | 
	
		
			
				|  |  | +    // we can return here.
 | 
	
		
			
				|  |  | +    return;
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  // If the ADS call is in backoff state, we don't need to do anything now
 | 
	
		
			
				|  |  | +  // because when the call is restarted it will resend all necessary requests.
 | 
	
		
			
				|  |  | +  if (ads_calld() == nullptr) return;
 | 
	
		
			
				|  |  | +  // Send the message if the ADS call is active.
 | 
	
		
			
				|  |  | +  ads_calld()->SendMessageLocked(type_url, "", nullptr, false);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::OnWatcherRemoved() {
 | 
	
		
			
				|  |  | +  // Keep the ADS call if there are watcher(s).
 | 
	
		
			
				|  |  | +  for (const auto& p : xds_client()->cluster_map_) {
 | 
	
		
			
				|  |  | +    const ClusterState& cluster_state = p.second;
 | 
	
		
			
				|  |  | +    if (!cluster_state.watchers.empty()) return;
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  if (!xds_client()->endpoint_map_.empty()) return;
 | 
	
		
			
				|  |  | +  ads_calld_.reset();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |  //
 | 
	
		
			
				|  |  |  // XdsClient::ChannelState::RetryableCall<>
 | 
	
		
			
				|  |  |  //
 | 
	
	
		
			
				|  | @@ -522,8 +576,7 @@ XdsClient::ChannelState::AdsCallState::AdsCallState(
 | 
	
		
			
				|  |  |    // activity in xds_client()->interested_parties_, which is comprised of
 | 
	
		
			
				|  |  |    // the polling entities from client_channel.
 | 
	
		
			
				|  |  |    GPR_ASSERT(xds_client() != nullptr);
 | 
	
		
			
				|  |  | -  GPR_ASSERT(xds_client()->server_name_ != nullptr);
 | 
	
		
			
				|  |  | -  GPR_ASSERT(*xds_client()->server_name_.get() != '\0');
 | 
	
		
			
				|  |  | +  GPR_ASSERT(!xds_client()->server_name_.empty());
 | 
	
		
			
				|  |  |    // Create a call with the specified method name.
 | 
	
		
			
				|  |  |    call_ = grpc_channel_create_pollset_set_call(
 | 
	
		
			
				|  |  |        chand()->channel_, nullptr, GRPC_PROPAGATE_DEFAULTS,
 | 
	
	
		
			
				|  | @@ -531,14 +584,7 @@ XdsClient::ChannelState::AdsCallState::AdsCallState(
 | 
	
		
			
				|  |  |        GRPC_MDSTR_SLASH_ENVOY_DOT_SERVICE_DOT_DISCOVERY_DOT_V2_DOT_AGGREGATEDDISCOVERYSERVICE_SLASH_STREAMAGGREGATEDRESOURCES,
 | 
	
		
			
				|  |  |        nullptr, GRPC_MILLIS_INF_FUTURE, nullptr);
 | 
	
		
			
				|  |  |    GPR_ASSERT(call_ != nullptr);
 | 
	
		
			
				|  |  | -  // Init the request payload.
 | 
	
		
			
				|  |  | -  grpc_slice request_payload_slice = XdsEdsRequestCreateAndEncode(
 | 
	
		
			
				|  |  | -      xds_client()->server_name_.get(), xds_client()->bootstrap_->node(),
 | 
	
		
			
				|  |  | -      xds_client()->build_version_.get());
 | 
	
		
			
				|  |  | -  send_message_payload_ =
 | 
	
		
			
				|  |  | -      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 | 
	
		
			
				|  |  | -  grpc_slice_unref_internal(request_payload_slice);
 | 
	
		
			
				|  |  | -  // Init other data associated with the call.
 | 
	
		
			
				|  |  | +  // Init data associated with the call.
 | 
	
		
			
				|  |  |    grpc_metadata_array_init(&initial_metadata_recv_);
 | 
	
		
			
				|  |  |    grpc_metadata_array_init(&trailing_metadata_recv_);
 | 
	
		
			
				|  |  |    // Start the call.
 | 
	
	
		
			
				|  | @@ -559,16 +605,20 @@ XdsClient::ChannelState::AdsCallState::AdsCallState(
 | 
	
		
			
				|  |  |    op->flags = 0;
 | 
	
		
			
				|  |  |    op->reserved = nullptr;
 | 
	
		
			
				|  |  |    op++;
 | 
	
		
			
				|  |  | -  // Op: send request message.
 | 
	
		
			
				|  |  | -  GPR_ASSERT(send_message_payload_ != nullptr);
 | 
	
		
			
				|  |  | -  op->op = GRPC_OP_SEND_MESSAGE;
 | 
	
		
			
				|  |  | -  op->data.send_message.send_message = send_message_payload_;
 | 
	
		
			
				|  |  | -  op->flags = 0;
 | 
	
		
			
				|  |  | -  op->reserved = nullptr;
 | 
	
		
			
				|  |  | -  op++;
 | 
	
		
			
				|  |  |    call_error = grpc_call_start_batch_and_execute(call_, ops, (size_t)(op - ops),
 | 
	
		
			
				|  |  |                                                   nullptr);
 | 
	
		
			
				|  |  |    GPR_ASSERT(GRPC_CALL_OK == call_error);
 | 
	
		
			
				|  |  | +  // Op: send request message.
 | 
	
		
			
				|  |  | +  GRPC_CLOSURE_INIT(&on_request_sent_, OnRequestSent, this,
 | 
	
		
			
				|  |  | +                    grpc_schedule_on_exec_ctx);
 | 
	
		
			
				|  |  | +  bool initial_message = true;
 | 
	
		
			
				|  |  | +  if (!xds_client()->cluster_map_.empty()) {
 | 
	
		
			
				|  |  | +    SendMessageLocked(kCdsTypeUrl, "", nullptr, initial_message);
 | 
	
		
			
				|  |  | +    initial_message = false;
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  if (!xds_client()->endpoint_map_.empty()) {
 | 
	
		
			
				|  |  | +    SendMessageLocked(kEdsTypeUrl, "", nullptr, initial_message);
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  |    // Op: recv initial metadata.
 | 
	
		
			
				|  |  |    op = ops;
 | 
	
		
			
				|  |  |    op->op = GRPC_OP_RECV_INITIAL_METADATA;
 | 
	
	
		
			
				|  | @@ -629,86 +679,126 @@ void XdsClient::ChannelState::AdsCallState::Orphan() {
 | 
	
		
			
				|  |  |    // corresponding unref happens in on_status_received_ instead of here.
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -void XdsClient::ChannelState::AdsCallState::OnResponseReceived(
 | 
	
		
			
				|  |  | -    void* arg, grpc_error* error) {
 | 
	
		
			
				|  |  | -  AdsCallState* ads_calld = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | -  ads_calld->xds_client()->combiner_->Run(
 | 
	
		
			
				|  |  | -      GRPC_CLOSURE_INIT(&ads_calld->on_response_received_,
 | 
	
		
			
				|  |  | -                        OnResponseReceivedLocked, ads_calld, nullptr),
 | 
	
		
			
				|  |  | -      GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | -}
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -void XdsClient::ChannelState::AdsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  | -    void* arg, grpc_error* /*error*/) {
 | 
	
		
			
				|  |  | -  AdsCallState* ads_calld = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | -  XdsClient* xds_client = ads_calld->xds_client();
 | 
	
		
			
				|  |  | -  // Empty payload means the call was cancelled.
 | 
	
		
			
				|  |  | -  if (!ads_calld->IsCurrentCallOnChannel() ||
 | 
	
		
			
				|  |  | -      ads_calld->recv_message_payload_ == nullptr) {
 | 
	
		
			
				|  |  | -    ads_calld->Unref(DEBUG_LOCATION, "ADS+OnResponseReceivedLocked");
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::SendMessageLocked(
 | 
	
		
			
				|  |  | +    const std::string& type_url, const std::string& nonce_for_unsupported_type,
 | 
	
		
			
				|  |  | +    grpc_error* error_for_unsupported_type, bool is_first_message) {
 | 
	
		
			
				|  |  | +  // Buffer message sending if an existing message is in flight.
 | 
	
		
			
				|  |  | +  if (send_message_payload_ != nullptr) {
 | 
	
		
			
				|  |  | +    buffered_request_map_[type_url].reset(new BufferedRequest(
 | 
	
		
			
				|  |  | +        nonce_for_unsupported_type, error_for_unsupported_type));
 | 
	
		
			
				|  |  |      return;
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  // Read the response.
 | 
	
		
			
				|  |  | -  grpc_byte_buffer_reader bbr;
 | 
	
		
			
				|  |  | -  grpc_byte_buffer_reader_init(&bbr, ads_calld->recv_message_payload_);
 | 
	
		
			
				|  |  | -  grpc_slice response_slice = grpc_byte_buffer_reader_readall(&bbr);
 | 
	
		
			
				|  |  | -  grpc_byte_buffer_reader_destroy(&bbr);
 | 
	
		
			
				|  |  | -  grpc_byte_buffer_destroy(ads_calld->recv_message_payload_);
 | 
	
		
			
				|  |  | -  ads_calld->recv_message_payload_ = nullptr;
 | 
	
		
			
				|  |  | -  // TODO(juanlishen): When we convert this to use the xds protocol, the
 | 
	
		
			
				|  |  | -  // balancer will send us a fallback timeout such that we should go into
 | 
	
		
			
				|  |  | -  // fallback mode if we have lost contact with the balancer after a certain
 | 
	
		
			
				|  |  | -  // period of time. We will need to save the timeout value here, and then
 | 
	
		
			
				|  |  | -  // when the balancer call ends, we will need to start a timer for the
 | 
	
		
			
				|  |  | -  // specified period of time, and if the timer fires, we go into fallback
 | 
	
		
			
				|  |  | -  // mode. We will also need to cancel the timer when we receive a serverlist
 | 
	
		
			
				|  |  | -  // from the balancer.
 | 
	
		
			
				|  |  | -  // This anonymous lambda is a hack to avoid the usage of goto.
 | 
	
		
			
				|  |  | -  [&]() {
 | 
	
		
			
				|  |  | -    // Parse the response.
 | 
	
		
			
				|  |  | -    EdsUpdate update;
 | 
	
		
			
				|  |  | -    grpc_error* parse_error =
 | 
	
		
			
				|  |  | -        XdsEdsResponseDecodeAndParse(response_slice, &update);
 | 
	
		
			
				|  |  | -    if (parse_error != GRPC_ERROR_NONE) {
 | 
	
		
			
				|  |  | -      gpr_log(GPR_ERROR,
 | 
	
		
			
				|  |  | -              "[xds_client %p] ADS response parsing failed. error=%s",
 | 
	
		
			
				|  |  | -              xds_client, grpc_error_string(parse_error));
 | 
	
		
			
				|  |  | -      GRPC_ERROR_UNREF(parse_error);
 | 
	
		
			
				|  |  | -      return;
 | 
	
		
			
				|  |  | +  grpc_slice request_payload_slice;
 | 
	
		
			
				|  |  | +  const XdsBootstrap::Node* node =
 | 
	
		
			
				|  |  | +      is_first_message ? xds_client()->bootstrap_->node() : nullptr;
 | 
	
		
			
				|  |  | +  const char* build_version =
 | 
	
		
			
				|  |  | +      is_first_message ? xds_client()->build_version_.get() : nullptr;
 | 
	
		
			
				|  |  | +  if (type_url == kCdsTypeUrl) {
 | 
	
		
			
				|  |  | +    request_payload_slice = XdsCdsRequestCreateAndEncode(
 | 
	
		
			
				|  |  | +        xds_client()->WatchedClusterNames(), node, build_version,
 | 
	
		
			
				|  |  | +        cds_version_.version_info, cds_version_.nonce, cds_version_.error);
 | 
	
		
			
				|  |  | +    cds_version_.error = GRPC_ERROR_NONE;
 | 
	
		
			
				|  |  | +    GRPC_ERROR_UNREF(error_for_unsupported_type);
 | 
	
		
			
				|  |  | +  } else if (type_url == kEdsTypeUrl) {
 | 
	
		
			
				|  |  | +    request_payload_slice = XdsEdsRequestCreateAndEncode(
 | 
	
		
			
				|  |  | +        xds_client()->EdsServiceNames(), node, build_version,
 | 
	
		
			
				|  |  | +        eds_version_.version_info, eds_version_.nonce, eds_version_.error);
 | 
	
		
			
				|  |  | +    eds_version_.error = GRPC_ERROR_NONE;
 | 
	
		
			
				|  |  | +    GRPC_ERROR_UNREF(error_for_unsupported_type);
 | 
	
		
			
				|  |  | +  } else {
 | 
	
		
			
				|  |  | +    request_payload_slice = XdsUnsupportedTypeNackRequestCreateAndEncode(
 | 
	
		
			
				|  |  | +        type_url, nonce_for_unsupported_type, error_for_unsupported_type);
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  // Create message payload.
 | 
	
		
			
				|  |  | +  send_message_payload_ =
 | 
	
		
			
				|  |  | +      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 | 
	
		
			
				|  |  | +  grpc_slice_unref_internal(request_payload_slice);
 | 
	
		
			
				|  |  | +  // Send the message.
 | 
	
		
			
				|  |  | +  grpc_op op;
 | 
	
		
			
				|  |  | +  memset(&op, 0, sizeof(op));
 | 
	
		
			
				|  |  | +  op.op = GRPC_OP_SEND_MESSAGE;
 | 
	
		
			
				|  |  | +  op.data.send_message.send_message = send_message_payload_;
 | 
	
		
			
				|  |  | +  Ref(DEBUG_LOCATION, "ADS+OnRequestSentLocked").release();
 | 
	
		
			
				|  |  | +  GRPC_CLOSURE_INIT(&on_request_sent_, OnRequestSent, this,
 | 
	
		
			
				|  |  | +                    grpc_schedule_on_exec_ctx);
 | 
	
		
			
				|  |  | +  grpc_call_error call_error =
 | 
	
		
			
				|  |  | +      grpc_call_start_batch_and_execute(call_, &op, 1, &on_request_sent_);
 | 
	
		
			
				|  |  | +  if (GPR_UNLIKELY(call_error != GRPC_CALL_OK)) {
 | 
	
		
			
				|  |  | +    gpr_log(GPR_ERROR,
 | 
	
		
			
				|  |  | +            "[xds_client %p] calld=%p call_error=%d sending ADS message",
 | 
	
		
			
				|  |  | +            xds_client(), this, call_error);
 | 
	
		
			
				|  |  | +    GPR_ASSERT(GRPC_CALL_OK == call_error);
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::AcceptCdsUpdate(
 | 
	
		
			
				|  |  | +    CdsUpdateMap cds_update_map, std::string new_version) {
 | 
	
		
			
				|  |  | +  for (auto& p : cds_update_map) {
 | 
	
		
			
				|  |  | +    const char* cluster_name = p.first.c_str();
 | 
	
		
			
				|  |  | +    CdsUpdate& cds_update = p.second;
 | 
	
		
			
				|  |  | +    if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  | +      gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  | +              "[xds_client %p] CDS update (cluster=%s) received: "
 | 
	
		
			
				|  |  | +              "eds_service_name=%s, "
 | 
	
		
			
				|  |  | +              "lrs_load_reporting_server_name=%s",
 | 
	
		
			
				|  |  | +              xds_client(), cluster_name, cds_update.eds_service_name.c_str(),
 | 
	
		
			
				|  |  | +              cds_update.lrs_load_reporting_server_name.has_value()
 | 
	
		
			
				|  |  | +                  ? cds_update.lrs_load_reporting_server_name.value().c_str()
 | 
	
		
			
				|  |  | +                  : "(N/A)");
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  | -    if (update.priority_list_update.empty() && !update.drop_all) {
 | 
	
		
			
				|  |  | -      char* response_slice_str =
 | 
	
		
			
				|  |  | -          grpc_dump_slice(response_slice, GPR_DUMP_ASCII | GPR_DUMP_HEX);
 | 
	
		
			
				|  |  | -      gpr_log(GPR_ERROR,
 | 
	
		
			
				|  |  | -              "[xds_client %p] ADS response '%s' doesn't contain any valid "
 | 
	
		
			
				|  |  | -              "locality but doesn't require to drop all calls. Ignoring.",
 | 
	
		
			
				|  |  | -              xds_client, response_slice_str);
 | 
	
		
			
				|  |  | -      gpr_free(response_slice_str);
 | 
	
		
			
				|  |  | -      return;
 | 
	
		
			
				|  |  | +    ClusterState& cluster_state = xds_client()->cluster_map_[cluster_name];
 | 
	
		
			
				|  |  | +    // Ignore identical update.
 | 
	
		
			
				|  |  | +    if (cluster_state.update.has_value() &&
 | 
	
		
			
				|  |  | +        cds_update.eds_service_name ==
 | 
	
		
			
				|  |  | +            cluster_state.update.value().eds_service_name &&
 | 
	
		
			
				|  |  | +        cds_update.lrs_load_reporting_server_name.value() ==
 | 
	
		
			
				|  |  | +            cluster_state.update.value()
 | 
	
		
			
				|  |  | +                .lrs_load_reporting_server_name.value()) {
 | 
	
		
			
				|  |  | +      if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  | +        gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  | +                "[xds_client %p] CDS update identical to current, ignoring.",
 | 
	
		
			
				|  |  | +                xds_client());
 | 
	
		
			
				|  |  | +      }
 | 
	
		
			
				|  |  | +      continue;
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +    // Update the cluster state.
 | 
	
		
			
				|  |  | +    cluster_state.update.set(std::move(cds_update));
 | 
	
		
			
				|  |  | +    // Notify all watchers.
 | 
	
		
			
				|  |  | +    for (const auto& p : cluster_state.watchers) {
 | 
	
		
			
				|  |  | +      p.first->OnClusterChanged(cluster_state.update.value());
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  | -    ads_calld->seen_response_ = true;
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  cds_version_.version_info = std::move(new_version);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::AcceptEdsUpdate(
 | 
	
		
			
				|  |  | +    EdsUpdateMap eds_update_map, std::string new_version) {
 | 
	
		
			
				|  |  | +  for (auto& p : eds_update_map) {
 | 
	
		
			
				|  |  | +    const char* eds_service_name = p.first.c_str();
 | 
	
		
			
				|  |  | +    EdsUpdate& eds_update = p.second;
 | 
	
		
			
				|  |  |      if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  |        gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  | -              "[xds_client %p] ADS response with %" PRIuPTR
 | 
	
		
			
				|  |  | +              "[xds_client %p] EDS response with %" PRIuPTR
 | 
	
		
			
				|  |  |                " priorities and %" PRIuPTR
 | 
	
		
			
				|  |  |                " drop categories received (drop_all=%d)",
 | 
	
		
			
				|  |  | -              xds_client, update.priority_list_update.size(),
 | 
	
		
			
				|  |  | -              update.drop_config->drop_category_list().size(), update.drop_all);
 | 
	
		
			
				|  |  | -      for (size_t priority = 0; priority < update.priority_list_update.size();
 | 
	
		
			
				|  |  | -           ++priority) {
 | 
	
		
			
				|  |  | -        const auto* locality_map_update =
 | 
	
		
			
				|  |  | -            update.priority_list_update.Find(static_cast<uint32_t>(priority));
 | 
	
		
			
				|  |  | +              xds_client(), eds_update.priority_list_update.size(),
 | 
	
		
			
				|  |  | +              eds_update.drop_config->drop_category_list().size(),
 | 
	
		
			
				|  |  | +              eds_update.drop_all);
 | 
	
		
			
				|  |  | +      for (size_t priority = 0;
 | 
	
		
			
				|  |  | +           priority < eds_update.priority_list_update.size(); ++priority) {
 | 
	
		
			
				|  |  | +        const auto* locality_map_update = eds_update.priority_list_update.Find(
 | 
	
		
			
				|  |  | +            static_cast<uint32_t>(priority));
 | 
	
		
			
				|  |  |          gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                  "[xds_client %p] Priority %" PRIuPTR " contains %" PRIuPTR
 | 
	
		
			
				|  |  |                  " localities",
 | 
	
		
			
				|  |  | -                xds_client, priority, locality_map_update->size());
 | 
	
		
			
				|  |  | +                xds_client(), priority, locality_map_update->size());
 | 
	
		
			
				|  |  |          size_t locality_count = 0;
 | 
	
		
			
				|  |  |          for (const auto& p : locality_map_update->localities) {
 | 
	
		
			
				|  |  |            const auto& locality = p.second;
 | 
	
		
			
				|  |  |            gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                    "[xds_client %p] Priority %" PRIuPTR ", locality %" PRIuPTR
 | 
	
		
			
				|  |  |                    " %s contains %" PRIuPTR " server addresses",
 | 
	
		
			
				|  |  | -                  xds_client, priority, locality_count,
 | 
	
		
			
				|  |  | +                  xds_client(), priority, locality_count,
 | 
	
		
			
				|  |  |                    locality.name->AsHumanReadableString(),
 | 
	
		
			
				|  |  |                    locality.serverlist.size());
 | 
	
		
			
				|  |  |            for (size_t i = 0; i < locality.serverlist.size(); ++i) {
 | 
	
	
		
			
				|  | @@ -718,59 +808,184 @@ void XdsClient::ChannelState::AdsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  |              gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                      "[xds_client %p] Priority %" PRIuPTR ", locality %" PRIuPTR
 | 
	
		
			
				|  |  |                      " %s, server address %" PRIuPTR ": %s",
 | 
	
		
			
				|  |  | -                    xds_client, priority, locality_count,
 | 
	
		
			
				|  |  | +                    xds_client(), priority, locality_count,
 | 
	
		
			
				|  |  |                      locality.name->AsHumanReadableString(), i, ipport);
 | 
	
		
			
				|  |  |              gpr_free(ipport);
 | 
	
		
			
				|  |  |            }
 | 
	
		
			
				|  |  |            ++locality_count;
 | 
	
		
			
				|  |  |          }
 | 
	
		
			
				|  |  |        }
 | 
	
		
			
				|  |  | -      for (size_t i = 0; i < update.drop_config->drop_category_list().size();
 | 
	
		
			
				|  |  | -           ++i) {
 | 
	
		
			
				|  |  | +      for (size_t i = 0;
 | 
	
		
			
				|  |  | +           i < eds_update.drop_config->drop_category_list().size(); ++i) {
 | 
	
		
			
				|  |  |          const XdsDropConfig::DropCategory& drop_category =
 | 
	
		
			
				|  |  | -            update.drop_config->drop_category_list()[i];
 | 
	
		
			
				|  |  | +            eds_update.drop_config->drop_category_list()[i];
 | 
	
		
			
				|  |  |          gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                  "[xds_client %p] Drop category %s has drop rate %d per million",
 | 
	
		
			
				|  |  | -                xds_client, drop_category.name.get(),
 | 
	
		
			
				|  |  | +                xds_client(), drop_category.name.c_str(),
 | 
	
		
			
				|  |  |                  drop_category.parts_per_million);
 | 
	
		
			
				|  |  |        }
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  | -    // Start load reporting if needed.
 | 
	
		
			
				|  |  | -    auto& lrs_call = ads_calld->chand()->lrs_calld_;
 | 
	
		
			
				|  |  | -    if (lrs_call != nullptr) {
 | 
	
		
			
				|  |  | -      LrsCallState* lrs_calld = lrs_call->calld();
 | 
	
		
			
				|  |  | -      if (lrs_calld != nullptr) lrs_calld->MaybeStartReportingLocked();
 | 
	
		
			
				|  |  | -    }
 | 
	
		
			
				|  |  | +    EndpointState& endpoint_state =
 | 
	
		
			
				|  |  | +        xds_client()->endpoint_map_[eds_service_name];
 | 
	
		
			
				|  |  |      // Ignore identical update.
 | 
	
		
			
				|  |  | -    const EdsUpdate& prev_update = xds_client->cluster_state_.eds_update;
 | 
	
		
			
				|  |  | +    const EdsUpdate& prev_update = endpoint_state.update;
 | 
	
		
			
				|  |  |      const bool priority_list_changed =
 | 
	
		
			
				|  |  | -        prev_update.priority_list_update != update.priority_list_update;
 | 
	
		
			
				|  |  | +        prev_update.priority_list_update != eds_update.priority_list_update;
 | 
	
		
			
				|  |  |      const bool drop_config_changed =
 | 
	
		
			
				|  |  |          prev_update.drop_config == nullptr ||
 | 
	
		
			
				|  |  | -        *prev_update.drop_config != *update.drop_config;
 | 
	
		
			
				|  |  | +        *prev_update.drop_config != *eds_update.drop_config;
 | 
	
		
			
				|  |  |      if (!priority_list_changed && !drop_config_changed) {
 | 
	
		
			
				|  |  |        if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  |          gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                  "[xds_client %p] EDS update identical to current, ignoring.",
 | 
	
		
			
				|  |  | -                xds_client);
 | 
	
		
			
				|  |  | +                xds_client());
 | 
	
		
			
				|  |  |        }
 | 
	
		
			
				|  |  | -      return;
 | 
	
		
			
				|  |  | +      continue;
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  |      // Update the cluster state.
 | 
	
		
			
				|  |  | -    ClusterState& cluster_state = xds_client->cluster_state_;
 | 
	
		
			
				|  |  | -    cluster_state.eds_update = std::move(update);
 | 
	
		
			
				|  |  | +    endpoint_state.update = std::move(eds_update);
 | 
	
		
			
				|  |  |      // Notify all watchers.
 | 
	
		
			
				|  |  | -    for (const auto& p : cluster_state.endpoint_watchers) {
 | 
	
		
			
				|  |  | -      p.first->OnEndpointChanged(cluster_state.eds_update);
 | 
	
		
			
				|  |  | +    for (const auto& p : endpoint_state.watchers) {
 | 
	
		
			
				|  |  | +      p.first->OnEndpointChanged(endpoint_state.update);
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  | -  }();
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  eds_version_.version_info = std::move(new_version);
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::OnRequestSent(void* arg,
 | 
	
		
			
				|  |  | +                                                          grpc_error* error) {
 | 
	
		
			
				|  |  | +  AdsCallState* ads_calld = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | +  ads_calld->xds_client()->combiner_->Run(
 | 
	
		
			
				|  |  | +      GRPC_CLOSURE_INIT(&ads_calld->on_request_sent_, OnRequestSentLocked,
 | 
	
		
			
				|  |  | +                        ads_calld, nullptr),
 | 
	
		
			
				|  |  | +      GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::OnRequestSentLocked(
 | 
	
		
			
				|  |  | +    void* arg, grpc_error* error) {
 | 
	
		
			
				|  |  | +  AdsCallState* self = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | +  if (self->IsCurrentCallOnChannel() && error == GRPC_ERROR_NONE) {
 | 
	
		
			
				|  |  | +    // Clean up the sent message.
 | 
	
		
			
				|  |  | +    grpc_byte_buffer_destroy(self->send_message_payload_);
 | 
	
		
			
				|  |  | +    self->send_message_payload_ = nullptr;
 | 
	
		
			
				|  |  | +    // Continue to send another pending message if any.
 | 
	
		
			
				|  |  | +    // TODO(roth): The current code to handle buffered messages has the
 | 
	
		
			
				|  |  | +    // advantage of sending only the most recent list of resource names for each
 | 
	
		
			
				|  |  | +    // resource type (no matter how many times that resource type has been
 | 
	
		
			
				|  |  | +    // requested to send while the current message sending is still pending).
 | 
	
		
			
				|  |  | +    // But its disadvantage is that we send the requests in fixed order of
 | 
	
		
			
				|  |  | +    // resource types. We need to fix this if we are seeing some resource
 | 
	
		
			
				|  |  | +    // type(s) starved due to frequent requests of other resource type(s).
 | 
	
		
			
				|  |  | +    for (auto& p : self->buffered_request_map_) {
 | 
	
		
			
				|  |  | +      const std::string& type_url = p.first;
 | 
	
		
			
				|  |  | +      std::unique_ptr<BufferedRequest>& buffered_request = p.second;
 | 
	
		
			
				|  |  | +      if (buffered_request != nullptr) {
 | 
	
		
			
				|  |  | +        self->SendMessageLocked(type_url, buffered_request->nonce,
 | 
	
		
			
				|  |  | +                                buffered_request->error, false);
 | 
	
		
			
				|  |  | +        buffered_request->error = GRPC_ERROR_NONE;
 | 
	
		
			
				|  |  | +        buffered_request.reset();
 | 
	
		
			
				|  |  | +        break;
 | 
	
		
			
				|  |  | +      }
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  self->Unref(DEBUG_LOCATION, "ADS+OnRequestSentLocked");
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::OnResponseReceived(
 | 
	
		
			
				|  |  | +    void* arg, grpc_error* error) {
 | 
	
		
			
				|  |  | +  AdsCallState* ads_calld = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | +  ads_calld->xds_client()->combiner_->Run(
 | 
	
		
			
				|  |  | +      GRPC_CLOSURE_INIT(&ads_calld->on_response_received_,
 | 
	
		
			
				|  |  | +                        OnResponseReceivedLocked, ads_calld, nullptr),
 | 
	
		
			
				|  |  | +      GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +void XdsClient::ChannelState::AdsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  | +    void* arg, grpc_error* /*error*/) {
 | 
	
		
			
				|  |  | +  AdsCallState* ads_calld = static_cast<AdsCallState*>(arg);
 | 
	
		
			
				|  |  | +  XdsClient* xds_client = ads_calld->xds_client();
 | 
	
		
			
				|  |  | +  // Empty payload means the call was cancelled.
 | 
	
		
			
				|  |  | +  if (!ads_calld->IsCurrentCallOnChannel() ||
 | 
	
		
			
				|  |  | +      ads_calld->recv_message_payload_ == nullptr) {
 | 
	
		
			
				|  |  | +    ads_calld->Unref(DEBUG_LOCATION, "ADS+OnResponseReceivedLocked");
 | 
	
		
			
				|  |  | +    return;
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  // Read the response.
 | 
	
		
			
				|  |  | +  grpc_byte_buffer_reader bbr;
 | 
	
		
			
				|  |  | +  grpc_byte_buffer_reader_init(&bbr, ads_calld->recv_message_payload_);
 | 
	
		
			
				|  |  | +  grpc_slice response_slice = grpc_byte_buffer_reader_readall(&bbr);
 | 
	
		
			
				|  |  | +  grpc_byte_buffer_reader_destroy(&bbr);
 | 
	
		
			
				|  |  | +  grpc_byte_buffer_destroy(ads_calld->recv_message_payload_);
 | 
	
		
			
				|  |  | +  ads_calld->recv_message_payload_ = nullptr;
 | 
	
		
			
				|  |  | +  // TODO(juanlishen): When we convert this to use the xds protocol, the
 | 
	
		
			
				|  |  | +  // balancer will send us a fallback timeout such that we should go into
 | 
	
		
			
				|  |  | +  // fallback mode if we have lost contact with the balancer after a certain
 | 
	
		
			
				|  |  | +  // period of time. We will need to save the timeout value here, and then
 | 
	
		
			
				|  |  | +  // when the balancer call ends, we will need to start a timer for the
 | 
	
		
			
				|  |  | +  // specified period of time, and if the timer fires, we go into fallback
 | 
	
		
			
				|  |  | +  // mode. We will also need to cancel the timer when we receive a serverlist
 | 
	
		
			
				|  |  | +  // from the balancer.
 | 
	
		
			
				|  |  | +  // Parse the response.
 | 
	
		
			
				|  |  | +  CdsUpdateMap cds_update_map;
 | 
	
		
			
				|  |  | +  EdsUpdateMap eds_update_map;
 | 
	
		
			
				|  |  | +  std::string version;
 | 
	
		
			
				|  |  | +  std::string nonce;
 | 
	
		
			
				|  |  | +  std::string type_url;
 | 
	
		
			
				|  |  | +  // Note that XdsAdsResponseDecodeAndParse() also validate the response.
 | 
	
		
			
				|  |  | +  grpc_error* parse_error = XdsAdsResponseDecodeAndParse(
 | 
	
		
			
				|  |  | +      response_slice, xds_client->EdsServiceNames(), &cds_update_map,
 | 
	
		
			
				|  |  | +      &eds_update_map, &version, &nonce, &type_url);
 | 
	
		
			
				|  |  |    grpc_slice_unref_internal(response_slice);
 | 
	
		
			
				|  |  | +  if (type_url.empty()) {
 | 
	
		
			
				|  |  | +    // Ignore unparsable response.
 | 
	
		
			
				|  |  | +    gpr_log(GPR_ERROR, "[xds_client %p] No type_url found. error=%s",
 | 
	
		
			
				|  |  | +            xds_client, grpc_error_string(parse_error));
 | 
	
		
			
				|  |  | +    GRPC_ERROR_UNREF(parse_error);
 | 
	
		
			
				|  |  | +  } else {
 | 
	
		
			
				|  |  | +    // Update nonce and error.
 | 
	
		
			
				|  |  | +    if (type_url == kCdsTypeUrl) {
 | 
	
		
			
				|  |  | +      ads_calld->cds_version_.nonce = nonce;
 | 
	
		
			
				|  |  | +      GRPC_ERROR_UNREF(ads_calld->cds_version_.error);
 | 
	
		
			
				|  |  | +      ads_calld->cds_version_.error = GRPC_ERROR_REF(parse_error);
 | 
	
		
			
				|  |  | +    } else if (type_url == kEdsTypeUrl) {
 | 
	
		
			
				|  |  | +      ads_calld->eds_version_.nonce = nonce;
 | 
	
		
			
				|  |  | +      GRPC_ERROR_UNREF(ads_calld->eds_version_.error);
 | 
	
		
			
				|  |  | +      ads_calld->eds_version_.error = GRPC_ERROR_REF(parse_error);
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +    // NACK or ACK the response.
 | 
	
		
			
				|  |  | +    if (parse_error != GRPC_ERROR_NONE) {
 | 
	
		
			
				|  |  | +      // NACK unacceptable update.
 | 
	
		
			
				|  |  | +      gpr_log(
 | 
	
		
			
				|  |  | +          GPR_ERROR,
 | 
	
		
			
				|  |  | +          "[xds_client %p] ADS response can't be accepted, NACKing. error=%s",
 | 
	
		
			
				|  |  | +          xds_client, grpc_error_string(parse_error));
 | 
	
		
			
				|  |  | +      ads_calld->SendMessageLocked(type_url, nonce, parse_error, false);
 | 
	
		
			
				|  |  | +    } else {
 | 
	
		
			
				|  |  | +      ads_calld->seen_response_ = true;
 | 
	
		
			
				|  |  | +      // Accept the (CDS or EDS) response.
 | 
	
		
			
				|  |  | +      if (type_url == kCdsTypeUrl) {
 | 
	
		
			
				|  |  | +        ads_calld->AcceptCdsUpdate(std::move(cds_update_map),
 | 
	
		
			
				|  |  | +                                   std::move(version));
 | 
	
		
			
				|  |  | +      } else if (type_url == kEdsTypeUrl) {
 | 
	
		
			
				|  |  | +        ads_calld->AcceptEdsUpdate(std::move(eds_update_map),
 | 
	
		
			
				|  |  | +                                   std::move(version));
 | 
	
		
			
				|  |  | +      }
 | 
	
		
			
				|  |  | +      // ACK the update.
 | 
	
		
			
				|  |  | +      ads_calld->SendMessageLocked(type_url, "", nullptr, false);
 | 
	
		
			
				|  |  | +      // Start load reporting if needed.
 | 
	
		
			
				|  |  | +      auto& lrs_call = ads_calld->chand()->lrs_calld_;
 | 
	
		
			
				|  |  | +      if (lrs_call != nullptr) {
 | 
	
		
			
				|  |  | +        LrsCallState* lrs_calld = lrs_call->calld();
 | 
	
		
			
				|  |  | +        if (lrs_calld != nullptr) lrs_calld->MaybeStartReportingLocked();
 | 
	
		
			
				|  |  | +      }
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  |    if (xds_client->shutting_down_) {
 | 
	
		
			
				|  |  |      ads_calld->Unref(DEBUG_LOCATION,
 | 
	
		
			
				|  |  |                       "ADS+OnResponseReceivedLocked+xds_shutdown");
 | 
	
		
			
				|  |  |      return;
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  // Keep listening for serverlist updates.
 | 
	
		
			
				|  |  | +  // Keep listening for updates.
 | 
	
		
			
				|  |  |    grpc_op op;
 | 
	
		
			
				|  |  |    memset(&op, 0, sizeof(op));
 | 
	
		
			
				|  |  |    op.op = GRPC_OP_RECV_MESSAGE;
 | 
	
	
		
			
				|  | @@ -869,15 +1084,8 @@ void XdsClient::ChannelState::LrsCallState::Reporter::OnNextReportTimerLocked(
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  void XdsClient::ChannelState::LrsCallState::Reporter::SendReportLocked() {
 | 
	
		
			
				|  |  |    // Create a request that contains the load report.
 | 
	
		
			
				|  |  | -  // TODO(roth): Currently, it is not possible to have multiple client
 | 
	
		
			
				|  |  | -  // stats objects for a given cluster.  However, in the future, we may
 | 
	
		
			
				|  |  | -  // run into cases where this happens (e.g., due to graceful LB policy
 | 
	
		
			
				|  |  | -  // switching).  If/when this becomes a problem, replace this assertion
 | 
	
		
			
				|  |  | -  // with code to merge data from multiple client stats objects.
 | 
	
		
			
				|  |  | -  GPR_ASSERT(xds_client()->cluster_state_.client_stats.size() == 1);
 | 
	
		
			
				|  |  | -  auto* client_stats = *xds_client()->cluster_state_.client_stats.begin();
 | 
	
		
			
				|  |  |    grpc_slice request_payload_slice =
 | 
	
		
			
				|  |  | -      XdsLrsRequestCreateAndEncode(parent_->cluster_name_.get(), client_stats);
 | 
	
		
			
				|  |  | +      XdsLrsRequestCreateAndEncode(xds_client()->ClientStatsMap());
 | 
	
		
			
				|  |  |    // Skip client load report if the counters were all zero in the last
 | 
	
		
			
				|  |  |    // report and they are still zero in this one.
 | 
	
		
			
				|  |  |    const bool old_val = last_report_counters_were_zero_;
 | 
	
	
		
			
				|  | @@ -945,8 +1153,7 @@ XdsClient::ChannelState::LrsCallState::LrsCallState(
 | 
	
		
			
				|  |  |    // activity in xds_client()->interested_parties_, which is comprised of
 | 
	
		
			
				|  |  |    // the polling entities from client_channel.
 | 
	
		
			
				|  |  |    GPR_ASSERT(xds_client() != nullptr);
 | 
	
		
			
				|  |  | -  GPR_ASSERT(xds_client()->server_name_ != nullptr);
 | 
	
		
			
				|  |  | -  GPR_ASSERT(*xds_client()->server_name_.get() != '\0');
 | 
	
		
			
				|  |  | +  GPR_ASSERT(!xds_client()->server_name_.empty());
 | 
	
		
			
				|  |  |    call_ = grpc_channel_create_pollset_set_call(
 | 
	
		
			
				|  |  |        chand()->channel_, nullptr, GRPC_PROPAGATE_DEFAULTS,
 | 
	
		
			
				|  |  |        xds_client()->interested_parties_,
 | 
	
	
		
			
				|  | @@ -955,7 +1162,7 @@ XdsClient::ChannelState::LrsCallState::LrsCallState(
 | 
	
		
			
				|  |  |    GPR_ASSERT(call_ != nullptr);
 | 
	
		
			
				|  |  |    // Init the request payload.
 | 
	
		
			
				|  |  |    grpc_slice request_payload_slice = XdsLrsRequestCreateAndEncode(
 | 
	
		
			
				|  |  | -      xds_client()->server_name_.get(), xds_client()->bootstrap_->node(),
 | 
	
		
			
				|  |  | +      xds_client()->server_name_, xds_client()->bootstrap_->node(),
 | 
	
		
			
				|  |  |        xds_client()->build_version_.get());
 | 
	
		
			
				|  |  |    send_message_payload_ =
 | 
	
		
			
				|  |  |        grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 | 
	
	
		
			
				|  | @@ -1069,13 +1276,22 @@ void XdsClient::ChannelState::LrsCallState::MaybeStartReportingLocked() {
 | 
	
		
			
				|  |  |    AdsCallState* ads_calld = chand()->ads_calld_->calld();
 | 
	
		
			
				|  |  |    if (ads_calld == nullptr || !ads_calld->seen_response()) return;
 | 
	
		
			
				|  |  |    // Start reporting.
 | 
	
		
			
				|  |  | -  for (auto* client_stats : chand()->xds_client_->cluster_state_.client_stats) {
 | 
	
		
			
				|  |  | -    client_stats->MaybeInitLastReportTime();
 | 
	
		
			
				|  |  | +  for (auto& p : chand()->xds_client_->endpoint_map_) {
 | 
	
		
			
				|  |  | +    for (auto* client_stats : p.second.client_stats) {
 | 
	
		
			
				|  |  | +      client_stats->MaybeInitLastReportTime();
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  |    reporter_ = MakeOrphanable<Reporter>(
 | 
	
		
			
				|  |  |        Ref(DEBUG_LOCATION, "LRS+load_report+start"), load_reporting_interval_);
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | +bool XdsClient::ChannelState::LrsCallState::ShouldSendLoadReports(
 | 
	
		
			
				|  |  | +    const StringView& cluster_name) const {
 | 
	
		
			
				|  |  | +  // Only send load reports for the clusters that are asked for by the LRS
 | 
	
		
			
				|  |  | +  // server.
 | 
	
		
			
				|  |  | +  return cluster_names_.find(std::string(cluster_name)) != cluster_names_.end();
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |  void XdsClient::ChannelState::LrsCallState::OnInitialRequestSent(
 | 
	
		
			
				|  |  |      void* arg, grpc_error* error) {
 | 
	
		
			
				|  |  |    LrsCallState* lrs_calld = static_cast<LrsCallState*>(arg);
 | 
	
	
		
			
				|  | @@ -1124,10 +1340,10 @@ void XdsClient::ChannelState::LrsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  |    // This anonymous lambda is a hack to avoid the usage of goto.
 | 
	
		
			
				|  |  |    [&]() {
 | 
	
		
			
				|  |  |      // Parse the response.
 | 
	
		
			
				|  |  | -    grpc_core::UniquePtr<char> new_cluster_name;
 | 
	
		
			
				|  |  | +    std::set<std::string> new_cluster_names;
 | 
	
		
			
				|  |  |      grpc_millis new_load_reporting_interval;
 | 
	
		
			
				|  |  |      grpc_error* parse_error = XdsLrsResponseDecodeAndParse(
 | 
	
		
			
				|  |  | -        response_slice, &new_cluster_name, &new_load_reporting_interval);
 | 
	
		
			
				|  |  | +        response_slice, &new_cluster_names, &new_load_reporting_interval);
 | 
	
		
			
				|  |  |      if (parse_error != GRPC_ERROR_NONE) {
 | 
	
		
			
				|  |  |        gpr_log(GPR_ERROR,
 | 
	
		
			
				|  |  |                "[xds_client %p] LRS response parsing failed. error=%s",
 | 
	
	
		
			
				|  | @@ -1138,9 +1354,15 @@ void XdsClient::ChannelState::LrsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  |      lrs_calld->seen_response_ = true;
 | 
	
		
			
				|  |  |      if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  |        gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  | -              "[xds_client %p] LRS response received, cluster_name=%s, "
 | 
	
		
			
				|  |  | -              "load_report_interval=%" PRId64 "ms",
 | 
	
		
			
				|  |  | -              xds_client, new_cluster_name.get(), new_load_reporting_interval);
 | 
	
		
			
				|  |  | +              "[xds_client %p] LRS response received, %" PRIuPTR
 | 
	
		
			
				|  |  | +              " cluster names, load_report_interval=%" PRId64 "ms",
 | 
	
		
			
				|  |  | +              xds_client, new_cluster_names.size(),
 | 
	
		
			
				|  |  | +              new_load_reporting_interval);
 | 
	
		
			
				|  |  | +      size_t i = 0;
 | 
	
		
			
				|  |  | +      for (const auto& name : new_cluster_names) {
 | 
	
		
			
				|  |  | +        gpr_log(GPR_INFO, "[xds_client %p] cluster_name %" PRIuPTR ": %s",
 | 
	
		
			
				|  |  | +                xds_client, i++, name.c_str());
 | 
	
		
			
				|  |  | +      }
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  |      if (new_load_reporting_interval <
 | 
	
		
			
				|  |  |          GRPC_XDS_MIN_CLIENT_LOAD_REPORTING_INTERVAL_MS) {
 | 
	
	
		
			
				|  | @@ -1154,8 +1376,8 @@ void XdsClient::ChannelState::LrsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  |        }
 | 
	
		
			
				|  |  |      }
 | 
	
		
			
				|  |  |      // Ignore identical update.
 | 
	
		
			
				|  |  | -    if (lrs_calld->load_reporting_interval_ == new_load_reporting_interval &&
 | 
	
		
			
				|  |  | -        strcmp(lrs_calld->cluster_name_.get(), new_cluster_name.get()) == 0) {
 | 
	
		
			
				|  |  | +    if (lrs_calld->cluster_names_ == new_cluster_names &&
 | 
	
		
			
				|  |  | +        lrs_calld->load_reporting_interval_ == new_load_reporting_interval) {
 | 
	
		
			
				|  |  |        if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
		
			
				|  |  |          gpr_log(GPR_INFO,
 | 
	
		
			
				|  |  |                  "[xds_client %p] Incoming LRS response identical to current, "
 | 
	
	
		
			
				|  | @@ -1167,7 +1389,7 @@ void XdsClient::ChannelState::LrsCallState::OnResponseReceivedLocked(
 | 
	
		
			
				|  |  |      // Stop current load reporting (if any) to adopt the new config.
 | 
	
		
			
				|  |  |      lrs_calld->reporter_.reset();
 | 
	
		
			
				|  |  |      // Record the new config.
 | 
	
		
			
				|  |  | -    lrs_calld->cluster_name_ = std::move(new_cluster_name);
 | 
	
		
			
				|  |  | +    lrs_calld->cluster_names_ = std::move(new_cluster_names);
 | 
	
		
			
				|  |  |      lrs_calld->load_reporting_interval_ = new_load_reporting_interval;
 | 
	
		
			
				|  |  |      // Try starting sending load report.
 | 
	
		
			
				|  |  |      lrs_calld->MaybeStartReportingLocked();
 | 
	
	
		
			
				|  | @@ -1253,11 +1475,12 @@ XdsClient::XdsClient(Combiner* combiner, grpc_pollset_set* interested_parties,
 | 
	
		
			
				|  |  |                       StringView server_name,
 | 
	
		
			
				|  |  |                       std::unique_ptr<ServiceConfigWatcherInterface> watcher,
 | 
	
		
			
				|  |  |                       const grpc_channel_args& channel_args, grpc_error** error)
 | 
	
		
			
				|  |  | -    : build_version_(GenerateBuildVersionString()),
 | 
	
		
			
				|  |  | +    : InternallyRefCounted<XdsClient>(&grpc_xds_client_trace),
 | 
	
		
			
				|  |  | +      build_version_(GenerateBuildVersionString()),
 | 
	
		
			
				|  |  |        combiner_(GRPC_COMBINER_REF(combiner, "xds_client")),
 | 
	
		
			
				|  |  |        interested_parties_(interested_parties),
 | 
	
		
			
				|  |  |        bootstrap_(XdsBootstrap::ReadFromFile(error)),
 | 
	
		
			
				|  |  | -      server_name_(StringViewToCString(server_name)),
 | 
	
		
			
				|  |  | +      server_name_(server_name),
 | 
	
		
			
				|  |  |        service_config_watcher_(std::move(watcher)) {
 | 
	
		
			
				|  |  |    if (*error != GRPC_ERROR_NONE) {
 | 
	
		
			
				|  |  |      if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
 | 
	
	
		
			
				|  | @@ -1286,77 +1509,95 @@ XdsClient::~XdsClient() { GRPC_COMBINER_UNREF(combiner_, "xds_client"); }
 | 
	
		
			
				|  |  |  void XdsClient::Orphan() {
 | 
	
		
			
				|  |  |    shutting_down_ = true;
 | 
	
		
			
				|  |  |    chand_.reset();
 | 
	
		
			
				|  |  | +  cluster_map_.clear();
 | 
	
		
			
				|  |  | +  endpoint_map_.clear();
 | 
	
		
			
				|  |  |    Unref(DEBUG_LOCATION, "XdsClient::Orphan()");
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  void XdsClient::WatchClusterData(
 | 
	
		
			
				|  |  | -    StringView cluster, std::unique_ptr<ClusterWatcherInterface> watcher) {
 | 
	
		
			
				|  |  | +    StringView cluster_name, std::unique_ptr<ClusterWatcherInterface> watcher) {
 | 
	
		
			
				|  |  | +  const bool new_name = cluster_map_.find(cluster_name) == cluster_map_.end();
 | 
	
		
			
				|  |  | +  ClusterState& cluster_state = cluster_map_[cluster_name];
 | 
	
		
			
				|  |  |    ClusterWatcherInterface* w = watcher.get();
 | 
	
		
			
				|  |  | -  cluster_state_.cluster_watchers[w] = std::move(watcher);
 | 
	
		
			
				|  |  | -  // TODO(juanlishen): Start CDS call if not already started and return
 | 
	
		
			
				|  |  | -  // real data via watcher.
 | 
	
		
			
				|  |  | -  CdsUpdate update;
 | 
	
		
			
				|  |  | -  update.eds_service_name = StringViewToCString(cluster);
 | 
	
		
			
				|  |  | -  update.lrs_load_reporting_server_name.reset(gpr_strdup(""));
 | 
	
		
			
				|  |  | -  w->OnClusterChanged(std::move(update));
 | 
	
		
			
				|  |  | +  cluster_state.watchers[w] = std::move(watcher);
 | 
	
		
			
				|  |  | +  // If we've already received an CDS update, notify the new watcher
 | 
	
		
			
				|  |  | +  // immediately.
 | 
	
		
			
				|  |  | +  if (cluster_state.update.has_value()) {
 | 
	
		
			
				|  |  | +    w->OnClusterChanged(cluster_state.update.value());
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  if (new_name) chand_->OnResourceNamesChanged(kCdsTypeUrl);
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -void XdsClient::CancelClusterDataWatch(StringView /*cluster*/,
 | 
	
		
			
				|  |  | +void XdsClient::CancelClusterDataWatch(StringView cluster_name,
 | 
	
		
			
				|  |  |                                         ClusterWatcherInterface* watcher) {
 | 
	
		
			
				|  |  | -  auto it = cluster_state_.cluster_watchers.find(watcher);
 | 
	
		
			
				|  |  | -  if (it != cluster_state_.cluster_watchers.end()) {
 | 
	
		
			
				|  |  | -    cluster_state_.cluster_watchers.erase(it);
 | 
	
		
			
				|  |  | -  }
 | 
	
		
			
				|  |  | -  if (chand_ != nullptr && cluster_state_.cluster_watchers.empty()) {
 | 
	
		
			
				|  |  | -    // TODO(juanlishen): Stop CDS call.
 | 
	
		
			
				|  |  | +  if (shutting_down_) return;
 | 
	
		
			
				|  |  | +  ClusterState& cluster_state = cluster_map_[cluster_name];
 | 
	
		
			
				|  |  | +  auto it = cluster_state.watchers.find(watcher);
 | 
	
		
			
				|  |  | +  if (it != cluster_state.watchers.end()) {
 | 
	
		
			
				|  |  | +    cluster_state.watchers.erase(it);
 | 
	
		
			
				|  |  | +    if (cluster_state.watchers.empty()) {
 | 
	
		
			
				|  |  | +      cluster_map_.erase(cluster_name);
 | 
	
		
			
				|  |  | +      chand_->OnResourceNamesChanged(kCdsTypeUrl);
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | +  chand_->OnWatcherRemoved();
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  void XdsClient::WatchEndpointData(
 | 
	
		
			
				|  |  | -    StringView /*cluster*/, std::unique_ptr<EndpointWatcherInterface> watcher) {
 | 
	
		
			
				|  |  | +    StringView eds_service_name,
 | 
	
		
			
				|  |  | +    std::unique_ptr<EndpointWatcherInterface> watcher) {
 | 
	
		
			
				|  |  | +  const bool new_name =
 | 
	
		
			
				|  |  | +      endpoint_map_.find(eds_service_name) == endpoint_map_.end();
 | 
	
		
			
				|  |  | +  EndpointState& endpoint_state = endpoint_map_[eds_service_name];
 | 
	
		
			
				|  |  |    EndpointWatcherInterface* w = watcher.get();
 | 
	
		
			
				|  |  | -  cluster_state_.endpoint_watchers[w] = std::move(watcher);
 | 
	
		
			
				|  |  | +  endpoint_state.watchers[w] = std::move(watcher);
 | 
	
		
			
				|  |  |    // If we've already received an EDS update, notify the new watcher
 | 
	
		
			
				|  |  |    // immediately.
 | 
	
		
			
				|  |  | -  if (!cluster_state_.eds_update.priority_list_update.empty()) {
 | 
	
		
			
				|  |  | -    w->OnEndpointChanged(cluster_state_.eds_update);
 | 
	
		
			
				|  |  | +  if (!endpoint_state.update.priority_list_update.empty()) {
 | 
	
		
			
				|  |  | +    w->OnEndpointChanged(endpoint_state.update);
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  chand_->MaybeStartAdsCall();
 | 
	
		
			
				|  |  | +  if (new_name) chand_->OnResourceNamesChanged(kEdsTypeUrl);
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -void XdsClient::CancelEndpointDataWatch(StringView /*cluster*/,
 | 
	
		
			
				|  |  | +void XdsClient::CancelEndpointDataWatch(StringView eds_service_name,
 | 
	
		
			
				|  |  |                                          EndpointWatcherInterface* watcher) {
 | 
	
		
			
				|  |  | -  auto it = cluster_state_.endpoint_watchers.find(watcher);
 | 
	
		
			
				|  |  | -  if (it != cluster_state_.endpoint_watchers.end()) {
 | 
	
		
			
				|  |  | -    cluster_state_.endpoint_watchers.erase(it);
 | 
	
		
			
				|  |  | -  }
 | 
	
		
			
				|  |  | -  if (chand_ != nullptr && cluster_state_.endpoint_watchers.empty()) {
 | 
	
		
			
				|  |  | -    chand_->StopAdsCall();
 | 
	
		
			
				|  |  | +  if (shutting_down_) return;
 | 
	
		
			
				|  |  | +  EndpointState& endpoint_state = endpoint_map_[eds_service_name];
 | 
	
		
			
				|  |  | +  auto it = endpoint_state.watchers.find(watcher);
 | 
	
		
			
				|  |  | +  if (it != endpoint_state.watchers.end()) {
 | 
	
		
			
				|  |  | +    endpoint_state.watchers.erase(it);
 | 
	
		
			
				|  |  | +    if (endpoint_state.watchers.empty()) {
 | 
	
		
			
				|  |  | +      endpoint_map_.erase(eds_service_name);
 | 
	
		
			
				|  |  | +      chand_->OnResourceNamesChanged(kEdsTypeUrl);
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | +  chand_->OnWatcherRemoved();
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  void XdsClient::AddClientStats(StringView /*lrs_server*/,
 | 
	
		
			
				|  |  | -                               StringView /*cluster*/,
 | 
	
		
			
				|  |  | +                               StringView cluster_name,
 | 
	
		
			
				|  |  |                                 XdsClientStats* client_stats) {
 | 
	
		
			
				|  |  | +  EndpointState& endpoint_state = endpoint_map_[cluster_name];
 | 
	
		
			
				|  |  |    // TODO(roth): When we add support for direct federation, use the
 | 
	
		
			
				|  |  |    // server name specified in lrs_server.
 | 
	
		
			
				|  |  | -  cluster_state_.client_stats.insert(client_stats);
 | 
	
		
			
				|  |  | +  endpoint_state.client_stats.insert(client_stats);
 | 
	
		
			
				|  |  |    chand_->MaybeStartLrsCall();
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  void XdsClient::RemoveClientStats(StringView /*lrs_server*/,
 | 
	
		
			
				|  |  | -                                  StringView /*cluster*/,
 | 
	
		
			
				|  |  | +                                  StringView cluster_name,
 | 
	
		
			
				|  |  |                                    XdsClientStats* client_stats) {
 | 
	
		
			
				|  |  | +  EndpointState& endpoint_state = endpoint_map_[cluster_name];
 | 
	
		
			
				|  |  |    // TODO(roth): When we add support for direct federation, use the
 | 
	
		
			
				|  |  |    // server name specified in lrs_server.
 | 
	
		
			
				|  |  |    // TODO(roth): In principle, we should try to send a final load report
 | 
	
		
			
				|  |  |    // containing whatever final stats have been accumulated since the
 | 
	
		
			
				|  |  |    // last load report.
 | 
	
		
			
				|  |  | -  auto it = cluster_state_.client_stats.find(client_stats);
 | 
	
		
			
				|  |  | -  if (it != cluster_state_.client_stats.end()) {
 | 
	
		
			
				|  |  | -    cluster_state_.client_stats.erase(it);
 | 
	
		
			
				|  |  | +  auto it = endpoint_state.client_stats.find(client_stats);
 | 
	
		
			
				|  |  | +  if (it != endpoint_state.client_stats.end()) {
 | 
	
		
			
				|  |  | +    endpoint_state.client_stats.erase(it);
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  if (chand_ != nullptr && cluster_state_.client_stats.empty()) {
 | 
	
		
			
				|  |  | +  if (chand_ != nullptr && endpoint_state.client_stats.empty()) {
 | 
	
		
			
				|  |  |      chand_->StopLrsCall();
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  |  }
 | 
	
	
		
			
				|  | @@ -1367,15 +1608,55 @@ void XdsClient::ResetBackoff() {
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  |  }
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | +std::set<StringView> XdsClient::WatchedClusterNames() const {
 | 
	
		
			
				|  |  | +  std::set<StringView> cluster_names;
 | 
	
		
			
				|  |  | +  for (const auto& p : cluster_map_) {
 | 
	
		
			
				|  |  | +    const StringView& cluster_name = p.first;
 | 
	
		
			
				|  |  | +    const ClusterState& cluster_state = p.second;
 | 
	
		
			
				|  |  | +    // Don't request for the clusters that are cached before watched.
 | 
	
		
			
				|  |  | +    if (cluster_state.watchers.empty()) continue;
 | 
	
		
			
				|  |  | +    cluster_names.emplace(cluster_name);
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  return cluster_names;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +std::set<StringView> XdsClient::EdsServiceNames() const {
 | 
	
		
			
				|  |  | +  std::set<StringView> eds_service_names;
 | 
	
		
			
				|  |  | +  for (const auto& p : endpoint_map_) {
 | 
	
		
			
				|  |  | +    const StringView& eds_service_name = p.first;
 | 
	
		
			
				|  |  | +    eds_service_names.emplace(eds_service_name);
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  return eds_service_names;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +std::map<StringView, std::set<XdsClientStats*>> XdsClient::ClientStatsMap()
 | 
	
		
			
				|  |  | +    const {
 | 
	
		
			
				|  |  | +  std::map<StringView, std::set<XdsClientStats*>> client_stats_map;
 | 
	
		
			
				|  |  | +  for (const auto& p : endpoint_map_) {
 | 
	
		
			
				|  |  | +    const StringView& cluster_name = p.first;
 | 
	
		
			
				|  |  | +    const auto& client_stats = p.second.client_stats;
 | 
	
		
			
				|  |  | +    if (chand_->lrs_calld()->ShouldSendLoadReports(cluster_name)) {
 | 
	
		
			
				|  |  | +      client_stats_map.emplace(cluster_name, client_stats);
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  | +  }
 | 
	
		
			
				|  |  | +  return client_stats_map;
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |  void XdsClient::NotifyOnError(grpc_error* error) {
 | 
	
		
			
				|  |  |    if (service_config_watcher_ != nullptr) {
 | 
	
		
			
				|  |  |      service_config_watcher_->OnError(GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  for (const auto& p : cluster_state_.cluster_watchers) {
 | 
	
		
			
				|  |  | -    p.first->OnError(GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +  for (const auto& p : cluster_map_) {
 | 
	
		
			
				|  |  | +    const ClusterState& cluster_state = p.second;
 | 
	
		
			
				|  |  | +    for (const auto& p : cluster_state.watchers) {
 | 
	
		
			
				|  |  | +      p.first->OnError(GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  | -  for (const auto& p : cluster_state_.endpoint_watchers) {
 | 
	
		
			
				|  |  | -    p.first->OnError(GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +  for (const auto& p : endpoint_map_) {
 | 
	
		
			
				|  |  | +    const EndpointState& endpoint_state = p.second;
 | 
	
		
			
				|  |  | +    for (const auto& p : endpoint_state.watchers) {
 | 
	
		
			
				|  |  | +      p.first->OnError(GRPC_ERROR_REF(error));
 | 
	
		
			
				|  |  | +    }
 | 
	
		
			
				|  |  |    }
 | 
	
		
			
				|  |  |    GRPC_ERROR_UNREF(error);
 | 
	
		
			
				|  |  |  }
 | 
	
	
		
			
				|  | @@ -1393,7 +1674,7 @@ void XdsClient::NotifyOnServiceConfig(void* arg, grpc_error* error) {
 | 
	
		
			
				|  |  |                 "    } }\n"
 | 
	
		
			
				|  |  |                 "  ]\n"
 | 
	
		
			
				|  |  |                 "}",
 | 
	
		
			
				|  |  | -               self->server_name_.get());
 | 
	
		
			
				|  |  | +               self->server_name_.c_str());
 | 
	
		
			
				|  |  |    RefCountedPtr<ServiceConfig> service_config =
 | 
	
		
			
				|  |  |        ServiceConfig::Create(json, &error);
 | 
	
		
			
				|  |  |    gpr_free(json);
 |