1 //
2 // Copyright 2018 gRPC authors.
3 //
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
7 //
8 //     http://www.apache.org/licenses/LICENSE-2.0
9 //
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
15 //
16 
17 #include <grpc/support/port_platform.h>
18 
19 #include <inttypes.h>
20 #include <stddef.h>
21 
22 #include <algorithm>
23 #include <map>
24 #include <memory>
25 #include <set>
26 #include <string>
27 #include <type_traits>
28 #include <utility>
29 #include <vector>
30 
31 #include "absl/status/status.h"
32 #include "absl/status/statusor.h"
33 #include "absl/strings/str_cat.h"
34 #include "absl/strings/str_join.h"
35 #include "absl/strings/string_view.h"
36 #include "absl/types/optional.h"
37 
38 #include <grpc/event_engine/event_engine.h>
39 #include <grpc/grpc.h>
40 #include <grpc/impl/connectivity_state.h>
41 #include <grpc/support/json.h>
42 #include <grpc/support/log.h>
43 
44 #include "src/core/ext/filters/client_channel/lb_policy/address_filtering.h"
45 #include "src/core/ext/filters/client_channel/lb_policy/child_policy_handler.h"
46 #include "src/core/ext/filters/client_channel/lb_policy/xds/xds_attributes.h"
47 #include "src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_args.h"
48 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
49 #include "src/core/ext/xds/xds_bootstrap.h"
50 #include "src/core/ext/xds/xds_bootstrap_grpc.h"
51 #include "src/core/ext/xds/xds_client.h"
52 #include "src/core/ext/xds/xds_client_grpc.h"
53 #include "src/core/ext/xds/xds_client_stats.h"
54 #include "src/core/ext/xds/xds_endpoint.h"
55 #include "src/core/lib/channel/channel_args.h"
56 #include "src/core/lib/config/core_configuration.h"
57 #include "src/core/lib/debug/trace.h"
58 #include "src/core/lib/gprpp/debug_location.h"
59 #include "src/core/lib/gprpp/orphanable.h"
60 #include "src/core/lib/gprpp/ref_counted_ptr.h"
61 #include "src/core/lib/gprpp/validation_errors.h"
62 #include "src/core/lib/gprpp/work_serializer.h"
63 #include "src/core/lib/iomgr/pollset_set.h"
64 #include "src/core/lib/json/json.h"
65 #include "src/core/lib/json/json_args.h"
66 #include "src/core/lib/json/json_object_loader.h"
67 #include "src/core/lib/json/json_writer.h"
68 #include "src/core/lib/load_balancing/lb_policy.h"
69 #include "src/core/lib/load_balancing/lb_policy_factory.h"
70 #include "src/core/lib/load_balancing/lb_policy_registry.h"
71 #include "src/core/lib/load_balancing/subchannel_interface.h"
72 #include "src/core/lib/resolver/resolver.h"
73 #include "src/core/lib/resolver/resolver_registry.h"
74 #include "src/core/lib/resolver/server_address.h"
75 #include "src/core/lib/transport/connectivity_state.h"
76 
77 #define GRPC_EDS_DEFAULT_FALLBACK_TIMEOUT 10000
78 
79 namespace grpc_core {
80 
81 TraceFlag grpc_lb_xds_cluster_resolver_trace(false, "xds_cluster_resolver_lb");
82 
83 namespace {
84 
85 constexpr absl::string_view kXdsClusterResolver =
86     "xds_cluster_resolver_experimental";
87 
88 // Config for EDS LB policy.
89 class XdsClusterResolverLbConfig : public LoadBalancingPolicy::Config {
90  public:
91   struct DiscoveryMechanism {
92     std::string cluster_name;
93     absl::optional<GrpcXdsBootstrap::GrpcXdsServer> lrs_load_reporting_server;
94     uint32_t max_concurrent_requests;
95     enum DiscoveryMechanismType {
96       EDS,
97       LOGICAL_DNS,
98     };
99     DiscoveryMechanismType type;
100     std::string eds_service_name;
101     std::string dns_hostname;
102 
103     Json::Array override_host_statuses;
104 
105     // This is type Json::Object instead of OutlierDetectionConfig, because we
106     // don't actually need to validate the contents of the outlier detection
107     // config here.  In this case, the JSON is generated by the CDS policy
108     // instead of coming from service config, so it's not actually any better
109     // to catch the problem here than it is to catch it in the
110     // outlier_detection policy itself, so here we just act as a pass-through.
111     absl::optional<Json::Object> outlier_detection_lb_config;
112 
operator ==grpc_core::__anon64dd73cf0111::XdsClusterResolverLbConfig::DiscoveryMechanism113     bool operator==(const DiscoveryMechanism& other) const {
114       return (cluster_name == other.cluster_name &&
115               lrs_load_reporting_server == other.lrs_load_reporting_server &&
116               max_concurrent_requests == other.max_concurrent_requests &&
117               type == other.type &&
118               eds_service_name == other.eds_service_name &&
119               dns_hostname == other.dns_hostname &&
120               override_host_statuses == other.override_host_statuses &&
121               outlier_detection_lb_config == other.outlier_detection_lb_config);
122     }
123 
124     static const JsonLoaderInterface* JsonLoader(const JsonArgs&);
125     void JsonPostLoad(const Json& json, const JsonArgs& args,
126                       ValidationErrors* errors);
127   };
128 
129   XdsClusterResolverLbConfig() = default;
130 
131   XdsClusterResolverLbConfig(const XdsClusterResolverLbConfig&) = delete;
132   XdsClusterResolverLbConfig& operator=(const XdsClusterResolverLbConfig&) =
133       delete;
134 
135   XdsClusterResolverLbConfig(XdsClusterResolverLbConfig&& other) = delete;
136   XdsClusterResolverLbConfig& operator=(XdsClusterResolverLbConfig&& other) =
137       delete;
138 
name() const139   absl::string_view name() const override { return kXdsClusterResolver; }
140 
discovery_mechanisms() const141   const std::vector<DiscoveryMechanism>& discovery_mechanisms() const {
142     return discovery_mechanisms_;
143   }
144 
xds_lb_policy() const145   const Json& xds_lb_policy() const { return xds_lb_policy_; }
146 
147   static const JsonLoaderInterface* JsonLoader(const JsonArgs&);
148   void JsonPostLoad(const Json& json, const JsonArgs& args,
149                     ValidationErrors* errors);
150 
151  private:
152   std::vector<DiscoveryMechanism> discovery_mechanisms_;
153   Json xds_lb_policy_;
154 };
155 
156 // Xds Cluster Resolver LB policy.
157 class XdsClusterResolverLb : public LoadBalancingPolicy {
158  public:
159   XdsClusterResolverLb(RefCountedPtr<XdsClient> xds_client, Args args);
160 
name() const161   absl::string_view name() const override { return kXdsClusterResolver; }
162 
163   absl::Status UpdateLocked(UpdateArgs args) override;
164   void ResetBackoffLocked() override;
165   void ExitIdleLocked() override;
166 
167  private:
168   // Discovery Mechanism Base class
169   //
170   // Implemented by EDS and LOGICAL_DNS.
171   //
172   // Implementations are responsible for calling the LB policy's
173   // OnEndpointChanged(), OnError(), and OnResourceDoesNotExist()
174   // methods when the corresponding events occur.
175   //
176   // Must implement Orphan() method to cancel the watchers.
177   class DiscoveryMechanism : public InternallyRefCounted<DiscoveryMechanism> {
178    public:
DiscoveryMechanism(RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,size_t index)179     DiscoveryMechanism(
180         RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,
181         size_t index)
182         : parent_(std::move(xds_cluster_resolver_lb)), index_(index) {}
183 
parent() const184     XdsClusterResolverLb* parent() const { return parent_.get(); }
index() const185     size_t index() const { return index_; }
186 
187     virtual void Start() = 0;
188     virtual Json::Array override_child_policy() = 0;
189     virtual bool disable_reresolution() = 0;
190 
191    private:
192     RefCountedPtr<XdsClusterResolverLb> parent_;
193     // Stores its own index in the vector of DiscoveryMechanism.
194     size_t index_;
195   };
196 
197   class EdsDiscoveryMechanism : public DiscoveryMechanism {
198    public:
EdsDiscoveryMechanism(RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,size_t index)199     EdsDiscoveryMechanism(
200         RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,
201         size_t index)
202         : DiscoveryMechanism(std::move(xds_cluster_resolver_lb), index) {}
203     void Start() override;
204     void Orphan() override;
override_child_policy()205     Json::Array override_child_policy() override { return Json::Array{}; }
disable_reresolution()206     bool disable_reresolution() override { return true; }
207 
208    private:
209     class EndpointWatcher : public XdsEndpointResourceType::WatcherInterface {
210      public:
EndpointWatcher(RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism)211       explicit EndpointWatcher(
212           RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism)
213           : discovery_mechanism_(std::move(discovery_mechanism)) {}
~EndpointWatcher()214       ~EndpointWatcher() override {
215         discovery_mechanism_.reset(DEBUG_LOCATION, "EndpointWatcher");
216       }
OnResourceChanged(XdsEndpointResource update)217       void OnResourceChanged(XdsEndpointResource update) override {
218         RefCountedPtr<EndpointWatcher> self = Ref();
219         discovery_mechanism_->parent()->work_serializer()->Run(
220             [self = std::move(self), update = std::move(update)]() mutable {
221               self->OnResourceChangedHelper(std::move(update));
222             },
223             DEBUG_LOCATION);
224       }
OnError(absl::Status status)225       void OnError(absl::Status status) override {
226         RefCountedPtr<EndpointWatcher> self = Ref();
227         discovery_mechanism_->parent()->work_serializer()->Run(
228             [self = std::move(self), status = std::move(status)]() mutable {
229               self->OnErrorHelper(std::move(status));
230             },
231             DEBUG_LOCATION);
232       }
OnResourceDoesNotExist()233       void OnResourceDoesNotExist() override {
234         RefCountedPtr<EndpointWatcher> self = Ref();
235         discovery_mechanism_->parent()->work_serializer()->Run(
236             [self = std::move(self)]() {
237               self->OnResourceDoesNotExistHelper();
238             },
239             DEBUG_LOCATION);
240       }
241 
242      private:
243       // Code accessing protected methods of `DiscoveryMechanism` need to be
244       // in methods of this class rather than in lambdas to work around an MSVC
245       // bug.
OnResourceChangedHelper(XdsEndpointResource update)246       void OnResourceChangedHelper(XdsEndpointResource update) {
247         std::string resolution_note;
248         if (update.priorities.empty()) {
249           resolution_note = absl::StrCat(
250               "EDS resource ", discovery_mechanism_->GetEdsResourceName(),
251               " contains no localities");
252         } else {
253           std::set<std::string> empty_localities;
254           for (const auto& priority : update.priorities) {
255             for (const auto& p : priority.localities) {
256               if (p.second.endpoints.empty()) {
257                 empty_localities.insert(p.first->AsHumanReadableString());
258               }
259             }
260           }
261           if (!empty_localities.empty()) {
262             resolution_note = absl::StrCat(
263                 "EDS resource ", discovery_mechanism_->GetEdsResourceName(),
264                 " contains empty localities: [",
265                 absl::StrJoin(empty_localities, "; "), "]");
266           }
267         }
268         discovery_mechanism_->parent()->OnEndpointChanged(
269             discovery_mechanism_->index(), std::move(update),
270             std::move(resolution_note));
271       }
OnErrorHelper(absl::Status status)272       void OnErrorHelper(absl::Status status) {
273         discovery_mechanism_->parent()->OnError(
274             discovery_mechanism_->index(),
275             absl::StrCat("EDS watcher error for resource ",
276                          discovery_mechanism_->GetEdsResourceName(), " (",
277                          status.ToString(), ")"));
278       }
OnResourceDoesNotExistHelper()279       void OnResourceDoesNotExistHelper() {
280         discovery_mechanism_->parent()->OnResourceDoesNotExist(
281             discovery_mechanism_->index(),
282             absl::StrCat("EDS resource ",
283                          discovery_mechanism_->GetEdsResourceName(),
284                          " does not exist"));
285       }
286       RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism_;
287     };
288 
289     // This is necessary only because of a bug in msvc where nested class
290     // cannot access protected member in base class.
291     friend class EndpointWatcher;
292 
GetEdsResourceName() const293     absl::string_view GetEdsResourceName() const {
294       auto& config = parent()->config_->discovery_mechanisms()[index()];
295       if (!config.eds_service_name.empty()) return config.eds_service_name;
296       return config.cluster_name;
297     }
298 
299     // Note that this is not owned, so this pointer must never be dereferenced.
300     EndpointWatcher* watcher_ = nullptr;
301   };
302 
303   class LogicalDNSDiscoveryMechanism : public DiscoveryMechanism {
304    public:
LogicalDNSDiscoveryMechanism(RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,size_t index)305     LogicalDNSDiscoveryMechanism(
306         RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_lb,
307         size_t index)
308         : DiscoveryMechanism(std::move(xds_cluster_resolver_lb), index) {}
309     void Start() override;
310     void Orphan() override;
override_child_policy()311     Json::Array override_child_policy() override {
312       return {
313           Json::FromObject({
314               {"pick_first", Json::FromObject({})},
315           }),
316       };
317     }
disable_reresolution()318     bool disable_reresolution() override { return false; };
319 
320    private:
321     class ResolverResultHandler : public Resolver::ResultHandler {
322      public:
ResolverResultHandler(RefCountedPtr<LogicalDNSDiscoveryMechanism> discovery_mechanism)323       explicit ResolverResultHandler(
324           RefCountedPtr<LogicalDNSDiscoveryMechanism> discovery_mechanism)
325           : discovery_mechanism_(std::move(discovery_mechanism)) {}
326 
~ResolverResultHandler()327       ~ResolverResultHandler() override {}
328 
329       void ReportResult(Resolver::Result result) override;
330 
331      private:
332       RefCountedPtr<LogicalDNSDiscoveryMechanism> discovery_mechanism_;
333     };
334 
335     // This is necessary only because of a bug in msvc where nested class cannot
336     // access protected member in base class.
337     friend class ResolverResultHandler;
338 
GetDnsHostname() const339     absl::string_view GetDnsHostname() const {
340       auto& config = parent()->config_->discovery_mechanisms()[index()];
341       return config.dns_hostname;
342     }
343 
344     OrphanablePtr<Resolver> resolver_;
345   };
346 
347   struct DiscoveryMechanismEntry {
348     OrphanablePtr<DiscoveryMechanism> discovery_mechanism;
349     // Most recent update reported by the discovery mechanism.
350     absl::optional<XdsEndpointResource> latest_update;
351     // Last resolution note reported by the discovery mechanism, if any.
352     std::string resolution_note;
353     // State used to retain child policy names for priority policy.
354     std::vector<size_t /*child_number*/> priority_child_numbers;
355     size_t next_available_child_number = 0;
356 
357     const XdsClusterResolverLbConfig::DiscoveryMechanism& config() const;
358 
359     // Returns the child policy name for a given priority.
360     std::string GetChildPolicyName(size_t priority) const;
361   };
362 
363   class Helper : public ChannelControlHelper {
364    public:
Helper(RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_policy)365     explicit Helper(
366         RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_policy)
367         : xds_cluster_resolver_policy_(std::move(xds_cluster_resolver_policy)) {
368     }
369 
~Helper()370     ~Helper() override {
371       xds_cluster_resolver_policy_.reset(DEBUG_LOCATION, "Helper");
372     }
373 
374     RefCountedPtr<SubchannelInterface> CreateSubchannel(
375         ServerAddress address, const ChannelArgs& args) override;
376     void UpdateState(grpc_connectivity_state state, const absl::Status& status,
377                      RefCountedPtr<SubchannelPicker> picker) override;
378     // This is a no-op, because we get the addresses from the xds
379     // client, which is a watch-based API.
RequestReresolution()380     void RequestReresolution() override {}
381     absl::string_view GetAuthority() override;
382     grpc_event_engine::experimental::EventEngine* GetEventEngine() override;
383     void AddTraceEvent(TraceSeverity severity,
384                        absl::string_view message) override;
385 
386    private:
387     RefCountedPtr<XdsClusterResolverLb> xds_cluster_resolver_policy_;
388   };
389 
390   ~XdsClusterResolverLb() override;
391 
392   void ShutdownLocked() override;
393 
394   void OnEndpointChanged(size_t index, XdsEndpointResource update,
395                          std::string resolution_note);
396   void OnError(size_t index, std::string resolution_note);
397   void OnResourceDoesNotExist(size_t index, std::string resolution_note);
398 
399   void MaybeDestroyChildPolicyLocked();
400 
401   absl::Status UpdateChildPolicyLocked();
402   OrphanablePtr<LoadBalancingPolicy> CreateChildPolicyLocked(
403       const ChannelArgs& args);
404   ServerAddressList CreateChildPolicyAddressesLocked();
405   std::string CreateChildPolicyResolutionNoteLocked();
406   RefCountedPtr<Config> CreateChildPolicyConfigLocked();
407   ChannelArgs CreateChildPolicyArgsLocked(const ChannelArgs& args_in);
408 
409   // The xds client and endpoint watcher.
410   RefCountedPtr<XdsClient> xds_client_;
411 
412   // Current channel args and config from the resolver.
413   ChannelArgs args_;
414   RefCountedPtr<XdsClusterResolverLbConfig> config_;
415 
416   // Internal state.
417   bool shutting_down_ = false;
418 
419   // Vector of discovery mechansism entries in priority order.
420   std::vector<DiscoveryMechanismEntry> discovery_mechanisms_;
421 
422   OrphanablePtr<LoadBalancingPolicy> child_policy_;
423 };
424 
425 //
426 // XdsClusterResolverLb::Helper
427 //
428 
429 RefCountedPtr<SubchannelInterface>
CreateSubchannel(ServerAddress address,const ChannelArgs & args)430 XdsClusterResolverLb::Helper::CreateSubchannel(ServerAddress address,
431                                                const ChannelArgs& args) {
432   if (xds_cluster_resolver_policy_->shutting_down_) return nullptr;
433   return xds_cluster_resolver_policy_->channel_control_helper()
434       ->CreateSubchannel(std::move(address), args);
435 }
436 
UpdateState(grpc_connectivity_state state,const absl::Status & status,RefCountedPtr<SubchannelPicker> picker)437 void XdsClusterResolverLb::Helper::UpdateState(
438     grpc_connectivity_state state, const absl::Status& status,
439     RefCountedPtr<SubchannelPicker> picker) {
440   if (xds_cluster_resolver_policy_->shutting_down_ ||
441       xds_cluster_resolver_policy_->child_policy_ == nullptr) {
442     return;
443   }
444   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
445     gpr_log(GPR_INFO,
446             "[xds_cluster_resolver_lb %p] child policy updated state=%s (%s) "
447             "picker=%p",
448             xds_cluster_resolver_policy_.get(), ConnectivityStateName(state),
449             status.ToString().c_str(), picker.get());
450   }
451   xds_cluster_resolver_policy_->channel_control_helper()->UpdateState(
452       state, status, std::move(picker));
453 }
454 
GetAuthority()455 absl::string_view XdsClusterResolverLb::Helper::GetAuthority() {
456   return xds_cluster_resolver_policy_->channel_control_helper()->GetAuthority();
457 }
458 
459 grpc_event_engine::experimental::EventEngine*
GetEventEngine()460 XdsClusterResolverLb::Helper::GetEventEngine() {
461   return xds_cluster_resolver_policy_->channel_control_helper()
462       ->GetEventEngine();
463 }
464 
AddTraceEvent(TraceSeverity severity,absl::string_view message)465 void XdsClusterResolverLb::Helper::AddTraceEvent(TraceSeverity severity,
466                                                  absl::string_view message) {
467   if (xds_cluster_resolver_policy_->shutting_down_) return;
468   xds_cluster_resolver_policy_->channel_control_helper()->AddTraceEvent(
469       severity, message);
470 }
471 
472 //
473 // XdsClusterResolverLb::EdsDiscoveryMechanism
474 //
475 
Start()476 void XdsClusterResolverLb::EdsDiscoveryMechanism::Start() {
477   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
478     gpr_log(GPR_INFO,
479             "[xds_cluster_resolver_lb %p] eds discovery mechanism %" PRIuPTR
480             ":%p starting xds watch for %s",
481             parent(), index(), this, std::string(GetEdsResourceName()).c_str());
482   }
483   auto watcher = MakeRefCounted<EndpointWatcher>(
484       Ref(DEBUG_LOCATION, "EdsDiscoveryMechanism"));
485   watcher_ = watcher.get();
486   XdsEndpointResourceType::StartWatch(parent()->xds_client_.get(),
487                                       GetEdsResourceName(), std::move(watcher));
488 }
489 
Orphan()490 void XdsClusterResolverLb::EdsDiscoveryMechanism::Orphan() {
491   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
492     gpr_log(GPR_INFO,
493             "[xds_cluster_resolver_lb %p] eds discovery mechanism %" PRIuPTR
494             ":%p cancelling xds watch for %s",
495             parent(), index(), this, std::string(GetEdsResourceName()).c_str());
496   }
497   XdsEndpointResourceType::CancelWatch(parent()->xds_client_.get(),
498                                        GetEdsResourceName(), watcher_);
499   Unref();
500 }
501 
502 //
503 // XdsClusterResolverLb::LogicalDNSDiscoveryMechanism
504 //
505 
Start()506 void XdsClusterResolverLb::LogicalDNSDiscoveryMechanism::Start() {
507   std::string target;
508   ChannelArgs args = parent()->args_;
509   auto* fake_resolver_response_generator =
510       args.GetPointer<FakeResolverResponseGenerator>(
511           GRPC_ARG_XDS_LOGICAL_DNS_CLUSTER_FAKE_RESOLVER_RESPONSE_GENERATOR);
512   if (fake_resolver_response_generator != nullptr) {
513     target = absl::StrCat("fake:", GetDnsHostname());
514     args = args.SetObject(fake_resolver_response_generator->Ref());
515   } else {
516     target = absl::StrCat("dns:", GetDnsHostname());
517   }
518   resolver_ = CoreConfiguration::Get().resolver_registry().CreateResolver(
519       target.c_str(), args, parent()->interested_parties(),
520       parent()->work_serializer(),
521       std::make_unique<ResolverResultHandler>(
522           Ref(DEBUG_LOCATION, "LogicalDNSDiscoveryMechanism")));
523   if (resolver_ == nullptr) {
524     parent()->OnResourceDoesNotExist(
525         index(),
526         absl::StrCat("error creating DNS resolver for ", GetDnsHostname()));
527     return;
528   }
529   resolver_->StartLocked();
530   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
531     gpr_log(GPR_INFO,
532             "[xds_cluster_resolver_lb %p] logical DNS discovery mechanism "
533             "%" PRIuPTR ":%p starting dns resolver %p",
534             parent(), index(), this, resolver_.get());
535   }
536 }
537 
Orphan()538 void XdsClusterResolverLb::LogicalDNSDiscoveryMechanism::Orphan() {
539   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
540     gpr_log(
541         GPR_INFO,
542         "[xds_cluster_resolver_lb %p] logical DNS discovery mechanism %" PRIuPTR
543         ":%p shutting down dns resolver %p",
544         parent(), index(), this, resolver_.get());
545   }
546   resolver_.reset();
547   Unref();
548 }
549 
550 //
551 // XdsClusterResolverLb::LogicalDNSDiscoveryMechanism::ResolverResultHandler
552 //
553 
554 void XdsClusterResolverLb::LogicalDNSDiscoveryMechanism::ResolverResultHandler::
ReportResult(Resolver::Result result)555     ReportResult(Resolver::Result result) {
556   XdsClusterResolverLb* lb_policy = discovery_mechanism_->parent();
557   size_t index = discovery_mechanism_->index();
558   if (!result.addresses.ok()) {
559     if (result.resolution_note.empty()) {
560       result.resolution_note = absl::StrCat(
561           "DNS resolution failed for ", discovery_mechanism_->GetDnsHostname(),
562           " (", result.addresses.status().ToString(), ")");
563     }
564     lb_policy->OnError(index, result.resolution_note);
565     return;
566   }
567   // Convert resolver result to EDS update.
568   XdsEndpointResource update;
569   XdsEndpointResource::Priority::Locality locality;
570   locality.name = MakeRefCounted<XdsLocalityName>("", "", "");
571   locality.lb_weight = 1;
572   locality.endpoints = std::move(*result.addresses);
573   XdsEndpointResource::Priority priority;
574   priority.localities.emplace(locality.name.get(), std::move(locality));
575   update.priorities.emplace_back(std::move(priority));
576   lb_policy->OnEndpointChanged(index, std::move(update),
577                                std::move(result.resolution_note));
578 }
579 
580 //
581 // XdsClusterResolverLb::DiscoveryMechanismEntry
582 //
583 
584 const XdsClusterResolverLbConfig::DiscoveryMechanism&
config() const585 XdsClusterResolverLb::DiscoveryMechanismEntry::config() const {
586   return discovery_mechanism->parent()
587       ->config_->discovery_mechanisms()[discovery_mechanism->index()];
588 }
589 
GetChildPolicyName(size_t priority) const590 std::string XdsClusterResolverLb::DiscoveryMechanismEntry::GetChildPolicyName(
591     size_t priority) const {
592   return absl::StrCat("{cluster=", config().cluster_name,
593                       ", child_number=", priority_child_numbers[priority], "}");
594 }
595 
596 //
597 // XdsClusterResolverLb public methods
598 //
599 
XdsClusterResolverLb(RefCountedPtr<XdsClient> xds_client,Args args)600 XdsClusterResolverLb::XdsClusterResolverLb(RefCountedPtr<XdsClient> xds_client,
601                                            Args args)
602     : LoadBalancingPolicy(std::move(args)), xds_client_(std::move(xds_client)) {
603   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
604     gpr_log(GPR_INFO, "[xds_cluster_resolver_lb %p] created -- xds_client=%p",
605             this, xds_client_.get());
606   }
607 }
608 
~XdsClusterResolverLb()609 XdsClusterResolverLb::~XdsClusterResolverLb() {
610   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
611     gpr_log(GPR_INFO,
612             "[xds_cluster_resolver_lb %p] destroying xds_cluster_resolver LB "
613             "policy",
614             this);
615   }
616 }
617 
ShutdownLocked()618 void XdsClusterResolverLb::ShutdownLocked() {
619   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
620     gpr_log(GPR_INFO, "[xds_cluster_resolver_lb %p] shutting down", this);
621   }
622   shutting_down_ = true;
623   MaybeDestroyChildPolicyLocked();
624   discovery_mechanisms_.clear();
625   xds_client_.reset(DEBUG_LOCATION, "XdsClusterResolverLb");
626   args_ = ChannelArgs();
627 }
628 
MaybeDestroyChildPolicyLocked()629 void XdsClusterResolverLb::MaybeDestroyChildPolicyLocked() {
630   if (child_policy_ != nullptr) {
631     grpc_pollset_set_del_pollset_set(child_policy_->interested_parties(),
632                                      interested_parties());
633     child_policy_.reset();
634   }
635 }
636 
UpdateLocked(UpdateArgs args)637 absl::Status XdsClusterResolverLb::UpdateLocked(UpdateArgs args) {
638   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
639     gpr_log(GPR_INFO, "[xds_cluster_resolver_lb %p] Received update", this);
640   }
641   const bool is_initial_update = args_ == ChannelArgs();
642   // Update config.
643   auto old_config = std::move(config_);
644   config_ = std::move(args.config);
645   // Update args.
646   args_ = std::move(args.args);
647   // Update child policy if needed.
648   absl::Status status;
649   if (child_policy_ != nullptr) status = UpdateChildPolicyLocked();
650   // Create endpoint watcher if needed.
651   if (is_initial_update) {
652     for (const auto& config : config_->discovery_mechanisms()) {
653       DiscoveryMechanismEntry entry;
654       if (config.type == XdsClusterResolverLbConfig::DiscoveryMechanism::
655                              DiscoveryMechanismType::EDS) {
656         entry.discovery_mechanism = MakeOrphanable<EdsDiscoveryMechanism>(
657             Ref(DEBUG_LOCATION, "EdsDiscoveryMechanism"),
658             discovery_mechanisms_.size());
659       } else if (config.type == XdsClusterResolverLbConfig::DiscoveryMechanism::
660                                     DiscoveryMechanismType::LOGICAL_DNS) {
661         entry.discovery_mechanism =
662             MakeOrphanable<LogicalDNSDiscoveryMechanism>(
663                 Ref(DEBUG_LOCATION, "LogicalDNSDiscoveryMechanism"),
664                 discovery_mechanisms_.size());
665       } else {
666         GPR_ASSERT(0);
667       }
668       discovery_mechanisms_.push_back(std::move(entry));
669     }
670     // Call start() on all discovery mechanisms after creation.
671     for (const auto& discovery_mechanism : discovery_mechanisms_) {
672       discovery_mechanism.discovery_mechanism->Start();
673     }
674   }
675   return status;
676 }
677 
ResetBackoffLocked()678 void XdsClusterResolverLb::ResetBackoffLocked() {
679   if (child_policy_ != nullptr) {
680     child_policy_->ResetBackoffLocked();
681   }
682 }
683 
ExitIdleLocked()684 void XdsClusterResolverLb::ExitIdleLocked() {
685   if (child_policy_ != nullptr) child_policy_->ExitIdleLocked();
686 }
687 
OnEndpointChanged(size_t index,XdsEndpointResource update,std::string resolution_note)688 void XdsClusterResolverLb::OnEndpointChanged(size_t index,
689                                              XdsEndpointResource update,
690                                              std::string resolution_note) {
691   if (shutting_down_) return;
692   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
693     gpr_log(GPR_INFO,
694             "[xds_cluster_resolver_lb %p] Received update from xds client"
695             " for discovery mechanism %" PRIuPTR " (resolution_note=\"%s\")",
696             this, index, resolution_note.c_str());
697   }
698   DiscoveryMechanismEntry& discovery_entry = discovery_mechanisms_[index];
699   // We need at least one priority for each discovery mechanism, just so that we
700   // have a child in which to create the xds_cluster_impl policy.  This ensures
701   // that we properly handle the case of a discovery mechanism dropping 100% of
702   // calls, the OnError() case, and the OnResourceDoesNotExist() case.
703   if (update.priorities.empty()) update.priorities.emplace_back();
704   // Update priority_child_numbers, reusing old child numbers in an
705   // intelligent way to avoid unnecessary churn.
706   // First, build some maps from locality to child number and the reverse
707   // from the old data in the entry's update and priority_child_numbers.
708   std::map<XdsLocalityName*, size_t /*child_number*/, XdsLocalityName::Less>
709       locality_child_map;
710   std::map<size_t, std::set<XdsLocalityName*, XdsLocalityName::Less>>
711       child_locality_map;
712   if (discovery_entry.latest_update.has_value()) {
713     const auto& prev_priority_list = discovery_entry.latest_update->priorities;
714     for (size_t priority = 0; priority < prev_priority_list.size();
715          ++priority) {
716       size_t child_number = discovery_entry.priority_child_numbers[priority];
717       const auto& localities = prev_priority_list[priority].localities;
718       for (const auto& p : localities) {
719         XdsLocalityName* locality_name = p.first;
720         locality_child_map[locality_name] = child_number;
721         child_locality_map[child_number].insert(locality_name);
722       }
723     }
724   }
725   // Construct new list of children.
726   std::vector<size_t> priority_child_numbers;
727   for (size_t priority = 0; priority < update.priorities.size(); ++priority) {
728     const auto& localities = update.priorities[priority].localities;
729     absl::optional<size_t> child_number;
730     // If one of the localities in this priority already existed, reuse its
731     // child number.
732     for (const auto& p : localities) {
733       XdsLocalityName* locality_name = p.first;
734       if (!child_number.has_value()) {
735         auto it = locality_child_map.find(locality_name);
736         if (it != locality_child_map.end()) {
737           child_number = it->second;
738           locality_child_map.erase(it);
739           // Remove localities that *used* to be in this child number, so
740           // that we don't incorrectly reuse this child number for a
741           // subsequent priority.
742           for (XdsLocalityName* old_locality :
743                child_locality_map[*child_number]) {
744             locality_child_map.erase(old_locality);
745           }
746         }
747       } else {
748         // Remove all localities that are now in this child number, so
749         // that we don't accidentally reuse this child number for a
750         // subsequent priority.
751         locality_child_map.erase(locality_name);
752       }
753     }
754     // If we didn't find an existing child number, assign a new one.
755     if (!child_number.has_value()) {
756       for (child_number = discovery_entry.next_available_child_number;
757            child_locality_map.find(*child_number) != child_locality_map.end();
758            ++(*child_number)) {
759       }
760       discovery_entry.next_available_child_number = *child_number + 1;
761       // Add entry so we know that the child number is in use.
762       // (Don't need to add the list of localities, since we won't use them.)
763       child_locality_map[*child_number];
764     }
765     priority_child_numbers.push_back(*child_number);
766   }
767   // Save update.
768   discovery_entry.latest_update = std::move(update);
769   discovery_entry.resolution_note = std::move(resolution_note);
770   discovery_entry.priority_child_numbers = std::move(priority_child_numbers);
771   // If any discovery mechanism has not received its first update,
772   // wait until that happens before creating the child policy.
773   // TODO(roth): If this becomes problematic in the future (e.g., a
774   // secondary discovery mechanism delaying us from starting up at all),
775   // we can consider some sort of optimization whereby we can create the
776   // priority policy with only a subset of its children.  But we need to
777   // make sure not to get into a situation where the priority policy
778   // will put the channel into TRANSIENT_FAILURE instead of CONNECTING
779   // while we're still waiting for the other discovery mechanism(s).
780   for (DiscoveryMechanismEntry& mechanism : discovery_mechanisms_) {
781     if (!mechanism.latest_update.has_value()) return;
782   }
783   // Update child policy.
784   // TODO(roth): If the child policy reports an error with the update,
785   // we need to propagate that error back to the resolver somehow.
786   (void)UpdateChildPolicyLocked();
787 }
788 
OnError(size_t index,std::string resolution_note)789 void XdsClusterResolverLb::OnError(size_t index, std::string resolution_note) {
790   gpr_log(GPR_ERROR,
791           "[xds_cluster_resolver_lb %p] discovery mechanism %" PRIuPTR
792           " reported error: %s",
793           this, index, resolution_note.c_str());
794   if (shutting_down_) return;
795   if (!discovery_mechanisms_[index].latest_update.has_value()) {
796     // Call OnEndpointChanged() with an empty update just like
797     // OnResourceDoesNotExist().
798     OnEndpointChanged(index, XdsEndpointResource(), std::move(resolution_note));
799   }
800 }
801 
OnResourceDoesNotExist(size_t index,std::string resolution_note)802 void XdsClusterResolverLb::OnResourceDoesNotExist(size_t index,
803                                                   std::string resolution_note) {
804   gpr_log(GPR_ERROR,
805           "[xds_cluster_resolver_lb %p] discovery mechanism %" PRIuPTR
806           " resource does not exist: %s",
807           this, index, resolution_note.c_str());
808   if (shutting_down_) return;
809   // Call OnEndpointChanged() with an empty update.
810   OnEndpointChanged(index, XdsEndpointResource(), std::move(resolution_note));
811 }
812 
813 //
814 // child policy-related methods
815 //
816 
CreateChildPolicyAddressesLocked()817 ServerAddressList XdsClusterResolverLb::CreateChildPolicyAddressesLocked() {
818   ServerAddressList addresses;
819   for (const auto& discovery_entry : discovery_mechanisms_) {
820     for (size_t priority = 0;
821          priority < discovery_entry.latest_update->priorities.size();
822          ++priority) {
823       const auto& priority_entry =
824           discovery_entry.latest_update->priorities[priority];
825       std::string priority_child_name =
826           discovery_entry.GetChildPolicyName(priority);
827       for (const auto& p : priority_entry.localities) {
828         const auto& locality_name = p.first;
829         const auto& locality = p.second;
830         std::vector<std::string> hierarchical_path = {
831             priority_child_name, locality_name->AsHumanReadableString()};
832         for (const auto& endpoint : locality.endpoints) {
833           const ServerAddressWeightAttribute* weight_attribute = static_cast<
834               const ServerAddressWeightAttribute*>(endpoint.GetAttribute(
835               ServerAddressWeightAttribute::kServerAddressWeightAttributeKey));
836           uint32_t weight = locality.lb_weight;
837           if (weight_attribute != nullptr) {
838             weight = locality.lb_weight * weight_attribute->weight();
839           }
840           addresses.emplace_back(
841               endpoint
842                   .WithAttribute(
843                       kHierarchicalPathAttributeKey,
844                       MakeHierarchicalPathAttribute(hierarchical_path))
845                   .WithAttribute(kXdsLocalityNameAttributeKey,
846                                  std::make_unique<XdsLocalityAttribute>(
847                                      locality_name->Ref(), locality.lb_weight))
848                   .WithAttribute(
849                       ServerAddressWeightAttribute::
850                           kServerAddressWeightAttributeKey,
851                       std::make_unique<ServerAddressWeightAttribute>(weight)));
852         }
853       }
854     }
855   }
856   return addresses;
857 }
858 
CreateChildPolicyResolutionNoteLocked()859 std::string XdsClusterResolverLb::CreateChildPolicyResolutionNoteLocked() {
860   std::vector<absl::string_view> resolution_notes;
861   for (const auto& discovery_entry : discovery_mechanisms_) {
862     if (!discovery_entry.resolution_note.empty()) {
863       resolution_notes.push_back(discovery_entry.resolution_note);
864     }
865   }
866   return absl::StrJoin(resolution_notes, "; ");
867 }
868 
869 RefCountedPtr<LoadBalancingPolicy::Config>
CreateChildPolicyConfigLocked()870 XdsClusterResolverLb::CreateChildPolicyConfigLocked() {
871   Json::Object priority_children;
872   Json::Array priority_priorities;
873   for (const auto& discovery_entry : discovery_mechanisms_) {
874     const auto& discovery_config = discovery_entry.config();
875     for (size_t priority = 0;
876          priority < discovery_entry.latest_update->priorities.size();
877          ++priority) {
878       // Determine what xDS LB policy to use.
879       Json child_policy;
880       if (!discovery_entry.discovery_mechanism->override_child_policy()
881                .empty()) {
882         child_policy = Json::FromArray(
883             discovery_entry.discovery_mechanism->override_child_policy());
884       } else {
885         child_policy = config_->xds_lb_policy();
886       }
887       // Wrap the xDS LB policy in the xds_override_host policy.
888       Json::Object xds_override_host_lb_config = {
889           {"childPolicy", std::move(child_policy)},
890       };
891       if (!discovery_config.override_host_statuses.empty()) {
892         xds_override_host_lb_config["overrideHostStatus"] =
893             Json::FromArray(discovery_config.override_host_statuses);
894       }
895       Json::Array xds_override_host_config = {Json::FromObject({
896           {"xds_override_host_experimental",
897            Json::FromObject(std::move(xds_override_host_lb_config))},
898       })};
899       // Wrap it in the xds_cluster_impl policy.
900       Json::Array drop_categories;
901       if (discovery_entry.latest_update->drop_config != nullptr) {
902         for (const auto& category :
903              discovery_entry.latest_update->drop_config->drop_category_list()) {
904           drop_categories.push_back(Json::FromObject({
905               {"category", Json::FromString(category.name)},
906               {"requests_per_million",
907                Json::FromNumber(category.parts_per_million)},
908           }));
909         }
910       }
911       Json::Object xds_cluster_impl_config = {
912           {"clusterName", Json::FromString(discovery_config.cluster_name)},
913           {"childPolicy", Json::FromArray(std::move(xds_override_host_config))},
914           {"dropCategories", Json::FromArray(std::move(drop_categories))},
915           {"maxConcurrentRequests",
916            Json::FromNumber(discovery_config.max_concurrent_requests)},
917       };
918       if (!discovery_config.eds_service_name.empty()) {
919         xds_cluster_impl_config["edsServiceName"] =
920             Json::FromString(discovery_config.eds_service_name);
921       }
922       if (discovery_config.lrs_load_reporting_server.has_value()) {
923         xds_cluster_impl_config["lrsLoadReportingServer"] =
924             discovery_config.lrs_load_reporting_server->ToJson();
925       }
926       // Wrap it in the outlier_detection policy.
927       Json::Object outlier_detection_config;
928       if (discovery_entry.config().outlier_detection_lb_config.has_value()) {
929         outlier_detection_config =
930             discovery_entry.config().outlier_detection_lb_config.value();
931       }
932       outlier_detection_config["childPolicy"] =
933           Json::FromArray({Json::FromObject({
934               {"xds_cluster_impl_experimental",
935                Json::FromObject(std::move(xds_cluster_impl_config))},
936           })});
937       Json locality_picking_policy = Json::FromArray({Json::FromObject({
938           {"outlier_detection_experimental",
939            Json::FromObject(std::move(outlier_detection_config))},
940       })});
941       // Add priority entry, with the appropriate child name.
942       std::string child_name = discovery_entry.GetChildPolicyName(priority);
943       priority_priorities.emplace_back(Json::FromString(child_name));
944       Json::Object child_config = {
945           {"config", std::move(locality_picking_policy)},
946       };
947       if (discovery_entry.discovery_mechanism->disable_reresolution()) {
948         child_config["ignore_reresolution_requests"] = Json::FromBool(true);
949       }
950       priority_children[child_name] = Json::FromObject(std::move(child_config));
951     }
952   }
953   Json json = Json::FromArray({Json::FromObject({
954       {"priority_experimental",
955        Json::FromObject({
956            {"children", Json::FromObject(std::move(priority_children))},
957            {"priorities", Json::FromArray(std::move(priority_priorities))},
958        })},
959   })});
960   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
961     gpr_log(
962         GPR_INFO,
963         "[xds_cluster_resolver_lb %p] generated config for child policy: %s",
964         this, JsonDump(json, /*indent=*/1).c_str());
965   }
966   auto config =
967       CoreConfiguration::Get().lb_policy_registry().ParseLoadBalancingConfig(
968           json);
969   if (!config.ok()) {
970     // This should never happen, but if it does, we basically have no
971     // way to fix it, so we put the channel in TRANSIENT_FAILURE.
972     gpr_log(GPR_ERROR,
973             "[xds_cluster_resolver_lb %p] error parsing generated child policy "
974             "config -- "
975             "will put channel in TRANSIENT_FAILURE: %s",
976             this, config.status().ToString().c_str());
977     absl::Status status = absl::InternalError(
978         "xds_cluster_resolver LB policy: error parsing generated child policy "
979         "config");
980     channel_control_helper()->UpdateState(
981         GRPC_CHANNEL_TRANSIENT_FAILURE, status,
982         MakeRefCounted<TransientFailurePicker>(status));
983     return nullptr;
984   }
985   return std::move(*config);
986 }
987 
UpdateChildPolicyLocked()988 absl::Status XdsClusterResolverLb::UpdateChildPolicyLocked() {
989   if (shutting_down_) return absl::OkStatus();
990   UpdateArgs update_args;
991   update_args.config = CreateChildPolicyConfigLocked();
992   if (update_args.config == nullptr) return absl::OkStatus();
993   update_args.addresses = CreateChildPolicyAddressesLocked();
994   update_args.resolution_note = CreateChildPolicyResolutionNoteLocked();
995   update_args.args = CreateChildPolicyArgsLocked(args_);
996   if (child_policy_ == nullptr) {
997     child_policy_ = CreateChildPolicyLocked(update_args.args);
998   }
999   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
1000     gpr_log(GPR_INFO, "[xds_cluster_resolver_lb %p] Updating child policy %p",
1001             this, child_policy_.get());
1002   }
1003   return child_policy_->UpdateLocked(std::move(update_args));
1004 }
1005 
CreateChildPolicyArgsLocked(const ChannelArgs & args)1006 ChannelArgs XdsClusterResolverLb::CreateChildPolicyArgsLocked(
1007     const ChannelArgs& args) {
1008   // Inhibit client-side health checking, since the balancer does this
1009   // for us.
1010   return args.Set(GRPC_ARG_INHIBIT_HEALTH_CHECKING, 1);
1011 }
1012 
1013 OrphanablePtr<LoadBalancingPolicy>
CreateChildPolicyLocked(const ChannelArgs & args)1014 XdsClusterResolverLb::CreateChildPolicyLocked(const ChannelArgs& args) {
1015   LoadBalancingPolicy::Args lb_policy_args;
1016   lb_policy_args.work_serializer = work_serializer();
1017   lb_policy_args.args = args;
1018   lb_policy_args.channel_control_helper =
1019       std::make_unique<Helper>(Ref(DEBUG_LOCATION, "Helper"));
1020   OrphanablePtr<LoadBalancingPolicy> lb_policy =
1021       CoreConfiguration::Get().lb_policy_registry().CreateLoadBalancingPolicy(
1022           "priority_experimental", std::move(lb_policy_args));
1023   if (GPR_UNLIKELY(lb_policy == nullptr)) {
1024     gpr_log(GPR_ERROR,
1025             "[xds_cluster_resolver_lb %p] failure creating child policy", this);
1026     return nullptr;
1027   }
1028   if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_xds_cluster_resolver_trace)) {
1029     gpr_log(GPR_INFO,
1030             "[xds_cluster_resolver_lb %p]: Created new child policy %p", this,
1031             lb_policy.get());
1032   }
1033   // Add our interested_parties pollset_set to that of the newly created
1034   // child policy. This will make the child policy progress upon activity on
1035   // this policy, which in turn is tied to the application's call.
1036   grpc_pollset_set_add_pollset_set(lb_policy->interested_parties(),
1037                                    interested_parties());
1038   return lb_policy;
1039 }
1040 
1041 //
1042 // factory
1043 //
1044 
1045 const JsonLoaderInterface*
JsonLoader(const JsonArgs &)1046 XdsClusterResolverLbConfig::DiscoveryMechanism::JsonLoader(const JsonArgs&) {
1047   static const auto* loader =
1048       JsonObjectLoader<DiscoveryMechanism>()
1049           // Note: Several fields requires custom processing,
1050           // so they are handled in JsonPostLoad() instead.
1051           .Field("clusterName", &DiscoveryMechanism::cluster_name)
1052           .OptionalField("lrsLoadReportingServer",
1053                          &DiscoveryMechanism::lrs_load_reporting_server)
1054           .OptionalField("max_concurrent_requests",
1055                          &DiscoveryMechanism::max_concurrent_requests)
1056           .OptionalField("outlierDetection",
1057                          &DiscoveryMechanism::outlier_detection_lb_config)
1058           .OptionalField("overrideHostStatus",
1059                          &DiscoveryMechanism::override_host_statuses)
1060           .Finish();
1061   return loader;
1062 }
1063 
JsonPostLoad(const Json & json,const JsonArgs & args,ValidationErrors * errors)1064 void XdsClusterResolverLbConfig::DiscoveryMechanism::JsonPostLoad(
1065     const Json& json, const JsonArgs& args, ValidationErrors* errors) {
1066   // Parse "type".
1067   {
1068     auto type_field =
1069         LoadJsonObjectField<std::string>(json.object(), args, "type", errors);
1070     if (type_field.has_value()) {
1071       if (*type_field == "EDS") {
1072         type = DiscoveryMechanismType::EDS;
1073       } else if (*type_field == "LOGICAL_DNS") {
1074         type = DiscoveryMechanismType::LOGICAL_DNS;
1075       } else {
1076         ValidationErrors::ScopedField field(errors, ".type");
1077         errors->AddError(absl::StrCat("unknown type \"", *type_field, "\""));
1078       }
1079     }
1080   }
1081   // Parse "edsServiceName" if type is EDS.
1082   if (type == DiscoveryMechanismType::EDS) {
1083     auto value = LoadJsonObjectField<std::string>(json.object(), args,
1084                                                   "edsServiceName", errors,
1085                                                   /*required=*/false);
1086     if (value.has_value()) eds_service_name = std::move(*value);
1087   }
1088   // Parse "dnsHostname" if type is LOGICAL_DNS.
1089   if (type == DiscoveryMechanismType::LOGICAL_DNS) {
1090     auto value = LoadJsonObjectField<std::string>(json.object(), args,
1091                                                   "dnsHostname", errors);
1092     if (value.has_value()) dns_hostname = std::move(*value);
1093   }
1094 }
1095 
JsonLoader(const JsonArgs &)1096 const JsonLoaderInterface* XdsClusterResolverLbConfig::JsonLoader(
1097     const JsonArgs&) {
1098   static const auto* loader =
1099       JsonObjectLoader<XdsClusterResolverLbConfig>()
1100           // Note: The "xdsLbPolicy" field requires custom processing,
1101           // so it's handled in JsonPostLoad() instead.
1102           .Field("discoveryMechanisms",
1103                  &XdsClusterResolverLbConfig::discovery_mechanisms_)
1104           .Finish();
1105   return loader;
1106 }
1107 
JsonPostLoad(const Json & json,const JsonArgs &,ValidationErrors * errors)1108 void XdsClusterResolverLbConfig::JsonPostLoad(const Json& json, const JsonArgs&,
1109                                               ValidationErrors* errors) {
1110   // Validate discoveryMechanisms.
1111   {
1112     ValidationErrors::ScopedField field(errors, ".discoveryMechanisms");
1113     if (!errors->FieldHasErrors() && discovery_mechanisms_.empty()) {
1114       errors->AddError("must be non-empty");
1115     }
1116   }
1117   // Parse "xdsLbPolicy".
1118   {
1119     ValidationErrors::ScopedField field(errors, ".xdsLbPolicy");
1120     auto it = json.object().find("xdsLbPolicy");
1121     if (it == json.object().end()) {
1122       errors->AddError("field not present");
1123     } else {
1124       auto lb_config = CoreConfiguration::Get()
1125                            .lb_policy_registry()
1126                            .ParseLoadBalancingConfig(it->second);
1127       if (!lb_config.ok()) errors->AddError(lb_config.status().message());
1128       xds_lb_policy_ = it->second;
1129     }
1130   }
1131 }
1132 
1133 class XdsClusterResolverLbFactory : public LoadBalancingPolicyFactory {
1134  public:
CreateLoadBalancingPolicy(LoadBalancingPolicy::Args args) const1135   OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy(
1136       LoadBalancingPolicy::Args args) const override {
1137     auto xds_client = args.args.GetObjectRef<GrpcXdsClient>(
1138         DEBUG_LOCATION, "XdsClusterResolverLbFactory");
1139     if (xds_client == nullptr) {
1140       gpr_log(GPR_ERROR,
1141               "XdsClient not present in channel args -- cannot instantiate "
1142               "xds_cluster_resolver LB policy");
1143       return nullptr;
1144     }
1145     return MakeOrphanable<XdsClusterResolverChildHandler>(std::move(xds_client),
1146                                                           std::move(args));
1147   }
1148 
name() const1149   absl::string_view name() const override { return kXdsClusterResolver; }
1150 
1151   absl::StatusOr<RefCountedPtr<LoadBalancingPolicy::Config>>
ParseLoadBalancingConfig(const Json & json) const1152   ParseLoadBalancingConfig(const Json& json) const override {
1153     return LoadFromJson<RefCountedPtr<XdsClusterResolverLbConfig>>(
1154         json, JsonArgs(),
1155         "errors validating xds_cluster_resolver LB policy config");
1156   }
1157 
1158  private:
1159   class XdsClusterResolverChildHandler : public ChildPolicyHandler {
1160    public:
XdsClusterResolverChildHandler(RefCountedPtr<XdsClient> xds_client,Args args)1161     XdsClusterResolverChildHandler(RefCountedPtr<XdsClient> xds_client,
1162                                    Args args)
1163         : ChildPolicyHandler(std::move(args),
1164                              &grpc_lb_xds_cluster_resolver_trace),
1165           xds_client_(std::move(xds_client)) {}
1166 
~XdsClusterResolverChildHandler()1167     ~XdsClusterResolverChildHandler() override {
1168       xds_client_.reset(DEBUG_LOCATION, "XdsClusterResolverChildHandler");
1169     }
1170 
ConfigChangeRequiresNewPolicyInstance(LoadBalancingPolicy::Config * old_config,LoadBalancingPolicy::Config * new_config) const1171     bool ConfigChangeRequiresNewPolicyInstance(
1172         LoadBalancingPolicy::Config* old_config,
1173         LoadBalancingPolicy::Config* new_config) const override {
1174       GPR_ASSERT(old_config->name() == kXdsClusterResolver);
1175       GPR_ASSERT(new_config->name() == kXdsClusterResolver);
1176       XdsClusterResolverLbConfig* old_xds_cluster_resolver_config =
1177           static_cast<XdsClusterResolverLbConfig*>(old_config);
1178       XdsClusterResolverLbConfig* new_xds_cluster_resolver_config =
1179           static_cast<XdsClusterResolverLbConfig*>(new_config);
1180       if (old_xds_cluster_resolver_config->discovery_mechanisms().size() !=
1181           new_xds_cluster_resolver_config->discovery_mechanisms().size()) {
1182         return true;
1183       }
1184       for (size_t i = 0;
1185            i < old_xds_cluster_resolver_config->discovery_mechanisms().size();
1186            ++i) {
1187         auto& old_discovery_mechanism =
1188             old_xds_cluster_resolver_config->discovery_mechanisms()[i];
1189         auto& new_discovery_mechanism =
1190             new_xds_cluster_resolver_config->discovery_mechanisms()[i];
1191         if (old_discovery_mechanism.type != new_discovery_mechanism.type ||
1192             old_discovery_mechanism.cluster_name !=
1193                 new_discovery_mechanism.cluster_name ||
1194             old_discovery_mechanism.eds_service_name !=
1195                 new_discovery_mechanism.eds_service_name ||
1196             old_discovery_mechanism.dns_hostname !=
1197                 new_discovery_mechanism.dns_hostname ||
1198             !(old_discovery_mechanism.lrs_load_reporting_server ==
1199               new_discovery_mechanism.lrs_load_reporting_server)) {
1200           return true;
1201         }
1202       }
1203       return false;
1204     }
1205 
CreateLoadBalancingPolicy(absl::string_view,LoadBalancingPolicy::Args args) const1206     OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy(
1207         absl::string_view /*name*/,
1208         LoadBalancingPolicy::Args args) const override {
1209       return MakeOrphanable<XdsClusterResolverLb>(
1210           xds_client_->Ref(DEBUG_LOCATION, "XdsClusterResolverLb"),
1211           std::move(args));
1212     }
1213 
1214    private:
1215     RefCountedPtr<XdsClient> xds_client_;
1216   };
1217 };
1218 
1219 }  // namespace
1220 
RegisterXdsClusterResolverLbPolicy(CoreConfiguration::Builder * builder)1221 void RegisterXdsClusterResolverLbPolicy(CoreConfiguration::Builder* builder) {
1222   builder->lb_policy_registry()->RegisterLoadBalancingPolicyFactory(
1223       std::make_unique<XdsClusterResolverLbFactory>());
1224 }
1225 
1226 }  // namespace grpc_core
1227