// // Copyright 2019 gRPC authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // #ifndef GRPC_CORE_EXT_XDS_XDS_CLIENT_H #define GRPC_CORE_EXT_XDS_XDS_CLIENT_H #include #include #include "absl/strings/string_view.h" #include "absl/types/optional.h" #include "src/core/ext/xds/xds_api.h" #include "src/core/ext/xds/xds_bootstrap.h" #include "src/core/ext/xds/xds_client_stats.h" #include "src/core/lib/gprpp/map.h" #include "src/core/lib/gprpp/memory.h" #include "src/core/lib/gprpp/orphanable.h" #include "src/core/lib/gprpp/ref_counted.h" #include "src/core/lib/gprpp/ref_counted_ptr.h" #include "src/core/lib/iomgr/work_serializer.h" namespace grpc_core { extern TraceFlag xds_client_trace; class XdsClient : public InternallyRefCounted { public: // Listener data watcher interface. Implemented by callers. class ListenerWatcherInterface { public: virtual ~ListenerWatcherInterface() = default; virtual void OnListenerChanged(std::vector routes) = 0; virtual void OnError(grpc_error* error) = 0; virtual void OnResourceDoesNotExist() = 0; }; // Cluster data watcher interface. Implemented by callers. class ClusterWatcherInterface { public: virtual ~ClusterWatcherInterface() = default; virtual void OnClusterChanged(XdsApi::CdsUpdate cluster_data) = 0; virtual void OnError(grpc_error* error) = 0; virtual void OnResourceDoesNotExist() = 0; }; // Endpoint data watcher interface. Implemented by callers. class EndpointWatcherInterface { public: virtual ~EndpointWatcherInterface() = default; virtual void OnEndpointChanged(XdsApi::EdsUpdate update) = 0; virtual void OnError(grpc_error* error) = 0; virtual void OnResourceDoesNotExist() = 0; }; // If *error is not GRPC_ERROR_NONE after construction, then there was // an error initializing the client. XdsClient(std::shared_ptr work_serializer, grpc_pollset_set* interested_parties, absl::string_view server_name, std::unique_ptr watcher, const grpc_channel_args& channel_args, grpc_error** error); ~XdsClient(); void Orphan() override; // Start and cancel cluster data watch for a cluster. // The XdsClient takes ownership of the watcher, but the caller may // keep a raw pointer to the watcher, which may be used only for // cancellation. (Because the caller does not own the watcher, the // pointer must not be used for any other purpose.) // If the caller is going to start a new watch after cancelling the // old one, it should set delay_unsubscription to true. void WatchClusterData(absl::string_view cluster_name, std::unique_ptr watcher); void CancelClusterDataWatch(absl::string_view cluster_name, ClusterWatcherInterface* watcher, bool delay_unsubscription = false); // Start and cancel endpoint data watch for a cluster. // The XdsClient takes ownership of the watcher, but the caller may // keep a raw pointer to the watcher, which may be used only for // cancellation. (Because the caller does not own the watcher, the // pointer must not be used for any other purpose.) // If the caller is going to start a new watch after cancelling the // old one, it should set delay_unsubscription to true. void WatchEndpointData(absl::string_view eds_service_name, std::unique_ptr watcher); void CancelEndpointDataWatch(absl::string_view eds_service_name, EndpointWatcherInterface* watcher, bool delay_unsubscription = false); // Adds and removes drop stats for cluster_name and eds_service_name. RefCountedPtr AddClusterDropStats( absl::string_view lrs_server, absl::string_view cluster_name, absl::string_view eds_service_name); void RemoveClusterDropStats(absl::string_view /*lrs_server*/, absl::string_view cluster_name, absl::string_view eds_service_name, XdsClusterDropStats* cluster_drop_stats); // Adds and removes locality stats for cluster_name and eds_service_name // for the specified locality. RefCountedPtr AddClusterLocalityStats( absl::string_view lrs_server, absl::string_view cluster_name, absl::string_view eds_service_name, RefCountedPtr locality); void RemoveClusterLocalityStats( absl::string_view /*lrs_server*/, absl::string_view cluster_name, absl::string_view eds_service_name, const RefCountedPtr& locality, XdsClusterLocalityStats* cluster_locality_stats); // Resets connection backoff state. void ResetBackoff(); // Helpers for encoding the XdsClient object in channel args. grpc_arg MakeChannelArg() const; static RefCountedPtr GetFromChannelArgs( const grpc_channel_args& args); static grpc_channel_args* RemoveFromChannelArgs( const grpc_channel_args& args); private: // Contains a channel to the xds server and all the data related to the // channel. Holds a ref to the xds client object. // TODO(roth): This is separate from the XdsClient object because it was // originally designed to be able to swap itself out in case the // balancer name changed. Now that the balancer name is going to be // coming from the bootstrap file, we don't really need this level of // indirection unless we decide to support watching the bootstrap file // for changes. At some point, if we decide that we're never going to // need to do that, then we can eliminate this class and move its // contents directly into the XdsClient class. class ChannelState : public InternallyRefCounted { public: template class RetryableCall; class AdsCallState; class LrsCallState; ChannelState(RefCountedPtr xds_client, grpc_channel* channel); ~ChannelState(); void Orphan() override; grpc_channel* channel() const { return channel_; } XdsClient* xds_client() const { return xds_client_.get(); } AdsCallState* ads_calld() const; LrsCallState* lrs_calld() const; void MaybeStartLrsCall(); void StopLrsCall(); bool HasActiveAdsCall() const; void StartConnectivityWatchLocked(); void CancelConnectivityWatchLocked(); void Subscribe(const std::string& type_url, const std::string& name); void Unsubscribe(const std::string& type_url, const std::string& name, bool delay_unsubscription); private: class StateWatcher; // The owning xds client. RefCountedPtr xds_client_; // The channel and its status. grpc_channel* channel_; bool shutting_down_ = false; StateWatcher* watcher_ = nullptr; // The retryable XDS calls. OrphanablePtr> ads_calld_; OrphanablePtr> lrs_calld_; }; struct ClusterState { std::map> watchers; // The latest data seen from CDS. absl::optional update; }; struct EndpointState { std::map> watchers; // The latest data seen from EDS. absl::optional update; }; struct LoadReportState { struct LocalityState { std::set locality_stats; std::vector deleted_locality_stats; }; std::set drop_stats; XdsClusterDropStats::DroppedRequestsMap deleted_drop_stats; std::map, LocalityState, XdsLocalityName::Less> locality_stats; grpc_millis last_report_time = ExecCtx::Get()->Now(); }; // Sends an error notification to all watchers. void NotifyOnError(grpc_error* error); XdsApi::ClusterLoadReportMap BuildLoadReportSnapshot( bool send_all_clusters, const std::set& clusters); // Channel arg vtable functions. static void* ChannelArgCopy(void* p); static void ChannelArgDestroy(void* p); static int ChannelArgCmp(void* p, void* q); static const grpc_arg_pointer_vtable kXdsClientVtable; const grpc_millis request_timeout_; std::shared_ptr work_serializer_; grpc_pollset_set* interested_parties_; std::unique_ptr bootstrap_; XdsApi api_; const std::string server_name_; std::unique_ptr listener_watcher_; // The channel for communicating with the xds server. OrphanablePtr chand_; absl::optional lds_result_; absl::optional rds_result_; // One entry for each watched CDS resource. std::map cluster_map_; // One entry for each watched EDS resource. std::map endpoint_map_; std::map< std::pair, LoadReportState> load_report_map_; bool shutting_down_ = false; }; } // namespace grpc_core #endif /* GRPC_CORE_EXT_XDS_XDS_CLIENT_H */